diff --git a/internal/connector/blocked_retry_test.go b/internal/connector/blocked_retry_test.go index cf3a262e4..550a314ef 100644 --- a/internal/connector/blocked_retry_test.go +++ b/internal/connector/blocked_retry_test.go @@ -204,9 +204,9 @@ func TestARecordTheScheduleIsFinishedWithLeavesTheSweepEntirely(t *testing.T) { scope := BlockedRetryScope{Now: clock.Now(), Limit: 10} assert.Equal(t, []int64{2}, dueIDs(t, ledger, scope)) - scheduled, err := ledger.ScheduledBlockedIDs(context.Background(), scope) + live, err := ledger.StillScheduledBlockedIDs(context.Background(), scope, []int64{1, 2}) require.NoError(t, err) - assert.Equal(t, []int64{2}, scheduled, "nothing keeps a claim for a record it can never offer") + assert.Equal(t, []int64{2}, live, "nothing keeps a claim for a record it can never offer") } // retryIntake is an intake and ledger on one clock the test moves, with @@ -624,7 +624,7 @@ func TestTheSweepsQueriesUseTheDueTimeIndex(t *testing.T) { // assertion is on the same SQL the connector executes. _, err := ledger.DueBlockedRetries(ctx, scope) require.NoError(t, err) - _, err = ledger.ScheduledBlockedIDs(ctx, scope) + _, err = ledger.StillScheduledBlockedIDs(ctx, scope, []int64{1}) require.NoError(t, err) where, args := scheduledBlockedWhere(nil) @@ -635,8 +635,64 @@ func TestTheSweepsQueriesUseTheDueTimeIndex(t *testing.T) { assert.NotContains(t, due, "events_state_id", "not by walking every blocked row the ledger holds") assert.NotContains(t, due, "TEMP B-TREE", "and the order is the index's own, so nothing is sorted") - scheduled := queryPlan(t, ledger, `SELECT id FROM events`+where+blockedRetryOrder, args...) - assert.Contains(t, scheduled, "events_next_retry", "so is the live schedule the claims are pruned against") - assert.NotContains(t, scheduled, "events_state_id") - assert.NotContains(t, scheduled, "TEMP B-TREE") + // The pruning read is by id, so the rowid finds it directly: the point of + // the assertion is that it visits the claims it was handed and not the + // blocked history behind them, whichever access path SQLite picks. + pruneWhere, pruneArgs := scheduledBlockedWhere(nil) + pruneWhere += ` AND id IN (` + placeholders(1) + `)` + scheduled := queryPlan(t, ledger, `SELECT id FROM events`+pruneWhere, append(append([]any{}, pruneArgs...), int64(1))...) + assert.NotContains(t, scheduled, "SCAN events", "the claims are looked up, never scanned for") + assert.NotContains(t, scheduled, "events_state_id", "and never through the whole blocked history") +} + +// Copilot on #770, the fourth time this mechanism found a way to cost the +// backlog instead of the batch. The claim map is pruned against the schedule, +// and the read that pruned it asked for the whole live schedule: every sweep +// materialized every scheduled record to decide the fate of a handful of +// claims, and the very first sweep — holding no claims at all — paid for it +// too. After an outage that is the entire outage, once a minute, in the +// goroutine the stranded and loss repair share. +// +// The answer was never wrong, which is why nothing caught it: the assertion +// has to be on the work, so the ledger counts the reads. +func TestPruningAsksAboutTheClaimsHeldAndNotTheBacklog(t *testing.T) { + intake, ledger, _, clock := retryIntake(t) + ctx := context.Background() + for id := int64(1); id <= 50; id++ { + blockRecord(t, ledger, id, adapterBucketID, admission.ReasonReadFailed) + } + clock.Advance(admission.BlockedRetryInterval + time.Minute) + + // Holding nothing, there is nothing to prune, and no reason to ask. + intake.sweepBlockedRetries(ctx) + assert.Zero(t, ledger.blockedScheduleReads.Load(), + "a sweep holding no claims reads no schedule") + + held := intake.claimedBlockedIDs() + require.NotEmpty(t, held, "the sweep offered and so is holding claims") + + // Holding some, it asks about those. The read is bounded by the claims, + // so it carries as many ids as there are claims and no more. + before := ledger.blockedScheduleReads.Load() + live, err := ledger.StillScheduledBlockedIDs(ctx, BlockedRetryScope{Now: clock.Now()}, held) + require.NoError(t, err) + assert.Equal(t, before+1, ledger.blockedScheduleReads.Load()) + assert.LessOrEqual(t, len(live), len(held), + "the answer is about the claims asked about, not about the 50 records scheduled") +} + +// The safety property the bound must not cost: pruning drops a claim only +// when the ledger was asked about it and did not name it. A claim taken while +// the read was in flight was never put to the ledger, and dropping it for not +// coming back would offer its record a second time. +func TestPruningNeverDropsAClaimItDidNotAskAbout(t *testing.T) { + intake, _, _, _ := retryIntake(t) + intake.retriedBlocked = map[int64]int64{7: 1, 9: 1} + + // Asked about 7 alone, and told it is finished with. 9 arrived after the + // question and is untouched by the answer. + intake.pruneBlockedClaims([]int64{7}, nil) + + assert.NotContains(t, intake.retriedBlocked, int64(7), "asked about, and not named: dropped") + assert.Contains(t, intake.retriedBlocked, int64(9), "never asked about: kept") } diff --git a/internal/connector/intake.go b/internal/connector/intake.go index 63a15b445..d881c63fe 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -908,7 +908,9 @@ const blockedRetryBatch = 100 // It claims before it offers, and gives the claim back when the offer fails, // so a record is never left claimed and unoffered. It also drops the claims // the ledger has nothing left to say about, which is what keeps the claim set -// the size of the backlog rather than the size of the history. +// the size of the backlog rather than the size of the history — asking about +// the claims it holds, so that pruning costs the backlog in flight and not +// the whole retained schedule. // // The batch counts records offered, not records read. A row already claimed // is due and will stay due until admission decides it, so a batch that @@ -930,10 +932,16 @@ func (in *Intake) sweepBlockedRetries(ctx context.Context) { // Pruned before the offers, from one reading: a record the schedule is // finished with cannot come back, and a record that is still on it keeps // its claim whether or not it is due in this tick. - if scheduled, err := in.ledger.ScheduledBlockedIDs(ctx, scope); err != nil { - in.log.Warn("could not read the blocked records still on the retry schedule", "error", err) - } else { - in.pruneBlockedClaims(scheduled) + // + // The reading is of the claims held, not of the schedule: holding none is + // the common case and asks nothing at all, and holding some asks about + // those and not about the backlog behind them. + if held := in.claimedBlockedIDs(); len(held) > 0 { + if live, err := in.ledger.StillScheduledBlockedIDs(ctx, scope, held); err != nil { + in.log.Warn("could not read the blocked records still on the retry schedule", "error", err) + } else { + in.pruneBlockedClaims(held, live) + } } for offered := 0; offered < batch; { records, err := in.ledger.DueBlockedRetries(ctx, scope) @@ -968,21 +976,42 @@ func (in *Intake) sweepBlockedRetries(ctx context.Context) { } } -// pruneBlockedClaims keeps the claims of the records still on the schedule -// and drops the rest. A dropped claim can only belong to a record no sweep -// will offer again, so dropping it cannot cause a second offer. -func (in *Intake) pruneBlockedClaims(scheduled []int64) { +// claimedBlockedIDs is the ids the sweep currently holds a retry claim for. +// It is the question the pruning read asks about, and it bounds that read by +// the backlog in flight rather than by the whole retained schedule. +func (in *Intake) claimedBlockedIDs() []int64 { + in.mu.Lock() + defer in.mu.Unlock() + if len(in.retriedBlocked) == 0 { + return nil + } + held := make([]int64, 0, len(in.retriedBlocked)) + for id := range in.retriedBlocked { + held = append(held, id) + } + return held +} + +// pruneBlockedClaims drops the claims among asked whose records the schedule +// is finished with — the ones the ledger did not name in live. A dropped +// claim can only belong to a record no sweep will offer again, so dropping it +// cannot cause a second offer. +// +// It considers only the ids it asked about. A claim taken while the read was +// in flight was never put to the ledger, and dropping it for not coming back +// would offer its record a second time. +func (in *Intake) pruneBlockedClaims(asked, live []int64) { in.mu.Lock() defer in.mu.Unlock() if len(in.retriedBlocked) == 0 { return } - live := make(map[int64]struct{}, len(scheduled)) - for _, id := range scheduled { - live[id] = struct{}{} + scheduled := make(map[int64]struct{}, len(live)) + for _, id := range live { + scheduled[id] = struct{}{} } - for id := range in.retriedBlocked { - if _, ok := live[id]; !ok { + for _, id := range asked { + if _, ok := scheduled[id]; !ok { delete(in.retriedBlocked, id) } } diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go index fcd592181..bf837272e 100644 --- a/internal/connector/ledger.go +++ b/internal/connector/ledger.go @@ -85,6 +85,11 @@ type Ledger struct { closed sync.Once now func() time.Time hooks Hooks + // blockedScheduleReads counts the reads the retry sweep makes against the + // schedule to prune its claims. The sweep must make none when it holds no + // claims: the read it used to make was over the whole live backlog, and a + // test that only checks the answer cannot see the work (Copilot on #770). + blockedScheduleReads atomic.Int64 } // OpenLedger opens (creating if absent) the ledger at path and brings its diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index 2a4e40176..59f8a0f6b 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -275,30 +275,58 @@ func (l *Ledger) DueBlockedRetries(ctx context.Context, scope BlockedRetryScope) // history. const blockedRetryOrder = ` ORDER BY next_retry_at, id` -// ScheduledBlockedIDs is every record in scope the retry schedule still has +// StillScheduledBlockedIDs is the subset of ids the retry schedule still has // something to do about, due now or later. // -// It is the live bound on the sweep's claims: a claim is worth keeping only -// while the record it names can still be offered, and a record that has been -// decided, or whose window has passed, can never be. Without it the claim set -// held one entry per record ever retried and dropped none (Copilot on #770). +// It answers the only question the sweep asks — "of the records I hold a +// claim for, which can still be offered?" — rather than the one it used to +// ask, which was for the entire live backlog. For the claims held those have +// the same answer and wildly different costs: after an outage the schedule +// holds every record the outage blocked, and a sweep reading all of them once +// a minute to prune a handful of claims does work proportional to the backlog +// instead of to the claims, and delays the stranded and loss repair sharing +// its goroutine (Copilot on #770). // -// It is deliberately unlimited. A short answer would read as "these are all -// the records still on the schedule" and prune the claims of the ones it left -// out, which is the offer-twice this whole mechanism exists to prevent. The -// set it returns is the connector's live backlog, not its history. +// A claim is worth keeping only while the record it names can still be +// offered, and a record that has been decided, or whose window has passed, +// can never be. Asking about exactly the claimed ids is what makes pruning +// safe: the caller drops only the ids it asked about and did not get back, so +// an id this was never asked about cannot be pruned by its absence. // -// It comes back in the schedule's order rather than by id, for the reason -// DueBlockedRetries does: asked for id order, SQLite reads every blocked row -// the ledger holds instead of only the scheduled ones. The caller reads it as -// a set, so the order is the index's to choose. -func (l *Ledger) ScheduledBlockedIDs(ctx context.Context, scope BlockedRetryScope) ([]int64, error) { - where, args := scheduledBlockedWhere(scope.Buckets) - //nolint:gosec // G202: the clauses are this package's constants and placeholders, never values - rows, err := l.db.QueryContext(ctx, `SELECT id FROM events`+where+blockedRetryOrder, args...) - if err != nil { - return nil, fmt.Errorf("connector: list blocked records on the retry schedule: %w", err) +// It reads in chunks and returns their union. A chunk dropped or shortened +// would read as "no longer scheduled" and prune live claims, which is the +// offer-twice the claim exists to prevent, so a failed chunk fails the call. +func (l *Ledger) StillScheduledBlockedIDs(ctx context.Context, scope BlockedRetryScope, ids []int64) ([]int64, error) { + if len(ids) == 0 { + return nil, nil } + l.blockedScheduleReads.Add(1) + var live []int64 + for chunk := range slices.Chunk(ids, scheduledIDChunk) { + where, args := scheduledBlockedWhere(scope.Buckets) + where += ` AND id IN (` + placeholders(len(chunk)) + `)` + for _, id := range chunk { + args = append(args, id) + } + //nolint:gosec // G202: the clauses are this package's constants and placeholders, never values + rows, err := l.db.QueryContext(ctx, `SELECT id FROM events`+where, args...) + if err != nil { + return nil, fmt.Errorf("connector: list the claimed blocked records still on the retry schedule: %w", err) + } + chunkLive, err := scanIDs(rows) + if err != nil { + return nil, fmt.Errorf("connector: list the claimed blocked records still on the retry schedule: %w", err) + } + live = append(live, chunkLive...) + } + return live, nil +} + +// scheduledIDChunk bounds the placeholders one read may carry, well under +// SQLite's variable limit with the bucket scope's own placeholders alongside. +const scheduledIDChunk = 400 + +func scanIDs(rows *sql.Rows) ([]int64, error) { defer func() { _ = rows.Close() }() var ids []int64 for rows.Next() {