The retry sweep read the whole schedule to prune a handful of claims

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")
 }

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)
 		}
 	}

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

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() {