Skip to content

Commit f1dc007

Browse files
gfyragGeoffrey Ragot
gfyrag
and
Geoffrey Ragot
authored
fix(reads): preserve routed consistency guarantees (#1876)
Co-authored-by: Geoffrey Ragot <geoffrey@formance.com>
1 parent 56eaf4e commit f1dc007

20 files changed

Lines changed: 449 additions & 107 deletions

‎internal/adapter/grpc/client_bucket.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -190,7 +190,7 @@ func (g *BucketGrpcClient) GetLedgerByName(ctx context.Context, name string) (*c
190190
})
191191
}
192192

193-
func (g *BucketGrpcClient) ListAuditEntries(ctx context.Context, pageSize uint32, afterSequence uint64, filter *commonpb.QueryFilter, reverse bool) (cursor.Cursor[*auditpb.AuditEntry], error) {
193+
func (g *BucketGrpcClient) ListAuditEntries(ctx context.Context, pageSize uint32, afterSequence uint64, filter *commonpb.QueryFilter, reverse bool, minLogSequence uint64) (cursor.Cursor[*auditpb.AuditEntry], error) {
194194
var cursorStr string
195195
if afterSequence > 0 {
196196
cursorStr = strconv.FormatUint(afterSequence, 10)
@@ -202,6 +202,7 @@ func (g *BucketGrpcClient) ListAuditEntries(ctx context.Context, pageSize uint32
202202
Cursor: cursorStr,
203203
Reverse: reverse,
204204
Filter: filter,
205+
Read: &commonpb.ReadOptions{MinLogSequence: minLogSequence},
205206
},
206207
})
207208
if err != nil {

‎internal/adapter/grpc/client_bucket_test.go‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -385,10 +385,16 @@ func TestListAuditEntries_Success(t *testing.T) {
385385
stream := newRecvStream[auditpb.AuditEntry](ctrl, []*auditpb.AuditEntry{
386386
{Sequence: 1},
387387
}, nil)
388-
mock.EXPECT().ListAuditEntries(gomock.Any(), gomock.Any()).Return(stream, nil)
388+
mock.EXPECT().ListAuditEntries(gomock.Any(), gomock.Any()).DoAndReturn(
389+
func(_ context.Context, req *servicepb.ListAuditEntriesRequest, _ ...grpc.CallOption) (servicepb.BucketService_ListAuditEntriesClient, error) {
390+
require.Equal(t, uint64(7), req.GetOptions().GetRead().GetMinLogSequence())
391+
392+
return stream, nil
393+
},
394+
)
389395

390396
client := NewLedgerGrpcClient(mock)
391-
cursor, err := client.ListAuditEntries(context.Background(), 10, 5, nil, false)
397+
cursor, err := client.ListAuditEntries(context.Background(), 10, 5, nil, false, 7)
392398
require.NoError(t, err)
393399

394400
entry, err := cursor.Next()
@@ -404,7 +410,7 @@ func TestListAuditEntries_StreamError(t *testing.T) {
404410
mock.EXPECT().ListAuditEntries(gomock.Any(), gomock.Any()).Return(nil, errors.New("audit error"))
405411

406412
client := NewLedgerGrpcClient(mock)
407-
_, err := client.ListAuditEntries(context.Background(), 10, 0, nil, false)
413+
_, err := client.ListAuditEntries(context.Background(), 10, 0, nil, false, 0)
408414
require.Error(t, err)
409415
}
410416

‎internal/adapter/grpc/consistency.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,9 @@ const (
1515
// ConsistencyStale skips the ReadIndex barrier and reads from the local store directly.
1616
// Data may lag behind the latest committed index.
1717
ConsistencyStale = "stale"
18-
// ConsistencyLeader forwards the read to the leader node, which always has
19-
// the most up-to-date data and a fast ReadIndex barrier.
18+
// ConsistencyLeader routes the read to the node currently considered leader.
19+
// A remote leader applies the default ReadIndex barrier, but a node that
20+
// already considers itself leader serves the read locally without one.
2021
ConsistencyLeader = "leader"
2122
)
2223

‎internal/adapter/grpc/controller_generated_test.go‎

Lines changed: 6 additions & 6 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎internal/adapter/grpc/server_bucket.go‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1045,21 +1045,19 @@ func (impl *BucketServiceServerImpl) ListAuditEntries(req *servicepb.ListAuditEn
10451045
} else {
10461046
minLogSeq := opts.GetRead().GetMinLogSequence()
10471047

1048-
// A filtered live read resolves through the async audit secondary index,
1049-
// which lags the audit zone independently of the log index. Gate it on the
1050-
// audit-index progress so the requested consistency bound actually covers
1051-
// the index this read consults. An unfiltered read scans the Cold/Audit
1052-
// zone directly and is always current, so it keeps the plain log-index wait
1053-
// (its fast path is unchanged).
1054-
if opts.GetFilter() != nil {
1048+
// Filters that consult the async audit secondary index need its independent
1049+
// progress gate. Nil filters and conjunctions made only of seq bounds scan
1050+
// the Cold/Audit zone directly, so coupling them to audit-index progress
1051+
// would make an authoritative read wait for a projection it never uses.
1052+
if query.AuditFilterNeedsIndex(opts.GetFilter()) {
10551053
if waitErr := impl.waitFilteredAuditConsistency(ctx, minLogSeq); waitErr != nil {
10561054
return waitErr
10571055
}
10581056
} else if waitErr := impl.waitMinLogSequence(ctx, minLogSeq); waitErr != nil {
10591057
return waitErr
10601058
}
10611059

1062-
c, err = impl.ctrl.ListAuditEntries(ctx, fetchSize, afterSeq, opts.GetFilter(), opts.GetReverse())
1060+
c, err = impl.ctrl.ListAuditEntries(ctx, fetchSize, afterSeq, opts.GetFilter(), opts.GetReverse(), minLogSeq)
10631061
}
10641062

10651063
if err != nil {

‎internal/adapter/grpc/server_bucket_audit_consistency_test.go‎

Lines changed: 133 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -112,12 +112,32 @@ func newAuditConsistencyHarness(t *testing.T) (*BucketServiceServerImpl, *ctrlmo
112112
return impl, mockCtrl, mainStore, rs
113113
}
114114

115-
// singleFieldFilter builds a minimal non-nil QueryFilter so the handler takes
116-
// the filtered branch. Its content is irrelevant here: the controller is mocked,
117-
// so the filter is never actually compiled — only its presence matters.
118-
func singleFieldFilter() *commonpb.QueryFilter {
115+
func indexedAuditFilter() *commonpb.QueryFilter {
119116
return &commonpb.QueryFilter{
120-
Filter: &commonpb.QueryFilter_Field{Field: &commonpb.FieldCondition{}},
117+
Filter: &commonpb.QueryFilter_Audit{Audit: &commonpb.AuditCondition{
118+
Field: commonpb.AuditField_AUDIT_FIELD_OUTCOME,
119+
Condition: &commonpb.AuditCondition_StringCond{StringCond: &commonpb.StringCondition{
120+
Value: &commonpb.StringCondition_Hardcoded{Hardcoded: "failure"},
121+
}},
122+
}},
123+
}
124+
}
125+
126+
func auditSequenceFilter(lower, upper *uint64) *commonpb.QueryFilter {
127+
return &commonpb.QueryFilter{
128+
Filter: &commonpb.QueryFilter_Audit{Audit: &commonpb.AuditCondition{
129+
Field: commonpb.AuditField_AUDIT_FIELD_SEQUENCE,
130+
Condition: &commonpb.AuditCondition_UintCond{UintCond: &commonpb.UintCondition{
131+
Min: lower,
132+
Max: upper,
133+
}},
134+
}},
135+
}
136+
}
137+
138+
func auditAnd(filters ...*commonpb.QueryFilter) *commonpb.QueryFilter {
139+
return &commonpb.QueryFilter{
140+
Filter: &commonpb.QueryFilter_And{And: &commonpb.AndFilter{Filters: filters}},
121141
}
122142
}
123143

@@ -139,8 +159,8 @@ func TestListAuditEntriesFilteredWaitsForAuditProgress(t *testing.T) {
139159
// The controller must only be called AFTER the audit index catches up.
140160
controllerCalled := make(chan struct{})
141161
mockCtrl.EXPECT().
142-
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Not(gomock.Nil()), gomock.Any()).
143-
DoAndReturn(func(_ context.Context, _ uint32, _ uint64, _ *commonpb.QueryFilter, _ bool) (cursor.Cursor[*auditpb.AuditEntry], error) {
162+
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Not(gomock.Nil()), gomock.Any(), uint64(5)).
163+
DoAndReturn(func(_ context.Context, _ uint32, _ uint64, _ *commonpb.QueryFilter, _ bool, _ uint64) (cursor.Cursor[*auditpb.AuditEntry], error) {
144164
close(controllerCalled)
145165

146166
return cursor.NewSliceCursor([]*auditpb.AuditEntry(nil)), nil
@@ -154,7 +174,7 @@ func TestListAuditEntriesFilteredWaitsForAuditProgress(t *testing.T) {
154174
stream := newAuditStream(t, ctx)
155175
req := &servicepb.ListAuditEntriesRequest{Options: &commonpb.ListOptions{
156176
PageSize: 2,
157-
Filter: singleFieldFilter(),
177+
Filter: indexedAuditFilter(),
158178
Read: &commonpb.ReadOptions{MinLogSequence: 5},
159179
}}
160180
handlerErr <- impl.ListAuditEntries(req, stream)
@@ -205,7 +225,7 @@ func TestListAuditEntriesFilteredDoesNotWaitOnLogSequenceInAuditSpace(t *testing
205225
writeReadstoreAuditProgress(t, rs, 2)
206226

207227
mockCtrl.EXPECT().
208-
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Not(gomock.Nil()), gomock.Any()).
228+
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Not(gomock.Nil()), gomock.Any(), uint64(10)).
209229
Return(cursor.NewSliceCursor([]*auditpb.AuditEntry(nil)), nil)
210230

211231
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
@@ -214,7 +234,7 @@ func TestListAuditEntriesFilteredDoesNotWaitOnLogSequenceInAuditSpace(t *testing
214234
stream := newAuditStream(t, ctx)
215235
req := &servicepb.ListAuditEntriesRequest{Options: &commonpb.ListOptions{
216236
PageSize: 2,
217-
Filter: singleFieldFilter(),
237+
Filter: indexedAuditFilter(),
218238
Read: &commonpb.ReadOptions{MinLogSequence: 10},
219239
}}
220240

@@ -237,7 +257,7 @@ func TestListAuditEntriesUnfilteredDoesNotWaitOnAuditProgress(t *testing.T) {
237257
// Audit index deliberately left at 0 — an unfiltered read must not care.
238258

239259
mockCtrl.EXPECT().
240-
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Nil(), gomock.Any()).
260+
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Nil(), gomock.Any(), uint64(5)).
241261
Return(cursor.NewSliceCursor([]*auditpb.AuditEntry(nil)), nil)
242262

243263
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
@@ -253,6 +273,106 @@ func TestListAuditEntriesUnfilteredDoesNotWaitOnAuditProgress(t *testing.T) {
253273
require.NoError(t, impl.ListAuditEntries(req, stream))
254274
}
255275

276+
func TestListAuditEntriesSequenceBoundsDoNotWaitOnAuditProgress(t *testing.T) {
277+
t.Parallel()
278+
279+
tests := []struct {
280+
name string
281+
filter *commonpb.QueryFilter
282+
}{
283+
{
284+
name: "single sequence bound",
285+
filter: auditSequenceFilter(new(uint64(3)), nil),
286+
},
287+
{
288+
name: "and-combined sequence bounds",
289+
filter: auditAnd(
290+
auditSequenceFilter(new(uint64(3)), nil),
291+
auditSequenceFilter(nil, new(uint64(12))),
292+
),
293+
},
294+
}
295+
296+
for _, tt := range tests {
297+
t.Run(tt.name, func(t *testing.T) {
298+
t.Parallel()
299+
300+
impl, mockCtrl, mainStore, rs := newAuditConsistencyHarness(t)
301+
writeReadstoreLogProgress(t, rs, 5)
302+
writeMainAuditEntry(t, mainStore, 9)
303+
// The query scans the audit zone directly, so a stalled audit index
304+
// must not prevent the controller call.
305+
writeReadstoreAuditProgress(t, rs, 1)
306+
307+
mockCtrl.EXPECT().
308+
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), tt.filter, gomock.Any(), uint64(5)).
309+
Return(cursor.NewSliceCursor([]*auditpb.AuditEntry(nil)), nil)
310+
311+
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
312+
defer cancel()
313+
314+
req := &servicepb.ListAuditEntriesRequest{Options: &commonpb.ListOptions{
315+
PageSize: 2,
316+
Filter: tt.filter,
317+
Read: &commonpb.ReadOptions{MinLogSequence: 5},
318+
}}
319+
320+
require.NoError(t, impl.ListAuditEntries(req, newAuditStream(t, ctx)))
321+
})
322+
}
323+
}
324+
325+
func TestListAuditEntriesSequenceAndIndexedFieldWaitsForAuditProgress(t *testing.T) {
326+
t.Parallel()
327+
328+
impl, mockCtrl, mainStore, rs := newAuditConsistencyHarness(t)
329+
writeReadstoreLogProgress(t, rs, 5)
330+
writeMainAuditEntry(t, mainStore, 9)
331+
writeReadstoreAuditProgress(t, rs, 3)
332+
333+
filter := auditAnd(
334+
auditSequenceFilter(new(uint64(3)), nil),
335+
indexedAuditFilter(),
336+
)
337+
controllerCalled := make(chan struct{})
338+
mockCtrl.EXPECT().
339+
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), filter, gomock.Any(), uint64(5)).
340+
DoAndReturn(func(_ context.Context, _ uint32, _ uint64, _ *commonpb.QueryFilter, _ bool, _ uint64) (cursor.Cursor[*auditpb.AuditEntry], error) {
341+
close(controllerCalled)
342+
343+
return cursor.NewSliceCursor([]*auditpb.AuditEntry(nil)), nil
344+
})
345+
346+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
347+
defer cancel()
348+
349+
handlerErr := make(chan error, 1)
350+
go func() {
351+
req := &servicepb.ListAuditEntriesRequest{Options: &commonpb.ListOptions{
352+
PageSize: 2,
353+
Filter: filter,
354+
Read: &commonpb.ReadOptions{MinLogSequence: 5},
355+
}}
356+
handlerErr <- impl.ListAuditEntries(req, newAuditStream(t, ctx))
357+
}()
358+
359+
select {
360+
case <-controllerCalled:
361+
t.Fatal("controller called before audit index caught up for a mixed filter")
362+
case <-time.After(100 * time.Millisecond):
363+
}
364+
365+
writeReadstoreAuditProgress(t, rs, 9)
366+
rs.NotifyProgress()
367+
368+
select {
369+
case err := <-handlerErr:
370+
require.NoError(t, err)
371+
case <-time.After(5 * time.Second):
372+
t.Fatal("handler did not return after audit index caught up")
373+
}
374+
}
375+
256376
// TestListAuditEntriesFilteredContextCancelWhileWaiting verifies a filtered read
257377
// blocked on audit progress returns promptly with a context error when the
258378
// request context is cancelled, rather than hanging.
@@ -267,7 +387,7 @@ func TestListAuditEntriesFilteredContextCancelWhileWaiting(t *testing.T) {
267387

268388
// Controller must never be reached.
269389
mockCtrl.EXPECT().
270-
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
390+
ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
271391
Times(0)
272392

273393
ctx, cancel := context.WithCancel(context.Background())
@@ -277,7 +397,7 @@ func TestListAuditEntriesFilteredContextCancelWhileWaiting(t *testing.T) {
277397
stream := newAuditStream(t, ctx)
278398
req := &servicepb.ListAuditEntriesRequest{Options: &commonpb.ListOptions{
279399
PageSize: 2,
280-
Filter: singleFieldFilter(),
400+
Filter: indexedAuditFilter(),
281401
Read: &commonpb.ReadOptions{MinLogSequence: 5},
282402
}}
283403
handlerErr <- impl.ListAuditEntries(req, stream)

‎internal/adapter/grpc/server_bucket_list_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -268,8 +268,8 @@ func TestListAuditEntries(t *testing.T) {
268268
t.Parallel()
269269

270270
impl, mockCtrl := newListHandlerHarness(t)
271-
// ListAuditEntries(ctx, pageSize, afterSequence, filter, reverse)
272-
mockCtrl.EXPECT().ListAuditEntries(gomock.Any(), uint32(3), uint64(0), nil, false).
271+
// ListAuditEntries(ctx, pageSize, afterSequence, filter, reverse, minLogSequence)
272+
mockCtrl.EXPECT().ListAuditEntries(gomock.Any(), uint32(3), uint64(0), nil, false, uint64(0)).
273273
Return(page(
274274
&auditpb.AuditEntry{Sequence: 1},
275275
&auditpb.AuditEntry{Sequence: 2},

‎internal/adapter/http/backend_generated_test.go‎

Lines changed: 6 additions & 6 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎internal/adapter/http/handlers_get_audit_entry_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -111,8 +111,8 @@ func TestAuditRoutes_FullRouteIntegration(t *testing.T) {
111111
t.Parallel()
112112

113113
backend := NewMockBackend(gomock.NewController(t))
114-
backend.EXPECT().ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
115-
func(_ context.Context, _ uint32, _ uint64, _ *commonpb.QueryFilter, _ bool) (cursor.Cursor[*auditpb.AuditEntry], error) {
114+
backend.EXPECT().ListAuditEntries(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), uint64(0)).DoAndReturn(
115+
func(_ context.Context, _ uint32, _ uint64, _ *commonpb.QueryFilter, _ bool, _ uint64) (cursor.Cursor[*auditpb.AuditEntry], error) {
116116
return cursor.NewSliceCursor([]*auditpb.AuditEntry{{Sequence: 1}}), nil
117117
}).AnyTimes()
118118
backend.EXPECT().GetAuditEntry(gomock.Any(), uint64(1)).DoAndReturn(

0 commit comments

Comments
 (0)