From f3f14632e4c531c05aed726533aea69c990a158d Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Mon, 21 Sep 2026 13:51:03 +0000 Subject: [PATCH 1/5] Improve antispam follower efficiency --- storage/aws/antispam/aws.go | 40 +++++++++++++++++++++++--------- storage/gcp/antispam/gcp.go | 36 +++++++++++++++++++++------- storage/posix/antispam/badger.go | 33 ++++++++++++++++++++------ 3 files changed, 83 insertions(+), 26 deletions(-) diff --git a/storage/aws/antispam/aws.go b/storage/aws/antispam/aws.go index 4d10eeb19..e73343d7a 100644 --- a/storage/aws/antispam/aws.go +++ b/storage/aws/antispam/aws.go @@ -53,6 +53,8 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second + // defaultStreamTimeout is the maximum duration to spend streaming entries from the log. + defaultStreamTimeout = time.Minute ) // AntispamOpts allows configuration of some tunable options. @@ -259,7 +261,7 @@ func (f *follower) Name() string { } // Follow uses entry data from the log to populate the antispam storage. -func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { +func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { errOutOfSync := errors.New("out-of-sync") t := time.NewTicker(time.Second) @@ -267,9 +269,15 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { next func() (client.Entry[[]byte], error, bool) stop func() ) + // Ensure we tear down any in-flight entry stream when we're done. + defer func() { + if stop != nil { + stop() + } + }() for { select { - case <-ctx.Done(): + case <-followCtx.Done(): return case <-t.C: } @@ -281,11 +289,11 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { // Busy loop while there's work to be done for streamDone := false; !streamDone; { select { - case <-ctx.Done(): + case <-followCtx.Done(): return default: } - err := otel.TraceErr(ctx, "tessera.antispam.aws.FollowTask", tracer, func(ctx context.Context, span trace.Span) error { + err := otel.TraceErr(followCtx, "tessera.antispam.aws.FollowTask", tracer, func(ctx context.Context, span trace.Span) error { ctx, cancel := context.WithTimeout(ctx, defaultBatchTimeout) defer cancel() @@ -335,11 +343,22 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { // If this is the first time around the loop we need to start the stream of entries now that we know where we want to // start reading from: if next == nil { + streamSize := logSize sizeFn := func(_ context.Context) (uint64, error) { - return logSize, nil + return streamSize, nil } numFetchers := uint(10) - next, stop = iter.Pull2(client.Entries(client.EntryBundles(ctx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + + // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. + // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). + streamCtx, sCancel := context.WithTimeout(trace.ContextWithSpan(followCtx, span), defaultStreamTimeout) + + streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + + next, stop = streamNext, func() { + sCancel() + streamStop() + } } bs := uint64(f.as.opts.MaxBatchSize) @@ -352,7 +371,7 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { if !ok { // The entry stream has ended so we'll need to start a new stream next time around the loop: stop() - next = nil + next, stop = nil, nil break } if err != nil { @@ -400,13 +419,12 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { }) if err != nil { if err != errOutOfSync { - slog.ErrorContext(ctx, "Failed to commit antispam population tx", slog.Any("error", err)) + slog.ErrorContext(followCtx, "Failed to commit antispam population tx", slog.Any("error", err)) } - if next != nil { + if stop != nil { stop() - next = nil - stop = nil } + next, stop = nil, nil streamDone = true continue } diff --git a/storage/gcp/antispam/gcp.go b/storage/gcp/antispam/gcp.go index 207561fa2..60dd05489 100644 --- a/storage/gcp/antispam/gcp.go +++ b/storage/gcp/antispam/gcp.go @@ -52,6 +52,8 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second + // defaultStreamTimeout is the maximum duration to spend streaming entries from the log. + defaultStreamTimeout = time.Minute ) // AntispamOpts allows configuration of some tunable options. @@ -279,7 +281,7 @@ func (f *follower) Name() string { } // Follow uses entry data from the log to populate the antispam storage. -func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { +func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { errOutOfSync := errors.New("out-of-sync") t := time.NewTicker(time.Second) @@ -290,9 +292,15 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { curEntries [][]byte curIndex uint64 ) + // Ensure we tear down any in-flight entry stream when we're done. + defer func() { + if stop != nil { + stop() + } + }() for { select { - case <-ctx.Done(): + case <-followCtx.Done(): return case <-t.C: } @@ -303,7 +311,7 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { // Busy loop while there are entries to be consumed from the stream for streamDone := false; !streamDone; { - err := otel.TraceErr(ctx, "tessera.antispam.gcp.FollowTask", tracer, func(ctx context.Context, span trace.Span) error { + err := otel.TraceErr(followCtx, "tessera.antispam.gcp.FollowTask", tracer, func(ctx context.Context, span trace.Span) error { ctx, cancel := context.WithTimeout(ctx, defaultBatchTimeout) defer cancel() _, err := f.as.dbPool.ReadWriteTransactionWithOptions(ctx, func(txctx context.Context, txn *spanner.ReadWriteTransaction) error { @@ -352,11 +360,23 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { // start reading from: if next == nil { span.AddEvent("Start streaming entries") + streamSize := logSize sizeFn := func(_ context.Context) (uint64, error) { - return logSize, nil + return streamSize, nil } + numFetchers := uint(10) - next, stop = iter.Pull2(client.Entries(client.EntryBundles(txctx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + + // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. + // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). + streamCtx, sCancel := context.WithTimeout(trace.ContextWithSpan(followCtx, span), defaultStreamTimeout) + + streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + + next, stop = streamNext, func() { + sCancel() + streamStop() + } } if curIndex == followFrom && curEntries != nil { @@ -375,7 +395,7 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { if !ok { // The entry stream has ended so we'll need to start a new stream next time around the loop: stop() - next = nil + next, stop = nil, nil break } if err != nil { @@ -420,12 +440,12 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { }) if err != nil { if err != errOutOfSync { - slog.ErrorContext(ctx, "Failed to commit antispam population tx", slog.Any("error", err)) + slog.ErrorContext(followCtx, "Failed to commit antispam population tx", slog.Any("error", err)) } if stop != nil { stop() } - next = nil + next, stop = nil, nil streamDone = true continue } diff --git a/storage/posix/antispam/badger.go b/storage/posix/antispam/badger.go index 135ebf839..102aadd71 100644 --- a/storage/posix/antispam/badger.go +++ b/storage/posix/antispam/badger.go @@ -45,6 +45,8 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second + // defaultStreamTimeout is the maximum duration to spend streaming entries from the log. + defaultStreamTimeout = time.Minute ) var ( @@ -256,7 +258,7 @@ func (f *follower) Name() string { } // Follow uses entry data from the log to populate the antispam storage. -func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { +func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { errOutOfSync := errors.New("out-of-sync") t := time.NewTicker(time.Second) @@ -267,9 +269,15 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { curEntries [][]byte curIndex uint64 ) + // Ensure we tear down any in-flight entry stream when we're done. + defer func() { + if stop != nil { + stop() + } + }() for { select { - case <-ctx.Done(): + case <-followCtx.Done(): return case <-t.C: } @@ -282,7 +290,6 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { for moreWork := true; moreWork; { err := f.as.db.Update(func(txn *badger.Txn) error { - return otel.TraceErr(ctx, "tessera.antispam.badger.follow_txn", tracer, func(ctx context.Context, span trace.Span) error { batchStart := time.Now() ctx, cancel := context.WithTimeout(ctx, defaultBatchTimeout) defer cancel() @@ -346,7 +353,17 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { return logSize, nil } numFetchers := uint(10) - next, stop = iter.Pull2(client.Entries(client.EntryBundles(ctx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + + // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. + // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). + streamCtx, sCancel := context.WithTimeout(trace.ContextWithSpan(followCtx, span), defaultStreamTimeout) + + streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + + next, stop = streamNext, func() { + sCancel() + streamStop() + } } if curIndex == followFrom && curEntries != nil { @@ -364,7 +381,8 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { e, err, ok := next() if !ok { // The entry stream has ended so we'll need to start a new stream next time around the loop: - next = nil + stop() + next, stop = nil, nil break } if err != nil { @@ -412,12 +430,13 @@ func (f *follower) Follow(ctx context.Context, lr tessera.LogReader) { }) if err != nil { if err != errOutOfSync { - slog.ErrorContext(ctx, "Failed to commit antispam population tx", slog.Any("error", err)) + slog.ErrorContext(followCtx, "Failed to commit antispam population tx", slog.Any("error", err)) } if stop != nil { stop() } - next = nil + next, stop = nil, nil + streamDone = true continue } curEntries = nil From 6591fc48e5175f8e515cc9bb245edb2e2c4f96e0 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Tue, 22 Sep 2026 09:47:20 +0000 Subject: [PATCH 2/5] Stream cancel not timeout --- storage/aws/antispam/aws.go | 6 ++--- storage/gcp/antispam/gcp.go | 6 ++--- storage/posix/antispam/badger.go | 44 ++++++++++++++++++++++---------- 3 files changed, 35 insertions(+), 21 deletions(-) diff --git a/storage/aws/antispam/aws.go b/storage/aws/antispam/aws.go index e73343d7a..0a35a3d35 100644 --- a/storage/aws/antispam/aws.go +++ b/storage/aws/antispam/aws.go @@ -53,8 +53,6 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second - // defaultStreamTimeout is the maximum duration to spend streaming entries from the log. - defaultStreamTimeout = time.Minute ) // AntispamOpts allows configuration of some tunable options. @@ -351,11 +349,11 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). - streamCtx, sCancel := context.WithTimeout(trace.ContextWithSpan(followCtx, span), defaultStreamTimeout) - + streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) next, stop = streamNext, func() { + // Cancel first so in-flight fetches abort, then kill the iterator sCancel() streamStop() } diff --git a/storage/gcp/antispam/gcp.go b/storage/gcp/antispam/gcp.go index 60dd05489..2490ab5d5 100644 --- a/storage/gcp/antispam/gcp.go +++ b/storage/gcp/antispam/gcp.go @@ -52,8 +52,6 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second - // defaultStreamTimeout is the maximum duration to spend streaming entries from the log. - defaultStreamTimeout = time.Minute ) // AntispamOpts allows configuration of some tunable options. @@ -369,11 +367,11 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). - streamCtx, sCancel := context.WithTimeout(trace.ContextWithSpan(followCtx, span), defaultStreamTimeout) - + streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) next, stop = streamNext, func() { + // Cancel first so in-flight fetches abort, then kill the iterator sCancel() streamStop() } diff --git a/storage/posix/antispam/badger.go b/storage/posix/antispam/badger.go index 102aadd71..cb965d03d 100644 --- a/storage/posix/antispam/badger.go +++ b/storage/posix/antispam/badger.go @@ -45,8 +45,6 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second - // defaultStreamTimeout is the maximum duration to spend streaming entries from the log. - defaultStreamTimeout = time.Minute ) var ( @@ -287,9 +285,15 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { var logSize uint64 // Busy loop while there's work to be done - for moreWork := true; moreWork; { + for streamDone := false; !streamDone; { + select { + case <-followCtx.Done(): + return + default: + } err := f.as.db.Update(func(txn *badger.Txn) error { + return otel.TraceErr(followCtx, "tessera.antispam.badger.follow_txn", tracer, func(ctx context.Context, span trace.Span) error { batchStart := time.Now() ctx, cancel := context.WithTimeout(ctx, defaultBatchTimeout) defer cancel() @@ -315,27 +319,31 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { var err error if followFrom >= logSize { + if stop != nil { + stop() + next, stop = nil, nil + } + // Our view of the log is out of date, update it logSize, err = lr.IntegratedSize(ctx) if err != nil { + // The log probably just hasn't completed its first integration yet, so break out of here + // and go back to sleep for a bit to avoid spamming errors into the log and scaring operators. + streamDone = true if errors.Is(err, os.ErrNotExist) { - // The log probably just hasn't completed its first integration yet, so break out of here - // and go back to sleep for a bit to avoid spamming errors into the log and scaring operators. - moreWork = false return nil } return fmt.Errorf("populate: IntegratedSize(): %v", err) } switch { case followFrom > logSize: - // Since we've got a stale view, there could be more work to do - loop and check without sleeping. - moreWork = true + streamDone = true return fmt.Errorf("followFrom %d > size %d", followFrom, logSize) case followFrom == logSize: // We're caught up, so unblock pushback and go back to sleep - moreWork = false + streamDone = true f.as.pushBack.Store(false) - return nil + return ctx.Err() default: // size > followFrom, so there's more work to be done! } @@ -349,18 +357,20 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // start reading from: if next == nil { span.AddEvent("Start streaming entries") + streamSize := logSize sizeFn := func(_ context.Context) (uint64, error) { - return logSize, nil + return streamSize, nil } + numFetchers := uint(10) // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). - streamCtx, sCancel := context.WithTimeout(trace.ContextWithSpan(followCtx, span), defaultStreamTimeout) - + streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) next, stop = streamNext, func() { + // Cancel first so in-flight fetches abort, then kill the iterator sCancel() streamStop() } @@ -399,6 +409,13 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { curIndex = followFrom } + if len(curEntries) == 0 { + // We didn't manage to read any entries, so there's nothing to commit. Break out of + // the busy loop and wait for the ticker rather than spinning. + streamDone = true + return ctx.Err() + } + // Now update the index. { for i, e := range curEntries { @@ -437,6 +454,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } next, stop = nil, nil streamDone = true + curEntries = nil continue } curEntries = nil From bcef28102194af7bae3b12df5dce2ca1140eca13 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Mon, 21 Sep 2026 13:51:03 +0000 Subject: [PATCH 3/5] Bring badger into line with GCP & AWS. --- storage/aws/antispam/aws.go | 12 +++++- storage/gcp/antispam/gcp.go | 65 ++++++++++++++++++------------ storage/posix/antispam/badger.go | 68 +++++++++++++------------------- 3 files changed, 77 insertions(+), 68 deletions(-) diff --git a/storage/aws/antispam/aws.go b/storage/aws/antispam/aws.go index 0a35a3d35..aeafc6c30 100644 --- a/storage/aws/antispam/aws.go +++ b/storage/aws/antispam/aws.go @@ -316,6 +316,11 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } if followFrom >= logSize { + if stop != nil { + stop() + next, stop = nil, nil + } + // Our view of the log is out of date, update it logSize, err = lr.IntegratedSize(ctx) if err != nil { @@ -373,7 +378,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { break } if err != nil { - return fmt.Errorf("entryReader.next: %v", err) + return fmt.Errorf("entryReader.next: %w", err) } if wantIdx := followFrom + uint64(i); e.Index != wantIdx { // We're out of sync @@ -383,6 +388,9 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } if len(curEntries) == 0 { + // We didn't manage to read any entries, so there's nothing to commit. Break out of + // the busy loop and wait for the ticker rather than spinning. + streamDone = true return ctx.Err() } @@ -416,7 +424,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { return ctx.Err() }) if err != nil { - if err != errOutOfSync { + if !errors.Is(err, errOutOfSync) { slog.ErrorContext(followCtx, "Failed to commit antispam population tx", slog.Any("error", err)) } if stop != nil { diff --git a/storage/gcp/antispam/gcp.go b/storage/gcp/antispam/gcp.go index 2490ab5d5..a873ad4b8 100644 --- a/storage/gcp/antispam/gcp.go +++ b/storage/gcp/antispam/gcp.go @@ -309,6 +309,12 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // Busy loop while there are entries to be consumed from the stream for streamDone := false; !streamDone; { + select { + case <-followCtx.Done(): + return + default: + } + err := otel.TraceErr(followCtx, "tessera.antispam.gcp.FollowTask", tracer, func(ctx context.Context, span trace.Span) error { ctx, cancel := context.WithTimeout(ctx, defaultBatchTimeout) defer cancel() @@ -328,6 +334,11 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { followFrom := uint64(nextIdx) if followFrom >= logSize { + if stop != nil { + stop() + next, stop = nil, nil + } + // Our view of the log is out of date, update it. // We use ctx here because Cloud Spanner doesn't support nested transactions. // This is okay because we're only reading the log size, not modifying anything. @@ -354,35 +365,35 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { span.SetAttributes(pushbackKey.Bool(pushback)) f.as.pushBack.Store(pushback) - // If this is the first time around the loop we need to start the stream of entries now that we know where we want to - // start reading from: - if next == nil { - span.AddEvent("Start streaming entries") - streamSize := logSize - sizeFn := func(_ context.Context) (uint64, error) { - return streamSize, nil - } - - numFetchers := uint(10) - - // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. - // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). - streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) - streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) - - next, stop = streamNext, func() { - // Cancel first so in-flight fetches abort, then kill the iterator - sCancel() - streamStop() - } - } - if curIndex == followFrom && curEntries != nil { // Note that it's possible for Spanner to automatically retry transactions in some circumstances, when it does // it'll call this function again. // If the above condition holds, then we're in a retry situation and we must use the same data again rather // than continue reading entries which will take us out of sync. } else { + // If this is the first time around the loop we need to start the stream of entries now that we know where we want to + // start reading from: + if next == nil { + span.AddEvent("Start streaming entries") + streamSize := logSize + sizeFn := func(_ context.Context) (uint64, error) { + return streamSize, nil + } + + numFetchers := uint(10) + + // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. + // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). + streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) + streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + + next, stop = streamNext, func() { + // Cancel first so in-flight fetches abort, then kill the iterator + sCancel() + streamStop() + } + } + bs := uint64(f.as.opts.MaxBatchSize) if r := logSize - followFrom; r < bs { bs = r @@ -397,7 +408,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { break } if err != nil { - return fmt.Errorf("entryReader.next: %v", err) + return fmt.Errorf("entryReader.next: %w", err) } if wantIdx := followFrom + uint64(i); e.Index != wantIdx { // We're out of sync @@ -410,6 +421,9 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } if len(curEntries) == 0 { + // We didn't manage to read any entries, so there's nothing to commit. Break out of + // the busy loop and wait for the ticker rather than spinning. + streamDone = true return ctx.Err() } @@ -437,7 +451,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { return err }) if err != nil { - if err != errOutOfSync { + if !errors.Is(err, errOutOfSync) { slog.ErrorContext(followCtx, "Failed to commit antispam population tx", slog.Any("error", err)) } if stop != nil { @@ -445,6 +459,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } next, stop = nil, nil streamDone = true + curEntries = nil continue } curEntries = nil diff --git a/storage/posix/antispam/badger.go b/storage/posix/antispam/badger.go index cb965d03d..8a807c9df 100644 --- a/storage/posix/antispam/badger.go +++ b/storage/posix/antispam/badger.go @@ -263,9 +263,6 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { var ( next func() (client.Entry[[]byte], error, bool) stop func() - - curEntries [][]byte - curIndex uint64 ) // Ensure we tear down any in-flight entry stream when we're done. defer func() { @@ -343,7 +340,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // We're caught up, so unblock pushback and go back to sleep streamDone = true f.as.pushBack.Store(false) - return ctx.Err() + return nil default: // size > followFrom, so there's more work to be done! } @@ -376,40 +373,31 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } } - if curIndex == followFrom && curEntries != nil { - // Note that it's possible for Spanner to automatically retry transactions in some circumstances, when it does - // it'll call this function again. - // If the above condition holds, then we're in a retry situation and we must use the same data again rather - // than continue reading entries which will take us out of sync. - } else { - bs := uint64(f.as.opts.MaxBatchSize) - if r := logSize - followFrom; r < bs { - bs = r + bs := uint64(f.as.opts.MaxBatchSize) + if r := logSize - followFrom; r < bs { + bs = r + } + batch := make([][]byte, 0, bs) + for i := range int(bs) { + e, err, ok := next() + if !ok { + // The entry stream has ended so we'll need to start a new stream next time around the loop: + stop() + next, stop = nil, nil + break } - batch := make([][]byte, 0, bs) - for i := range int(bs) { - e, err, ok := next() - if !ok { - // The entry stream has ended so we'll need to start a new stream next time around the loop: - stop() - next, stop = nil, nil - break - } - if err != nil { - return fmt.Errorf("entryReader.next: %v", err) - } - if wantIdx := followFrom + uint64(i); e.Index != wantIdx { - slog.InfoContext(ctx, "Out of sync", slog.Uint64("index", e.Index), slog.Uint64("wantidx", wantIdx)) - // We're out of sync - return errOutOfSync - } - batch = append(batch, e.Entry) + if err != nil { + return fmt.Errorf("entryReader.next: %w", err) + } + if wantIdx := followFrom + uint64(i); e.Index != wantIdx { + slog.InfoContext(ctx, "Out of sync", slog.Uint64("index", e.Index), slog.Uint64("wantidx", wantIdx)) + // We're out of sync + return errOutOfSync } - curEntries = batch - curIndex = followFrom + batch = append(batch, e.Entry) } - if len(curEntries) == 0 { + if len(batch) == 0 { // We didn't manage to read any entries, so there's nothing to commit. Break out of // the busy loop and wait for the ticker rather than spinning. streamDone = true @@ -418,10 +406,10 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // Now update the index. { - for i, e := range curEntries { + for i, e := range batch { if _, err := txn.Get(e); err == badger.ErrKeyNotFound { b := make([]byte, 8) - binary.BigEndian.PutUint64(b, curIndex+uint64(i)) + binary.BigEndian.PutUint64(b, followFrom+uint64(i)) if err := txn.Set(e, b); err != nil { return err } @@ -429,11 +417,11 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } } - numAdded := uint64(len(curEntries)) + numAdded := uint64(len(batch)) // and update the follower state b := make([]byte, 8) - binary.BigEndian.PutUint64(b, curIndex+numAdded) + binary.BigEndian.PutUint64(b, followFrom+numAdded) if err := txn.Set(nextKey, b); err != nil { return fmt.Errorf("failed to update follower state: %v", err) } @@ -446,7 +434,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { }) }) if err != nil { - if err != errOutOfSync { + if !errors.Is(err, errOutOfSync) { slog.ErrorContext(followCtx, "Failed to commit antispam population tx", slog.Any("error", err)) } if stop != nil { @@ -454,10 +442,8 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { } next, stop = nil, nil streamDone = true - curEntries = nil continue } - curEntries = nil } } } From fd278215112d1d32759ca3492253657b1de7d666 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Mon, 21 Sep 2026 16:31:27 +0000 Subject: [PATCH 4/5] Fix client bug --- client/stream.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/client/stream.go b/client/stream.go index 7b039557f..22ddf0d2b 100644 --- a/client/stream.go +++ b/client/stream.go @@ -93,6 +93,12 @@ func EntryBundles(ctx context.Context, numWorkers uint, getSize TreeSizeFunc, ge // to resolve it. for ri := range layout.Range(fromEntry, N, treeSize) { select { + case <-ctx.Done(): + select { + case bundles <- func() bundleOrErr { return bundleOrErr{err: ctx.Err()} }: + case <-exit: + } + return case <-exit: return case <-tokens: From 93e6051307704536737e4790e1dd961f8c57c03a Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Tue, 22 Sep 2026 10:20:08 +0000 Subject: [PATCH 5/5] Add a timeout for reading individual bundles --- storage/aws/antispam/aws.go | 17 ++++++++++++++++- storage/gcp/antispam/gcp.go | 17 ++++++++++++++++- storage/posix/antispam/badger.go | 17 ++++++++++++++++- 3 files changed, 48 insertions(+), 3 deletions(-) diff --git a/storage/aws/antispam/aws.go b/storage/aws/antispam/aws.go index aeafc6c30..add93af82 100644 --- a/storage/aws/antispam/aws.go +++ b/storage/aws/antispam/aws.go @@ -53,6 +53,8 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second + // defaultReadTimeout is the maximum duration to spend waiting for reads to complete. + defaultReadTimeout = 5 * time.Second ) // AntispamOpts allows configuration of some tunable options. @@ -258,9 +260,22 @@ func (f *follower) Name() string { return "AWS antispam" } +// readBundleWithTimeout returns a wrapper around a function which reads an entry bundle, +// which has the effect of adding a timeout to the context before calling the wrapped function. +func readBundleWithTimeout(timeout time.Duration, r func(context.Context, uint64, uint8) ([]byte, error)) func(context.Context, uint64, uint8) ([]byte, error) { + return func(ctx context.Context, idx uint64, p uint8) ([]byte, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + return r(ctx, idx, p) + } +} + // Follow uses entry data from the log to populate the antispam storage. func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { errOutOfSync := errors.New("out-of-sync") + // The timeout here defines how long we're prepared to wait for a single bundle to be read. + // If the log is slow to respond, we'll stop, and (potentially) restart streaming later. + readBundle := readBundleWithTimeout(defaultReadTimeout, lr.ReadEntryBundle) t := time.NewTicker(time.Second) var ( @@ -355,7 +370,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) - streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, readBundle, followFrom, logSize-followFrom), f.bundleHasher)) next, stop = streamNext, func() { // Cancel first so in-flight fetches abort, then kill the iterator diff --git a/storage/gcp/antispam/gcp.go b/storage/gcp/antispam/gcp.go index a873ad4b8..b1eb9de07 100644 --- a/storage/gcp/antispam/gcp.go +++ b/storage/gcp/antispam/gcp.go @@ -52,6 +52,8 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second + // defaultReadTimeout is the maximum duration to spend waiting for reads to complete. + defaultReadTimeout = 5 * time.Second ) // AntispamOpts allows configuration of some tunable options. @@ -278,9 +280,22 @@ func (f *follower) Name() string { return "GCP antispam" } +// readBundleWithTimeout returns a wrapper around a function which reads an entry bundle, +// which has the effect of adding a timeout to the context before calling the wrapped function. +func readBundleWithTimeout(timeout time.Duration, r func(context.Context, uint64, uint8) ([]byte, error)) func(context.Context, uint64, uint8) ([]byte, error) { + return func(ctx context.Context, idx uint64, p uint8) ([]byte, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + return r(ctx, idx, p) + } +} + // Follow uses entry data from the log to populate the antispam storage. func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { errOutOfSync := errors.New("out-of-sync") + // The timeout here defines how long we're prepared to wait for a single bundle to be read. + // If the log is slow to respond, we'll stop, and (potentially) restart streaming later. + readBundle := readBundleWithTimeout(defaultReadTimeout, lr.ReadEntryBundle) t := time.NewTicker(time.Second) var ( @@ -385,7 +400,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) - streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, readBundle, followFrom, logSize-followFrom), f.bundleHasher)) next, stop = streamNext, func() { // Cancel first so in-flight fetches abort, then kill the iterator diff --git a/storage/posix/antispam/badger.go b/storage/posix/antispam/badger.go index 8a807c9df..024072abf 100644 --- a/storage/posix/antispam/badger.go +++ b/storage/posix/antispam/badger.go @@ -45,6 +45,8 @@ const ( // defaultBatchTimeout is the max permitted duration for a single "chunk" of antispam updates. defaultBatchTimeout = 10 * time.Second + // defaultReadTimeout is the maximum duration to spend waiting for reads to complete. + defaultReadTimeout = 5 * time.Second ) var ( @@ -255,9 +257,22 @@ func (f *follower) Name() string { return "Badger antispam" } +// readBundleWithTimeout returns a wrapper around a function which reads an entry bundle, +// which has the effect of adding a timeout to the context before calling the wrapped function. +func readBundleWithTimeout(timeout time.Duration, r func(context.Context, uint64, uint8) ([]byte, error)) func(context.Context, uint64, uint8) ([]byte, error) { + return func(ctx context.Context, idx uint64, p uint8) ([]byte, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + return r(ctx, idx, p) + } +} + // Follow uses entry data from the log to populate the antispam storage. func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { errOutOfSync := errors.New("out-of-sync") + // The timeout here defines how long we're prepared to wait for a single bundle to be read. + // If the log is slow to respond, we'll stop, and (potentially) restart streaming later. + readBundle := readBundleWithTimeout(defaultReadTimeout, lr.ReadEntryBundle) t := time.NewTicker(time.Second) var ( @@ -364,7 +379,7 @@ func (f *follower) Follow(followCtx context.Context, lr tessera.LogReader) { // Start a new streaming read of entries, using a fresh context rooted in the "outermost" context passed to Follow. // This allows this stream to be re-used across loops where the stop function is not called (e.g. when we hit a conflict). streamCtx, sCancel := context.WithCancel(trace.ContextWithSpan(followCtx, span)) - streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher)) + streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, readBundle, followFrom, logSize-followFrom), f.bundleHasher)) next, stop = streamNext, func() { // Cancel first so in-flight fetches abort, then kill the iterator