Skip to content

Improve & align antispam followers #1193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
AlCutter wants to merge 3 commits into
base: main
Choose a base branch
from
+137 −35
Draft

Improve & align antispam followers #1193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ef399b6
Improve antispam follower efficiency
AlCutter Sep 21, 2026
234a610
Bring badger into line with GCP & AWS.
AlCutter Sep 21, 2026
159b442
Fix client bug
AlCutter Sep 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions client/stream.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ 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():
return
case <-exit:
return
case <-tokens:
Expand Down
48 changes: 37 additions & 11 deletions storage/aws/antispam/aws.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -259,17 +261,23 @@ 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)
var (
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:
}
Expand All @@ -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()

Expand All @@ -310,6 +318,11 @@ func (f *follower) Follow(ctx 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 {
Expand All @@ -335,11 +348,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(followCtx, defaultStreamTimeout)

streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher))

next, stop = streamNext, func() {
streamStop()
sCancel()
}
}

bs := uint64(f.as.opts.MaxBatchSize)
Expand All @@ -352,7 +376,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 {
Expand All @@ -366,6 +390,9 @@ func (f *follower) Follow(ctx 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()
}

Expand Down Expand Up @@ -400,13 +427,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
}
Expand Down
51 changes: 43 additions & 8 deletions storage/gcp/antispam/gcp.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand All @@ -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:
}
Expand All @@ -303,7 +311,13 @@ 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 {
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()
_, err := f.as.dbPool.ReadWriteTransactionWithOptions(ctx, func(txctx context.Context, txn *spanner.ReadWriteTransaction) error {
Expand All @@ -322,6 +336,11 @@ func (f *follower) Follow(ctx 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.
Expand Down Expand Up @@ -352,11 +371,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(followCtx, defaultStreamTimeout)

streamNext, streamStop := iter.Pull2(client.Entries(client.EntryBundles(streamCtx, numFetchers, sizeFn, lr.ReadEntryBundle, followFrom, logSize-followFrom), f.bundleHasher))

next, stop = streamNext, func() {
streamStop()
sCancel()
}
}

if curIndex == followFrom && curEntries != nil {
Expand All @@ -375,7 +406,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 {
Expand All @@ -392,6 +423,9 @@ func (f *follower) Follow(ctx 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()
}

Expand Down Expand Up @@ -420,13 +454,14 @@ 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
curEntries = nil
continue
}
curEntries = nil
Expand Down
Loading
Loading