From 4ecba0af8a972c7ecdad55533dc46302101b430f Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 09:16:16 +0200 Subject: [PATCH 1/5] A blocked record's timed retry runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit admission.NextBlockedRetry names the blocked reasons that should come round again, computes when each is next due, and nothing called it. The only path that re-offered a blocked record was `basecamp connect redispatch `, one id at a time, so a read that failed during an outage, a throttle, and a connect.json that could not be read all sat until a person noticed each one. The intake sweep that already re-offers stranded events and open losses now also asks the ledger which blocked records are due and offers those. The schedule stays where it was: NextBlockedRetry is still the one place that says when a record is owed an attempt, the ledger's query narrows to the rows it could say yes about, and the sweep only moves them. Three things the attempt removed from #765 was taken out over: - Scope. The query takes the run's --project buckets and will not return a record outside them. out_of_scope is a terminal discard and the projects a run leaves out are another run's to dispatch, so a retry that crossed the scope would cause the permanent loss it exists to prevent. - The claim. Queue.Offer does not deduplicate, so a due row is claimed before it is offered and the claim is given back if the offer fails. The claim is the revision a record was last offered at: a verdict bumps the revision, so the record's own next decision retires it. In memory is enough — AcquireInstanceLock is a flock the kernel drops on process death, so exactly one connector ever sweeps a ledger. - The join. Two tests: one drives the sweep, one drives the real ticker with nothing else offering anything. Deleting the hook fails the second and leaves the first green. config_unreadable now retries with no window at all, which is the one change to the schedule itself. A day is the right bound for a failing server; a broken connect.json is a local file, and an operator away for a week is ordinary. no_route, bucket_mismatch and unroutable stay untimed: the first waits on a decision, the other two on facts no later read changes. The comments across ledger.go, ledger_events.go, admission/verdict.go and retraction_test.go that already claimed blocked records were "retained and retried" are true as of this change, and point at what runs them. ledger_decisions.go's AuthorizedBlocked still has no caller and now says so. --- .../connector/admission/admission_test.go | 20 +- internal/connector/admission/commit.go | 57 +++- internal/connector/admission/commit_test.go | 38 +++ internal/connector/admission/matrix.go | 11 +- internal/connector/admission/verdict.go | 6 +- internal/connector/blocked_retry_test.go | 298 ++++++++++++++++++ internal/connector/intake.go | 110 ++++++- internal/connector/ledger.go | 4 +- internal/connector/ledger_decisions.go | 9 +- internal/connector/ledger_events.go | 105 +++++- internal/connector/retraction_test.go | 3 +- 11 files changed, 623 insertions(+), 38 deletions(-) create mode 100644 internal/connector/blocked_retry_test.go diff --git a/internal/connector/admission/admission_test.go b/internal/connector/admission/admission_test.go index c09a1051d..43f3ad57a 100644 --- a/internal/connector/admission/admission_test.go +++ b/internal/connector/admission/admission_test.go @@ -1138,7 +1138,8 @@ func TestTheGateReadsTheLiveServedSetToo(t *testing.T) { // and post the public holding reply — telling the person on the card that // their project is not served, when the project is there and the file is // what is broken. no_route also has no timed retry, so repairing the file -// would not reconsider those records. +// would not reconsider those records — where config_unreadable is retried on +// a timer with no window at all, so it does. func TestAnUnreadableConfigIsHeldAsOneRatherThanAnsweredAsUnserved(t *testing.T) { broken := errors.New("connect.json cannot be read") fail := true @@ -1157,14 +1158,17 @@ func TestAnUnreadableConfigIsHeldAsOneRatherThanAnsweredAsUnserved(t *testing.T) assert.Equal(t, ReasonConfigUnreadable, v.Reason, "not no_route: nothing read the file, so nothing can say the project is unserved") assert.False(t, v.Served) - // It waits for a person, as every blocked record does. An automatic - // sweep was built for this and taken back out: offering due blocked - // rows is a scheduler with its own claiming, and it re-decided five - // other blocked reasons besides this one. Carded, so that nothing here - // promises a timer that does not run. + // It comes round on its own, every ten minutes, for as long as the file + // is unreadable. The window the other blocked reasons stop at is right + // for a server that has failed for a day; it is wrong for a local file, + // where an operator away for a week is ordinary and giving up would + // strand the work silently. blockedAt := time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC) - _, retried := NextBlockedRetry(ReasonConfigUnreadable, blockedAt, blockedAt, time.Time{}) - assert.False(t, retried, "no schedule claims this record, and nothing would act on one if it did") + next, retried := NextBlockedRetry(ReasonConfigUnreadable, blockedAt, blockedAt, time.Time{}) + require.True(t, retried, "the sweep re-offers this one without anybody asking") + assert.Equal(t, blockedAt.Add(BlockedRetryInterval), next) + _, retried = NextBlockedRetry(ReasonConfigUnreadable, blockedAt, blockedAt.Add(8*24*time.Hour), time.Time{}) + assert.True(t, retried, "a week of it is an operator on holiday, not a verdict") // And once the file is readable the record decides normally. fail = false diff --git a/internal/connector/admission/commit.go b/internal/connector/admission/commit.go index 563fe3815..3830c7acc 100644 --- a/internal/connector/admission/commit.go +++ b/internal/connector/admission/commit.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "sync" "time" ) @@ -135,33 +136,71 @@ func (k *keyedMutex) lock(ctx context.Context, key string) (func(), error) { // unverified assignment delta is retried every ten minutes for a day after it // was first blocked, and on redispatch at any time after that. A throttled // record (throttled) is on the same schedule, never before the server's -// deadline. bucket_mismatch -// and unroutable are not timed: the pointer's bucket and type never change, so -// only a person's redispatch re-runs them. A blocked record is never -// discarded for having failed: the checkpoint may already be past the event, -// and a tombstone would turn an outage into a permanent loss. +// deadline. A record blocked because connect.json could not be read +// (config_unreadable) is on the same interval with no window at all: that is +// a local file a person repairs, and an operator away for a week is +// ordinary, where a day of a failing server means the server is not coming +// back on its own. bucket_mismatch and unroutable are not timed: the +// pointer's bucket and type never change, so only a person's redispatch +// re-runs them. no_route is not timed either — it waits for the operator to +// serve the project, which is a decision, not a delay. A blocked record is +// never discarded for having failed: the checkpoint may already be past the +// event, and a tombstone would turn an outage into a permanent loss. +// +// The intake sweep is what runs this schedule (internal/connector, +// Intake.sweepBlockedRetries): it asks the ledger for the records this +// function calls due and offers them back to admission. const ( BlockedRetryInterval = 10 * time.Minute BlockedRetryWindow = 24 * time.Hour ) +// timedBlockedReasons is every blocked reason the schedule re-runs, and for +// each whether its retries stop at BlockedRetryWindow. +// +// One list, read by both halves of the join: NextBlockedRetry decides whether +// a record is due, and the ledger's query narrows to these reasons so the +// sweep never reads the rows that wait for a person. +var timedBlockedReasons = map[Reason]bool{ + ReasonReadFailed: true, + ReasonReadUnresolved: true, + ReasonDeltaUnverified: true, + ReasonTrustUnverified: true, + ReasonThrottled: true, + ReasonConfigUnreadable: false, +} + +// TimedBlockedReasons is the blocked reasons NextBlockedRetry schedules, +// sorted so a query built from them is the same query every time. +func TimedBlockedReasons() []Reason { + out := make([]Reason, 0, len(timedBlockedReasons)) + for reason := range timedBlockedReasons { + out = append(out, reason) + } + slices.Sort(out) + return out +} + // NextBlockedRetry returns when a blocked record should next be re-run, and // false when it waits for something other than time: the operator serving the // project (no_route), or a person's redispatch once the window has passed. // notBefore is a throttled record's Verdict.RetryAt; no retry is scheduled // before it, and a deadline past the window hands the record to redispatch // rather than asking early. +// +// The returned time may be in the past — a connector that was not running +// when a retry came due is late, not excused — so a caller asks whether next +// is at or before now, never whether it is in the future. func NextBlockedRetry(reason Reason, blockedAt, lastAttempt, notBefore time.Time) (time.Time, bool) { - switch reason { - case ReasonReadFailed, ReasonReadUnresolved, ReasonDeltaUnverified, ReasonTrustUnverified, ReasonThrottled: - default: + bounded, timed := timedBlockedReasons[reason] + if !timed { return time.Time{}, false } next := lastAttempt.Add(BlockedRetryInterval) if notBefore.After(next) { next = notBefore } - if next.After(blockedAt.Add(BlockedRetryWindow)) { + if bounded && next.After(blockedAt.Add(BlockedRetryWindow)) { return time.Time{}, false } return next, true diff --git a/internal/connector/admission/commit_test.go b/internal/connector/admission/commit_test.go index 9473b7e7b..ac78b1704 100644 --- a/internal/connector/admission/commit_test.go +++ b/internal/connector/admission/commit_test.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "io" + "slices" "strings" "sync" "sync/atomic" @@ -246,6 +247,43 @@ func TestNextBlockedRetry(t *testing.T) { _, ok = NextBlockedRetry(ReasonNoRoute, blockedAt, blockedAt, time.Time{}) assert.False(t, ok, "no_route waits for connect.json, not for time") + + // config_unreadable is the one reason with no window. A day of a failing + // server means the server is not coming back on its own; a broken + // connect.json is a local file, and an operator away for a week is + // ordinary. + next, ok = NextBlockedRetry(ReasonConfigUnreadable, blockedAt, blockedAt, time.Time{}) + require.True(t, ok) + assert.Equal(t, blockedAt.Add(10*time.Minute), next) + next, ok = NextBlockedRetry(ReasonConfigUnreadable, blockedAt, blockedAt.Add(30*24*time.Hour), time.Time{}) + require.True(t, ok, "a month later it is still asking") + assert.Equal(t, blockedAt.Add(30*24*time.Hour+10*time.Minute), next) +} + +// The reasons the schedule names are one list, and the ledger's query is +// built from it: a reason added to NextBlockedRetry and not to the list would +// be scheduled and never read. +func TestTimedBlockedReasonsIsEveryReasonTheScheduleRuns(t *testing.T) { + every := []Reason{ + ReasonInvalidPointer, ReasonNotInMatrix, ReasonAgentAuthored, ReasonDelegated, + ReasonOutOfScope, ReasonUntrustedPerformer, ReasonAssignmentNotOperator, + ReasonUntrustedAuthor, ReasonStale, ReasonNotAddressed, + ReasonReadFailed, ReasonReadUnresolved, ReasonThrottled, ReasonTrustUnverified, + ReasonDeltaUnverified, ReasonBucketMismatch, ReasonUnroutable, ReasonNoRoute, + ReasonConfigUnreadable, + } + var scheduled []Reason + for _, reason := range every { + if _, ok := NextBlockedRetry(reason, testNow, testNow, time.Time{}); ok { + scheduled = append(scheduled, reason) + } + } + slices.Sort(scheduled) + assert.Equal(t, TimedBlockedReasons(), scheduled) + assert.Equal(t, []Reason{ + ReasonConfigUnreadable, ReasonDeltaUnverified, ReasonReadFailed, + ReasonReadUnresolved, ReasonThrottled, ReasonTrustUnverified, + }, TimedBlockedReasons(), "the six reasons automatic re-decision is turned on for") } type sliceSource struct { diff --git a/internal/connector/admission/matrix.go b/internal/connector/admission/matrix.go index 763fe21b6..c009978ae 100644 --- a/internal/connector/admission/matrix.go +++ b/internal/connector/admission/matrix.go @@ -152,10 +152,11 @@ const ( // would be a false thing to say — and to post a holding reply about — // when the truth is that nothing could read the file. // - // Like every other blocked reason, it waits for a person: repairing the - // file does not by itself decide these records, and `basecamp connect - // redispatch ` is what runs them. Nothing in the connector re-offers - // a blocked record on a timer — NextBlockedRetry describes a schedule - // no production code asks for — so this comment does not promise one. + // It is the one blocked reason with no window: the intake sweep re-offers + // it every BlockedRetryInterval for as long as it stands (NextBlockedRetry, + // internal/connector Intake.sweepBlockedRetries). A day is the right bound + // for a failing server, which is not coming back on its own after one; a + // broken connect.json is a local file, and an operator away for a week is + // ordinary. `basecamp connect redispatch ` still runs one at once. ReasonConfigUnreadable Reason = "config_unreadable" ) diff --git a/internal/connector/admission/verdict.go b/internal/connector/admission/verdict.go index 237baf4ad..237c14850 100644 --- a/internal/connector/admission/verdict.go +++ b/internal/connector/admission/verdict.go @@ -263,9 +263,9 @@ func (a *Admitter) Decide(ctx context.Context, ev Event) (out Verdict, err error // could read which projects are served. Held rather than // discarded: a discard is the one outcome repairing the file // cannot reverse, and a blocked record can still be run. It - // waits for a person — `basecamp connect redispatch ` once - // the file is back — as every blocked record does. Nothing - // re-offers one on a timer. + // comes round on the blocked schedule until the file is back, + // with no window (NextBlockedRetry), and a person may run + // `basecamp connect redispatch ` sooner. return v.end(StateBlocked, ReasonConfigUnreadable), nil } // Every other gate discard turns on the trust set, the matrix or the diff --git a/internal/connector/blocked_retry_test.go b/internal/connector/blocked_retry_test.go new file mode 100644 index 000000000..117ca2bab --- /dev/null +++ b/internal/connector/blocked_retry_test.go @@ -0,0 +1,298 @@ +package connector + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" + + "github.com/basecamp/basecamp-cli/internal/connector/admission" +) + +// blockRecord records a pointer in bucket and blocks it for reason, at the +// ledger's current clock. +func blockRecord(t *testing.T, ledger *Ledger, id, bucket int64, reason admission.Reason) { + t.Helper() + ctx := context.Background() + ev := testEvent(id) + ev.BucketID = bucket + _, err := ledger.RecordSeen(ctx, ev, LanePoll) + require.NoError(t, err) + _, err = ledger.Admission().Commit(ctx, blockedVerdict(id, getRecord(t, ledger, id).Revision, reason)) + require.NoError(t, err) + require.Equal(t, StateBlocked, getRecord(t, ledger, id).State) +} + +// reblock is the verdict a retry writes when the record blocks again: it +// bumps the revision and moves decided_at to the ledger's clock, leaving +// blocked_at where it was. +func reblock(t *testing.T, ledger *Ledger, id int64, reason admission.Reason) { + t.Helper() + _, err := ledger.Admission().Commit(context.Background(), + blockedVerdict(id, getRecord(t, ledger, id).Revision, reason)) + require.NoError(t, err) +} + +// dueIDs is the ids the query calls due, in the order it returned them. +func dueIDs(t *testing.T, ledger *Ledger, scope BlockedRetryScope) []int64 { + t.Helper() + records, err := ledger.DueBlockedRetries(context.Background(), scope) + require.NoError(t, err) + ids := make([]int64, 0, len(records)) + for _, record := range records { + ids = append(ids, record.ID) + } + return ids +} + +// retryLedger is a ledger on a clock the test moves. +func retryLedger(t *testing.T) (*Ledger, *obClock) { + t.Helper() + ledger := newTestLedger(t) + clock := &obClock{now: time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC)} + ledger.now = clock.Now + return ledger, clock +} + +// The requirement the previous attempt was taken out over: out_of_scope is a +// terminal discard, and the projects a run leaves out are another run's to +// dispatch. A sweep that offered one would cause the permanent loss it exists +// to prevent. +func TestDueBlockedRetriesKeepsToTheRunsProjectScope(t *testing.T) { + ledger, clock := retryLedger(t) + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + blockRecord(t, ledger, 2, adapterBucketID+1, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{ + Buckets: []int64{adapterBucketID}, Now: clock.Now(), Limit: 10, + }), "a run restricted to one project never offers the other's record") + + assert.Equal(t, []int64{1, 2}, dueIDs(t, ledger, BlockedRetryScope{ + Now: clock.Now(), Limit: 10, + }), "no --project is every project the agent sees, as it is for admission's own gate") + + assert.Equal(t, []int64{1, 2}, dueIDs(t, ledger, BlockedRetryScope{ + Buckets: []int64{adapterBucketID + 1, adapterBucketID, adapterBucketID}, Now: clock.Now(), Limit: 10, + }), "the scope is a set, whatever order or repeats it arrives in") +} + +// Which reasons automatic re-decision is turned on for. The three left out +// wait for a decision, not for time: no_route for the operator serving the +// project, bucket_mismatch and unroutable for facts about the pointer that no +// later read can change. +func TestDueBlockedRetriesRunsOnlyTheReasonsTheScheduleNames(t *testing.T) { + ledger, clock := retryLedger(t) + timed := []admission.Reason{ + admission.ReasonReadFailed, admission.ReasonReadUnresolved, + admission.ReasonDeltaUnverified, admission.ReasonTrustUnverified, + admission.ReasonConfigUnreadable, + } + untimed := []admission.Reason{ + admission.ReasonNoRoute, admission.ReasonBucketMismatch, admission.ReasonUnroutable, + } + want := make([]int64, 0, len(timed)) + id := int64(1) + for _, reason := range timed { + blockRecord(t, ledger, id, adapterBucketID, reason) + want = append(want, id) + id++ + } + for _, reason := range untimed { + blockRecord(t, ledger, id, adapterBucketID, reason) + id++ + } + clock.Advance(admission.BlockedRetryInterval) + + assert.Equal(t, want, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10})) +} + +// The schedule is admission.NextBlockedRetry's, read back off the columns the +// verdict wrote — not a second copy of it in SQL. +func TestDueBlockedRetriesWaitsOutTheIntervalAndTheWindow(t *testing.T) { + t.Run("nothing before the interval is up", func(t *testing.T) { + ledger, clock := retryLedger(t) + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + + clock.Advance(admission.BlockedRetryInterval - time.Minute) + assert.Empty(t, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10})) + clock.Advance(time.Minute) + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10})) + }) + + t.Run("a throttle is never asked before the server's deadline", func(t *testing.T) { + ledger, clock := retryLedger(t) + blockedAt := clock.Now() + ctx := context.Background() + _, err := ledger.RecordSeen(ctx, testEvent(1), LanePoll) + require.NoError(t, err) + v := blockedVerdict(1, 0, admission.ReasonThrottled) + v.RetryAt = blockedAt.Add(45 * time.Minute) + _, err = ledger.Admission().Commit(ctx, v) + require.NoError(t, err) + + clock.Advance(20 * time.Minute) + assert.Empty(t, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "the interval passed, the server's deadline did not") + clock.Advance(25 * time.Minute) + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10})) + }) + + t.Run("a day of a failing server hands the record to a person", func(t *testing.T) { + ledger, clock := retryLedger(t) + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + blockRecord(t, ledger, 2, adapterBucketID, admission.ReasonConfigUnreadable) + + // The schedule has been running all day: the last attempt is on the + // window's edge, and the next one would fall outside it. + clock.Advance(admission.BlockedRetryWindow) + reblock(t, ledger, 1, admission.ReasonReadFailed) + reblock(t, ledger, 2, admission.ReasonConfigUnreadable) + + clock.Advance(admission.BlockedRetryInterval) + assert.Equal(t, []int64{2}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "a day of a failing server is a server that is not coming back; an unreadable connect.json has no window") + + clock.Advance(30 * 24 * time.Hour) + assert.Equal(t, []int64{2}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "a month later it is still asking") + }) + + t.Run("a retry that came due while nothing was running is late, not excused", func(t *testing.T) { + ledger, clock := retryLedger(t) + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + + // The connector was down for a week. The attempt the record never + // got is owed to it: the window bounds how long the schedule keeps + // asking, not how long an answer stays worth having. + clock.Advance(7 * 24 * time.Hour) + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10})) + + reblock(t, ledger, 1, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + assert.Empty(t, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "it has had its catch-up; the window is long past and a person owns it now") + }) +} + +// A row the schedule has finished with stays blocked forever. Read under a +// plain LIMIT it would hold a place in every sweep's window for good, and the +// due rows behind it would never be reached. The query pages past them +// instead. +func TestDueBlockedRetriesIsNotStarvedByTheRecordsItSkips(t *testing.T) { + ledger, clock := retryLedger(t) + finished := int64(dueBlockedBatch + 10) + for id := int64(1); id <= finished; id++ { + blockRecord(t, ledger, id, adapterBucketID, admission.ReasonReadFailed) + } + // A page and more of rows the schedule is done with: blocked a day ago, + // last asked on the window's edge. + clock.Advance(admission.BlockedRetryWindow) + for id := int64(1); id <= finished; id++ { + reblock(t, ledger, id, admission.ReasonReadFailed) + } + blockRecord(t, ledger, finished+1, adapterBucketID, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + + assert.Equal(t, []int64{finished + 1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "the due row is found behind a page of rows the schedule is done with") +} + +// retryIntake is an intake and ledger on one clock the test moves, with +// nothing else offering anything to the queue. +func retryIntake(t *testing.T) (*Intake, *Ledger, *Queue, *obClock) { + t.Helper() + intake, ledger, queue := newTestIntake(t, nil, nil) + clock := &obClock{now: time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC)} + intake.now, ledger.now = clock.Now, clock.Now + return intake, ledger, queue, clock +} + +func takeID(t *testing.T, queue *Queue) int64 { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + id, err := queue.Take(ctx) + require.NoError(t, err) + return id +} + +// The sweep itself: a due blocked record is offered, and offered once. Queue +// .Offer does not deduplicate, so without the claim the same rows would be +// offered again on every tick while the rows behind them in the window +// starve, and a second copy could decide a record the moment the first +// re-blocked it. +func TestTheSweepOffersADueBlockedRecordOncePerRevision(t *testing.T) { + intake, ledger, queue, clock := retryIntake(t) + ctx := context.Background() + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + + intake.sweepBlockedRetries(ctx) + assert.Zero(t, queue.Depth(), "not due yet") + + clock.Advance(admission.BlockedRetryInterval) + intake.sweepBlockedRetries(ctx) + require.Equal(t, 1, queue.Depth()) + assert.Equal(t, int64(1), takeID(t, queue)) + + // The record is still blocked and still due; the claim is what stops it + // being offered a second time. + intake.sweepBlockedRetries(ctx) + clock.Advance(admission.BlockedRetryInterval) + intake.sweepBlockedRetries(ctx) + assert.Zero(t, queue.Depth(), "claimed at this revision, and nothing has decided it since") + + // Admission re-decides it and it blocks again. That bumps the revision + // and moves decided_at, so the claim is retired and the next retry is an + // interval away — never sooner. + _, err := ledger.Admission().Commit(ctx, blockedVerdict(1, getRecord(t, ledger, 1).Revision, admission.ReasonReadFailed)) + require.NoError(t, err) + intake.sweepBlockedRetries(ctx) + assert.Zero(t, queue.Depth(), "re-blocked just now: the interval has not passed") + + clock.Advance(admission.BlockedRetryInterval) + intake.sweepBlockedRetries(ctx) + require.Equal(t, 1, queue.Depth()) + assert.Equal(t, int64(1), takeID(t, queue)) +} + +// The sweep passes the run's scope down, so the query cannot be asked for a +// record this connector must not decide. +func TestTheSweepNeverOffersARecordOutsideTheRunsScope(t *testing.T) { + intake, ledger, queue, clock := retryIntake(t) + intake.opts.Filters = eventfeed.Filters{Buckets: []int64{adapterBucketID}} + blockRecord(t, ledger, 1, adapterBucketID+1, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + + intake.sweepBlockedRetries(context.Background()) + assert.Zero(t, queue.Depth(), "offering it would discard it out_of_scope, terminally") +} + +// The join, not the pieces. The previous attempt's end-to-end test called the +// query and the queue itself, so it stayed green with the periodic hook +// deleted. This one drives the real ticker and nothing else: delete the +// sweepBlockedRetries call from sweepLosses and this fails, while the test +// above it stays green. +func TestTheRepairSweepTickerRunsTheBlockedRetrySchedule(t *testing.T) { + intake, ledger, queue, clock := retryIntake(t) + intake.opts.RepairInterval = time.Hour + intake.repairSweep = 10 * time.Millisecond + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + intake.startRepairWorkers(ctx) + t.Cleanup(func() { + cancel() + intake.repairs.Wait() + intake.releaseRepairWorkers() + }) + + assert.Equal(t, int64(1), takeID(t, queue), + "the timer the connector actually runs is what re-offers a blocked record") +} diff --git a/internal/connector/intake.go b/internal/connector/intake.go index 21708f154..781001f6a 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -178,6 +178,30 @@ type Intake struct { // duplicate, and without this the id would wait for a restart. stranded map[int64]struct{} + // retriedBlocked is the revision each blocked record was last offered a + // timed retry at. It is the claim the sweep takes before it offers: + // Queue.Offer does not deduplicate, so without one the same due rows + // would be offered again on every tick — starving the rows behind them + // in the window, and letting a second copy decide a record the moment + // the first re-blocked it, which is the interval bypassed. + // + // One entry per record rather than one per (id, revision): a verdict + // bumps the revision, so the record's own next decision retires the + // claim, and nothing accumulates per attempt. + // + // In memory is enough. AcquireInstanceLock is a flock the kernel drops + // on process death, so exactly one connector ever sweeps a ledger: there + // is no second writer to race and no stale claim to reap. A crash loses + // the map, which costs at most one extra decision — and admission's + // revision guard already makes a second decision at the same revision + // harmless. + // + // It is never pruned. It holds one small entry per record this process + // has ever retried, which for a connector that runs for months is the + // records that blocked in those months, not the events that passed + // through it. + retriedBlocked map[int64]int64 + repairs sync.WaitGroup repairQueue chan Loss // repairQueueSize and repairSweep override the pool's defaults in tests. @@ -849,6 +873,77 @@ func (in *Intake) sweepStranded(ctx context.Context) { } } +// blockedRetryBatch is how many due blocked records one sweep offers. The +// sweep runs every defaultRepairSweep and the schedule's own interval is ten +// minutes, so this is a ceiling on a burst — an outage ending, or a repaired +// connect.json — not a rate the ordinary case reaches. +const blockedRetryBatch = 100 + +// sweepBlockedRetries offers the blocked records whose retry has come due. +// +// This is the whole of the blocked-record recovery schedule that +// admission.NextBlockedRetry describes: without it that function computes a +// time nothing acts on, and a read that failed during an outage, a throttle, +// or an unreadable connect.json waits for somebody to notice and run +// `basecamp connect redispatch ` on each one. +// +// The scope is the run's --project buckets, passed down rather than assumed. +// A connector restricted to project A that offered a blocked record from +// project B would send it to a terminal discarded(out_of_scope) — the retry +// causing the permanent loss it exists to prevent. +// +// It claims before it offers, and gives the claim back when the offer fails, +// so a record is never left claimed and unoffered. +func (in *Intake) sweepBlockedRetries(ctx context.Context) { + records, err := in.ledger.DueBlockedRetries(ctx, BlockedRetryScope{ + Buckets: in.opts.Filters.Buckets, + Now: in.now(), + Limit: blockedRetryBatch, + }) + if err != nil { + in.log.Warn("could not read the blocked records due for a retry", "error", err) + return + } + for _, record := range records { + release, claimed := in.claimBlockedRetry(record.ID, record.Revision) + if !claimed { + continue + } + if err := in.queue.Offer(ctx, record.ID); err != nil { + release() + in.log.Warn("a blocked record due for a retry could not be offered; it stays for the next sweep", + "event_id", record.ID, "reason", record.Reason, "error", err) + return + } + in.log.Debug("a blocked record was offered for its timed retry", "event_id", record.ID, "reason", record.Reason) + } +} + +// claimBlockedRetry claims one offer of a record at one revision, and returns +// the undo for an offer that then failed. It is false when this revision was +// already offered. +func (in *Intake) claimBlockedRetry(id, revision int64) (release func(), claimed bool) { + in.mu.Lock() + defer in.mu.Unlock() + if in.retriedBlocked == nil { + in.retriedBlocked = map[int64]int64{} + } + previous, held := in.retriedBlocked[id] + if held && previous == revision { + return nil, false + } + in.retriedBlocked[id] = revision + return func() { + in.mu.Lock() + defer in.mu.Unlock() + if held { + in.retriedBlocked[id] = previous + return + } + delete(in.retriedBlocked, id) + }, true +} + // requeueSeen hands every record still in seen to the queue. // // The ledger row is written before the pointer line and before the hand-off, @@ -984,14 +1079,14 @@ func (in *Intake) startRepairWorkers(ctx context.Context) { go in.sweepLosses(ctx) } -// sweepLosses is the periodic repair of both things that can be left behind: -// an event committed but never handed over, and an open loss nothing is -// walking. +// sweepLosses is the periodic repair of the three things that can be left +// behind: an event committed but never handed over, a blocked record whose +// timed retry has come due, and an open loss nothing is walking. // -// One goroutine does both, so a handover that waits at the pause threshold -// also holds up the re-offering of open losses. That is the right way round: -// the backlog is full, the pipeline is stopped, and starting more repair -// walks would only make the backlog worse. +// One goroutine does all three, so a handover that waits at the pause +// threshold also holds up the blocked retries and the re-offering of open +// losses. That is the right way round: the backlog is full, the pipeline is +// stopped, and adding to it would only make the backlog worse. // // The queue is bounded, so an overloaded connector can turn one away; and a // walk can end early, leaving its loss open. Neither may leave a loss with @@ -1012,6 +1107,7 @@ func (in *Intake) sweepLosses(ctx context.Context) { return case <-ticker.C: in.sweepStranded(ctx) + in.sweepBlockedRetries(ctx) losses, err := in.ledger.OpenLosses(ctx) if err != nil { in.log.Warn("could not read the open losses", "error", err) diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go index 1b340899d..4f0469c3a 100644 --- a/internal/connector/ledger.go +++ b/internal/connector/ledger.go @@ -41,7 +41,9 @@ const ( // StateQueued waits behind another event on its conversation. StateQueued RecordState = "queued" // StateBlocked is retained and retried: a reason, never a transport - // failure dressed up as a verdict. + // failure dressed up as a verdict. Retried by the intake sweep, on + // admission.NextBlockedRetry's schedule (Intake.sweepBlockedRetries), and + // by a person at any time with `basecamp connect redispatch `. StateBlocked RecordState = "blocked" // StateDispatched was handed to a worker. StateDispatched RecordState = "dispatched" diff --git a/internal/connector/ledger_decisions.go b/internal/connector/ledger_decisions.go index d008ca56d..cd6361561 100644 --- a/internal/connector/ledger_decisions.go +++ b/internal/connector/ledger_decisions.go @@ -465,9 +465,12 @@ WHERE event_id = ? AND state = 'pending' AND kind IN ('guard_ack', 'holding_repl // AuthorizedBlocked lists blocked records a person authorized, oldest first: // authorized since the record entered its current run of blocked states, so // an authorization that answered an earlier outcome does not count. -// The redispatch command runs the prerequisite itself; this is for the -// blocked-record recovery schedule to run it again when that did not settle -// it (the schedule is plan step 22's, and nothing calls this yet). +// The redispatch command runs the prerequisite itself; this would be for +// running it again when that did not settle it. Nothing calls it. The timed +// retry that now runs (Intake.sweepBlockedRetries) is not it: that schedule +// is keyed on the reason, not on who authorized the record, and the reasons +// it leaves alone — no_route, unroutable, bucket_mismatch — are the ones a +// re-run on a timer could only repeat the same answer for. func (l *Ledger) AuthorizedBlocked(ctx context.Context, limit int) ([]int64, error) { rows, err := l.db.QueryContext(ctx, `SELECT id FROM events WHERE state = 'blocked' AND authorized_at >= blocked_at ORDER BY id LIMIT ?`, limit) if err != nil { diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index fd96d27c7..ca00c91ac 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -11,6 +11,8 @@ import ( "time" "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" + + "github.com/basecamp/basecamp-cli/internal/connector/admission" ) // Record is one row of the events table. @@ -192,6 +194,106 @@ func (l *Ledger) CountInState(ctx context.Context, state RecordState) (int, erro return n, nil } +// BlockedRetryScope narrows DueBlockedRetries to what this run may decide. +type BlockedRetryScope struct { + // Buckets is the run's --project scope — the same slice admission's + // policy gates on, with the same meaning: empty is a run restricted to no + // project, which is every project the agent can see, NOT no project at + // all. + // + // It is not optional, and it is not a detail. out_of_scope is a terminal + // discard, and the projects a run leaves out are another run's to + // dispatch: a sweep that offered a record from outside its scope would + // cause the permanent loss the retry exists to prevent. Every other + // terminal verdict a re-offered record can reach — stale, not_addressed, + // untrusted_performer, agent_authored — is the verdict a fresh event + // would get, and is correct. + Buckets []int64 + // Now is the moment the sweep is asking about. + Now time.Time + // Limit is the most records one sweep takes. + Limit int +} + +// dueBlockedBatch is how many blocked rows one page of the query reads. +const dueBlockedBatch = 200 + +// DueBlockedRetries returns the blocked records whose next retry has come, in +// scope, oldest first, at most Limit of them. +// +// Due is admission.NextBlockedRetry's answer, not this query's: the schedule +// is described in one place, and the SQL only narrows to the rows it could +// possibly say yes about — blocked, in scope, and blocked for a reason the +// schedule names. The rest is paged through rather than cut off with a LIMIT, +// because a row the schedule has finished with (its window passed) stays +// blocked and would otherwise take a place in the window forever, starving +// everything behind it. +func (l *Ledger) DueBlockedRetries(ctx context.Context, scope BlockedRetryScope) ([]Record, error) { + if scope.Limit <= 0 { + return nil, nil + } + reasons := admission.TimedBlockedReasons() + args := make([]any, 0, len(reasons)+len(scope.Buckets)+2) + args = append(args, string(StateBlocked)) + for _, reason := range reasons { + args = append(args, string(reason)) + } + where := ` WHERE state = ? AND reason IN (` + placeholders(len(reasons)) + `)` + if len(scope.Buckets) > 0 { + buckets := slices.Compact(slices.Sorted(slices.Values(scope.Buckets))) + where += ` AND bucket_id IN (` + placeholders(len(buckets)) + `)` + for _, bucket := range buckets { + args = append(args, bucket) + } + } + //nolint:gosec // G202: the clauses are this package's constants and placeholders, never values + query := selectRecords + where + ` AND id > ? ORDER BY id LIMIT ?` + + var ( + due []Record + after int64 + ) + for { + rows, err := l.db.QueryContext(ctx, query, append(slices.Clone(args), after, dueBlockedBatch)...) + if err != nil { + return nil, fmt.Errorf("connector: list blocked records due for a retry: %w", err) + } + records, err := scanRecords(rows) + if err != nil { + return nil, err + } + for _, record := range records { + after = record.ID + if !blockedRetryDue(record, scope.Now) { + continue + } + due = append(due, record) + if len(due) == scope.Limit { + return due, nil + } + } + if len(records) < dueBlockedBatch { + return due, nil + } + } +} + +// blockedRetryDue reports whether record's next scheduled retry is at or +// before now. A record with no blocked_at or decided_at is not one this +// schedule can date, so it waits for a person. +func blockedRetryDue(record Record, now time.Time) bool { + d := record.Decision + if d.BlockedAt == nil || d.DecidedAt == nil { + return false + } + var notBefore time.Time + if d.RetryAt != nil { + notBefore = *d.RetryAt + } + next, ok := admission.NextBlockedRetry(admission.Reason(record.Reason), *d.BlockedAt, *d.DecidedAt, notBefore) + return ok && !next.After(now) +} + // lifecycle is the ledger's state machine: for each state, the states a record // may move to from it. // @@ -202,7 +304,8 @@ func (l *Ledger) CountInState(ctx context.Context, state RecordState) (int, erro // twice, or where a tombstone whose payload has been dropped is picked up as // work with nothing in it. // -// Blocked is not terminal on purpose: it is retained and retried, so it has +// Blocked is not terminal on purpose: it is retained and retried — on the +// timer DueBlockedRetries feeds, and on a person's redispatch — so it has // edges back into the working states. Completed and discarded have none. var lifecycle = map[RecordState][]RecordState{ // seen to queued is one edge, not two: admission commits an admitted diff --git a/internal/connector/retraction_test.go b/internal/connector/retraction_test.go index eb63b40e9..05bc03db4 100644 --- a/internal/connector/retraction_test.go +++ b/internal/connector/retraction_test.go @@ -286,7 +286,8 @@ func TestAskStillOpenReadsWhatTheRecordIsWaitingFor(t *testing.T) { require.True(t, open(t, ctx, ledger, holding, 1)) // The person ran the redispatch and the rerun blocked on something - // that retries by itself (admission.NextBlockedRetry): nobody is + // that retries by itself — admission.NextBlockedRetry's schedule, + // which the intake sweep runs (Intake.sweepBlockedRetries): nobody is // being asked for a redispatch any more. v := obNoRouteVerdict(1, getRecord(t, ledger, 1).Revision, obCommentReply) v.Reason = admission.ReasonReadFailed From 006efc4e08856f0d9b34e0a4e61de9512518ddc7 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 09:33:42 +0200 Subject: [PATCH 2/5] The retry schedule is written on the row, per reason MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot on #770, three findings, each proven with a test that failed first. A reason that changes gets its own window. blocked_at survives every blocked-to-blocked verdict, which is right for the person's authorization that reads it and wrong for a retry window: the reasons carry different ones. A record that spent a week as the unbounded config_unreadable and then blocked read_failed was measured against the week-old stamp and got none of the twenty-four hours read_failed promises — zero attempts, not fewer. That is the retry stranding the work it exists to recover. events.retry_since is when the record entered the reason it is blocked on now, and the window counts from it. The schedule is stored rather than recomputed over the whole table. events.next_retry_at is what NextBlockedRetry answered when the verdict was written, NULL when the record is owed no attempt, and indexed. The sweep's query is a comparison against it, so a record the schedule has finished with is not read at all, where before every blocked row in the ledger's history was decoded every minute to be dated and discarded. Claims are bounded by the live schedule. Each sweep drops the claims of records the ledger has nothing left to do about, so what is held is the backlog being worked through rather than one entry per record ever retried — which after a large outage was a copy of every event id it blocked. The card justified the in-memory memo on what a crash costs, which is about durability and says nothing about growth; growth was the actual problem. A Run no longer inherits the previous Run's claims. resetRunState says nothing a previous Run decided may leak into the next, and a claim is such a decision: a Run canceled between the offer and the verdict left the record blocked at the revision it was claimed at, and the claim suppressed its retry for the life of the process. The upgrade puts already-blocked records on the schedule due now, so a ledger written before this carries no record that silently stopped being retried. --- internal/connector/admission/commit.go | 24 +-- internal/connector/admission/commit_test.go | 12 +- internal/connector/blocked_retry_test.go | 170 +++++++++++++++-- internal/connector/intake.go | 68 ++++++- internal/connector/ledger.go | 41 ++++ internal/connector/ledger_events.go | 200 +++++++++++++------- 6 files changed, 396 insertions(+), 119 deletions(-) diff --git a/internal/connector/admission/commit.go b/internal/connector/admission/commit.go index 3830c7acc..e22a422c9 100644 --- a/internal/connector/admission/commit.go +++ b/internal/connector/admission/commit.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "slices" "sync" "time" ) @@ -170,17 +169,6 @@ var timedBlockedReasons = map[Reason]bool{ ReasonConfigUnreadable: false, } -// TimedBlockedReasons is the blocked reasons NextBlockedRetry schedules, -// sorted so a query built from them is the same query every time. -func TimedBlockedReasons() []Reason { - out := make([]Reason, 0, len(timedBlockedReasons)) - for reason := range timedBlockedReasons { - out = append(out, reason) - } - slices.Sort(out) - return out -} - // NextBlockedRetry returns when a blocked record should next be re-run, and // false when it waits for something other than time: the operator serving the // project (no_route), or a person's redispatch once the window has passed. @@ -188,10 +176,18 @@ func TimedBlockedReasons() []Reason { // before it, and a deadline past the window hands the record to redispatch // rather than asking early. // +// since is when the record entered the reason it is blocked on NOW, which is +// not when it entered blocked. The reasons carry different windows, so a +// record that spent a week as the unbounded config_unreadable and then blocks +// read_failed must get read_failed's twenty-four hours from the moment +// read_failed began; measured from the older stamp it would get none, which +// is the retry stranding the work it is there to recover (Copilot on #770). +// The ledger keeps that stamp per reason (events.retry_since). +// // The returned time may be in the past — a connector that was not running // when a retry came due is late, not excused — so a caller asks whether next // is at or before now, never whether it is in the future. -func NextBlockedRetry(reason Reason, blockedAt, lastAttempt, notBefore time.Time) (time.Time, bool) { +func NextBlockedRetry(reason Reason, since, lastAttempt, notBefore time.Time) (time.Time, bool) { bounded, timed := timedBlockedReasons[reason] if !timed { return time.Time{}, false @@ -200,7 +196,7 @@ func NextBlockedRetry(reason Reason, blockedAt, lastAttempt, notBefore time.Time if notBefore.After(next) { next = notBefore } - if bounded && next.After(blockedAt.Add(BlockedRetryWindow)) { + if bounded && next.After(since.Add(BlockedRetryWindow)) { return time.Time{}, false } return next, true diff --git a/internal/connector/admission/commit_test.go b/internal/connector/admission/commit_test.go index ac78b1704..f7028f7ff 100644 --- a/internal/connector/admission/commit_test.go +++ b/internal/connector/admission/commit_test.go @@ -260,10 +260,11 @@ func TestNextBlockedRetry(t *testing.T) { assert.Equal(t, blockedAt.Add(30*24*time.Hour+10*time.Minute), next) } -// The reasons the schedule names are one list, and the ledger's query is -// built from it: a reason added to NextBlockedRetry and not to the list would -// be scheduled and never read. -func TestTimedBlockedReasonsIsEveryReasonTheScheduleRuns(t *testing.T) { +// Which reasons automatic re-decision is turned on for, stated as a set +// rather than read off the implementation: this is the feature, and a reason +// joining or leaving it is a decision somebody makes, not a diff nobody +// notices. +func TestTheScheduleRunsExactlySixReasons(t *testing.T) { every := []Reason{ ReasonInvalidPointer, ReasonNotInMatrix, ReasonAgentAuthored, ReasonDelegated, ReasonOutOfScope, ReasonUntrustedPerformer, ReasonAssignmentNotOperator, @@ -279,11 +280,10 @@ func TestTimedBlockedReasonsIsEveryReasonTheScheduleRuns(t *testing.T) { } } slices.Sort(scheduled) - assert.Equal(t, TimedBlockedReasons(), scheduled) assert.Equal(t, []Reason{ ReasonConfigUnreadable, ReasonDeltaUnverified, ReasonReadFailed, ReasonReadUnresolved, ReasonThrottled, ReasonTrustUnverified, - }, TimedBlockedReasons(), "the six reasons automatic re-decision is turned on for") + }, scheduled, "the six reasons automatic re-decision is turned on for") } type sliceSource struct { diff --git a/internal/connector/blocked_retry_test.go b/internal/connector/blocked_retry_test.go index 117ca2bab..ea4bd8952 100644 --- a/internal/connector/blocked_retry_test.go +++ b/internal/connector/blocked_retry_test.go @@ -2,6 +2,8 @@ package connector import ( "context" + "path/filepath" + "strconv" "testing" "time" @@ -179,27 +181,31 @@ func TestDueBlockedRetriesWaitsOutTheIntervalAndTheWindow(t *testing.T) { }) } -// A row the schedule has finished with stays blocked forever. Read under a -// plain LIMIT it would hold a place in every sweep's window for good, and the -// due rows behind it would never be reached. The query pages past them -// instead. -func TestDueBlockedRetriesIsNotStarvedByTheRecordsItSkips(t *testing.T) { +// A row the schedule has finished with stays blocked forever, and the sweep +// must not pay for it every minute for the life of the connector. It carries +// no next_retry_at once its window has passed, so the query does not read it +// at all — and it is not on the schedule, so it holds no claim either. +func TestARecordTheScheduleIsFinishedWithLeavesTheSweepEntirely(t *testing.T) { ledger, clock := retryLedger(t) - finished := int64(dueBlockedBatch + 10) - for id := int64(1); id <= finished; id++ { - blockRecord(t, ledger, id, adapterBucketID, admission.ReasonReadFailed) - } - // A page and more of rows the schedule is done with: blocked a day ago, - // last asked on the window's edge. - clock.Advance(admission.BlockedRetryWindow) - for id := int64(1); id <= finished; id++ { - reblock(t, ledger, id, admission.ReasonReadFailed) - } - blockRecord(t, ledger, finished+1, adapterBucketID, admission.ReasonReadFailed) - clock.Advance(admission.BlockedRetryInterval) - - assert.Equal(t, []int64{finished + 1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), - "the due row is found behind a page of rows the schedule is done with") + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + blockRecord(t, ledger, 2, adapterBucketID, admission.ReasonReadFailed) + + // One is asked with room left in its window and stays on the schedule. + // The other is asked on the window's edge, so its next attempt would fall + // outside it and there is no next attempt. + clock.Advance(admission.BlockedRetryWindow - 2*admission.BlockedRetryInterval) + reblock(t, ledger, 2, admission.ReasonReadFailed) + clock.Advance(2 * admission.BlockedRetryInterval) + reblock(t, ledger, 1, admission.ReasonReadFailed) + + assert.Nil(t, getRecord(t, ledger, 1).Decision.NextRetryAt, "its window has passed") + assert.NotNil(t, getRecord(t, ledger, 2).Decision.NextRetryAt) + + scope := BlockedRetryScope{Now: clock.Now(), Limit: 10} + assert.Equal(t, []int64{2}, dueIDs(t, ledger, scope)) + scheduled, err := ledger.ScheduledBlockedIDs(context.Background(), scope) + require.NoError(t, err) + assert.Equal(t, []int64{2}, scheduled, "nothing keeps a claim for a record it can never offer") } // retryIntake is an intake and ledger on one clock the test moves, with @@ -296,3 +302,127 @@ func TestTheRepairSweepTickerRunsTheBlockedRetrySchedule(t *testing.T) { assert.Equal(t, int64(1), takeID(t, queue), "the timer the connector actually runs is what re-offers a blocked record") } + +// Copilot on #770, the high finding, and it is reachable. blocked_at is +// preserved across every blocked-to-blocked verdict, so a record that spent +// a week as the unbounded config_unreadable and then blocks on a transient +// read failure is measured against a week-old clock. read_failed promises 24 +// hours of retries and gets none: it is scheduled against a window that +// belongs to the reason it is no longer blocked on. +// +// That is the retry mechanism stranding the work it exists to recover, +// through a different door than the out_of_scope one the card named. +func TestAReasonThatChangesGetsItsOwnWindow(t *testing.T) { + ledger, clock := retryLedger(t) + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonConfigUnreadable) + + // A week of an unreadable connect.json, asked every ten minutes, which + // the unbounded window is there to allow. + clock.Advance(7 * 24 * time.Hour) + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10})) + + // The file is repaired, the record is decided again, and this time the + // recording's read fails. A fresh 24 hours of read_failed retries is what + // that reason promises. + reblock(t, ledger, 1, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "read_failed's window starts when read_failed does, not when the record first blocked") +} + +// resetRunState says nothing a previous Run decided may leak into the next. +// A claim is a decision about a record, and a Run that was canceled between +// the offer and the verdict leaves the record blocked at the revision it was +// claimed at — so a claim that survived the Run would suppress that record's +// retry for the life of the process. +func TestAClaimDoesNotOutliveItsRun(t *testing.T) { + intake, ledger, queue, clock := retryIntake(t) + ctx := context.Background() + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + + intake.sweepBlockedRetries(ctx) + require.Equal(t, 1, queue.Depth()) + assert.Equal(t, int64(1), takeID(t, queue)) + + // The run ended with the id taken from the queue and no verdict written: + // the record is still blocked, still at the revision it was claimed at. + intake.resetRunState() + intake.sweepBlockedRetries(ctx) + require.Equal(t, 1, queue.Depth(), "the next run offers it again; the revision guard makes a duplicate harmless") + assert.Equal(t, int64(1), takeID(t, queue)) +} + +// Copilot on #770: the claim map held one entry per record ever retried and +// dropped none, so a connector that runs for months kept a copy of every +// event id an outage ever blocked. The bound it needs is the live one — a +// claim is worth keeping only while the record it names is still on the +// schedule. +func TestClaimsAreBoundedByTheRecordsStillOnTheSchedule(t *testing.T) { + intake, ledger, queue, clock := retryIntake(t) + ctx := context.Background() + const records = 60 + for id := int64(1); id <= records; id++ { + blockRecord(t, ledger, id, adapterBucketID, admission.ReasonReadFailed) + } + clock.Advance(admission.BlockedRetryInterval) + intake.sweepBlockedRetries(ctx) + require.Equal(t, records, queue.Depth()) + for range records { + takeID(t, queue) + } + assert.Equal(t, records, intake.claimCount(), "every record on the schedule is claimed") + + // Every one of them is decided and leaves blocked. Their claims name + // records no sweep will ever return again. + for id := int64(1); id <= records; id++ { + _, err := ledger.Admission().Commit(ctx, admittedVerdict(id, getRecord(t, ledger, id).Revision, "recording:"+strconv.FormatInt(id, 10))) + require.NoError(t, err) + } + intake.sweepBlockedRetries(ctx) + assert.Zero(t, intake.claimCount(), "a claim outlives neither its record's block nor its window") + + // And it keeps working after the prune: a record that blocks again is + // claimed again. + blockRecord(t, ledger, records+1, adapterBucketID, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + intake.sweepBlockedRetries(ctx) + require.Equal(t, 1, queue.Depth()) + assert.Equal(t, 1, intake.claimCount()) +} + +// migrationsBeforeBlockedRetrySchedule is the last migration a ledger without +// retry_since and next_retry_at had applied. Migration 14 adds them. +const migrationsBeforeBlockedRetrySchedule = 13 + +// A ledger written before the schedule existed carries blocked records whose +// retry nothing ever computed. The upgrade owes them the attempt they were +// always promised, so the backfill puts them on the schedule due now and the +// ordinary interval takes over from their next verdict. A blocked reason the +// schedule does not run is left alone. +func TestTheUpgradePutsAlreadyBlockedRecordsOnTheSchedule(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "state", "connector.db") + old := applyMigrationsThrough(t, path, migrationsBeforeBlockedRetrySchedule) + for _, row := range []struct { + id int64 + reason string + }{{1, string(admission.ReasonReadFailed)}, {2, string(admission.ReasonNoRoute)}} { + _, err := old.ExecContext(ctx, ` +INSERT INTO events (id, state, reason, lane, event_type, kind, action, bucket_id, creator_id, + recording_id, created_at, seen_at, updated_at, decided_at, blocked_at) +VALUES (?, 'blocked', ?, 'poll', 'comment.created', 'comment_created', 'created', ?, ?, ?, + '2026-09-18T12:00:00.000000000Z', '2026-09-18T12:00:00.000000000Z', + '2026-09-18T12:00:00.000000000Z', '2026-09-18T12:00:00.000000000Z', + '2026-09-18T12:00:00.000000000Z')`, + row.id, row.reason, adapterBucketID, adapterOperatorID, 10304028972) + require.NoError(t, err) + } + require.NoError(t, old.Close()) + + ledger := openUpgraded(t, path) + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{ + Now: time.Date(2026, 9, 18, 12, 0, 0, 0, time.UTC), Limit: 10, + }), "read_failed is owed the attempt nothing ever ran; no_route still waits for the operator") + assert.Nil(t, getRecord(t, ledger, 2).Decision.NextRetryAt) +} diff --git a/internal/connector/intake.go b/internal/connector/intake.go index 781001f6a..e178100fc 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -189,17 +189,22 @@ type Intake struct { // bumps the revision, so the record's own next decision retires the // claim, and nothing accumulates per attempt. // + // It is bounded by the live schedule, not by history. Each sweep drops + // the claims of records the ledger no longer has anything to do about — + // decided, or past their window — so what is held is the backlog the + // connector is currently working through. Unpruned it held one entry per + // record ever retried and dropped none, which for a connector that runs + // for months after a large outage is a copy of every event id that + // outage blocked (Copilot on #770). + // // In memory is enough. AcquireInstanceLock is a flock the kernel drops // on process death, so exactly one connector ever sweeps a ledger: there // is no second writer to race and no stale claim to reap. A crash loses // the map, which costs at most one extra decision — and admission's // revision guard already makes a second decision at the same revision - // harmless. - // - // It is never pruned. It holds one small entry per record this process - // has ever retried, which for a connector that runs for months is the - // records that blocked in those months, not the events that passed - // through it. + // harmless. A Run that ends drops it for the same reason + // (resetRunState): a claim is a decision one run took, and the next run + // re-offers rather than inherits it. retriedBlocked map[int64]int64 repairs sync.WaitGroup @@ -391,6 +396,12 @@ func (in *Intake) resetRunState() { in.replaying = false in.enteredByReentry = false in.promotedThisRun = false + // A claim is a decision this run took about a record. A run canceled + // between the offer and the verdict leaves the record blocked at the + // revision it was claimed at, and a claim that outlived the run would + // suppress that record's retry for the life of the process; a duplicate + // offer is what admission's revision guard is for (Copilot on #770). + in.retriedBlocked = nil select { case <-in.reconnect: default: @@ -893,13 +904,24 @@ const blockedRetryBatch = 100 // causing the permanent loss it exists to prevent. // // It claims before it offers, and gives the claim back when the offer fails, -// so a record is never left claimed and unoffered. +// 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. func (in *Intake) sweepBlockedRetries(ctx context.Context) { - records, err := in.ledger.DueBlockedRetries(ctx, BlockedRetryScope{ + scope := BlockedRetryScope{ Buckets: in.opts.Filters.Buckets, Now: in.now(), Limit: blockedRetryBatch, - }) + } + // 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) + } + records, err := in.ledger.DueBlockedRetries(ctx, scope) if err != nil { in.log.Warn("could not read the blocked records due for a retry", "error", err) return @@ -919,6 +941,26 @@ 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) { + 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{}{} + } + for id := range in.retriedBlocked { + if _, ok := live[id]; !ok { + delete(in.retriedBlocked, id) + } + } +} + // claimBlockedRetry claims one offer of a record at one revision, and returns // the undo for an offer that then failed. It is false when this revision was // already offered. @@ -1492,3 +1534,11 @@ func (p *pointerWriter) write(event eventfeed.Event, lane Lane) error { } return nil } + +// claimCount is how many blocked-retry claims are held. Tests read it: an +// unbounded claim set is a leak nothing else would show. +func (in *Intake) claimCount() int { + in.mu.Lock() + defer in.mu.Unlock() + return len(in.retriedBlocked) +} diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go index 4f0469c3a..e7e9d1a95 100644 --- a/internal/connector/ledger.go +++ b/internal/connector/ledger.go @@ -893,8 +893,49 @@ END; // Nothing on disk is touched. A directory a connector ran work in is // still there, still whatever the worker left in it. migrationDropRoutePaths, + + // Migration 14. The blocked-record retry schedule, written onto the row + // the verdict is written on. + // + // retry_since is when the record entered its CURRENT blocked reason, and + // that is the whole reason it is not blocked_at. blocked_at is when the + // record entered its current run of blocked states and survives every + // blocked-to-blocked verdict, which is right for what reads it (a + // person's authorization, AuthorizedBlocked) and wrong for a window: the + // reasons carry different ones. A record that spent a week as the + // unbounded config_unreadable and then blocks read_failed would be + // measured against a week-old clock and get none of the twenty-four + // hours read_failed promises — the retry stranding the work it is there + // to recover (Copilot on #770). + // + // next_retry_at is admission.NextBlockedRetry's answer, stored: the + // moment the record is next owed an attempt, and NULL when it is owed + // none — an untimed reason, a window that has passed, or any state but + // blocked. It is written by the one move that writes the state, so it + // cannot disagree with the row it is on, and the sweep's query is an + // indexed comparison against it rather than a scan that dates every + // blocked row in the ledger from the beginning of its history. + // + // The backfill gives every blocked record on a timed reason one attempt + // now, and the ordinary schedule takes over from its verdict. The reason + // list is written out here rather than shared with the Go one: a + // migration says what was true at its own version, and it must keep + // saying that after the code moves on. + migrationBlockedRetrySchedule, } +// migrationBlockedRetrySchedule is migration 14. +const migrationBlockedRetrySchedule = ` +ALTER TABLE events ADD COLUMN retry_since TEXT; +ALTER TABLE events ADD COLUMN next_retry_at TEXT; +UPDATE events SET retry_since = blocked_at, next_retry_at = decided_at +WHERE state = 'blocked' AND decided_at IS NOT NULL AND reason IN ( + 'read_failed', 'read_unresolved', 'delta_unverified', + 'trust_unverified', 'throttled', 'config_unreadable' +); +CREATE INDEX events_next_retry ON events (state, next_retry_at); +` + // migrationDropRoutePaths is migration 13: the route's path, everywhere the // ledger held it. The index is dropped before its column, which is what // SQLite requires. diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index ca00c91ac..f04dedc4e 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -56,6 +56,15 @@ type Decision struct { BlockedAt *time.Time // RetryAt is a throttled verdict's server deadline; nil otherwise. RetryAt *time.Time + // RetrySince is when the record entered the reason it is blocked on now + // — not when it entered blocked, which is BlockedAt. The retry window is + // the reason's, so it counts from here. + RetrySince *time.Time + // NextRetryAt is when the schedule next owes this record an attempt, as + // admission.NextBlockedRetry answered when the verdict was written. Nil + // when it owes none: an untimed reason, a window that has passed, or any + // state but blocked. + NextRetryAt *time.Time Trigger string Acknowledge bool @@ -215,83 +224,75 @@ type BlockedRetryScope struct { Limit int } -// dueBlockedBatch is how many blocked rows one page of the query reads. -const dueBlockedBatch = 200 - -// DueBlockedRetries returns the blocked records whose next retry has come, in -// scope, oldest first, at most Limit of them. +// DueBlockedRetries returns the blocked records whose next retry has come, +// in scope, oldest first, at most Limit of them. // -// Due is admission.NextBlockedRetry's answer, not this query's: the schedule -// is described in one place, and the SQL only narrows to the rows it could -// possibly say yes about — blocked, in scope, and blocked for a reason the -// schedule names. The rest is paged through rather than cut off with a LIMIT, -// because a row the schedule has finished with (its window passed) stays -// blocked and would otherwise take a place in the window forever, starving -// everything behind it. +// The schedule is not re-derived here. next_retry_at is what +// admission.NextBlockedRetry answered when the verdict was written, so this +// is an indexed comparison against a stored moment rather than a scan that +// dates every blocked row the ledger has ever held. A record the schedule is +// finished with has no next_retry_at and is not read at all (Copilot on +// #770). func (l *Ledger) DueBlockedRetries(ctx context.Context, scope BlockedRetryScope) ([]Record, error) { if scope.Limit <= 0 { return nil, nil } - reasons := admission.TimedBlockedReasons() - args := make([]any, 0, len(reasons)+len(scope.Buckets)+2) - args = append(args, string(StateBlocked)) - for _, reason := range reasons { - args = append(args, string(reason)) - } - where := ` WHERE state = ? AND reason IN (` + placeholders(len(reasons)) + `)` - if len(scope.Buckets) > 0 { - buckets := slices.Compact(slices.Sorted(slices.Values(scope.Buckets))) - where += ` AND bucket_id IN (` + placeholders(len(buckets)) + `)` - for _, bucket := range buckets { - args = append(args, bucket) - } - } + where, args := scheduledBlockedWhere(scope.Buckets) + where += ` AND next_retry_at <= ?` + args = append(args, stamp(scope.Now)) //nolint:gosec // G202: the clauses are this package's constants and placeholders, never values - query := selectRecords + where + ` AND id > ? ORDER BY id LIMIT ?` + rows, err := l.db.QueryContext(ctx, selectRecords+where+` ORDER BY id LIMIT ?`, append(args, scope.Limit)...) + if err != nil { + return nil, fmt.Errorf("connector: list blocked records due for a retry: %w", err) + } + return scanRecords(rows) +} - var ( - due []Record - after int64 - ) - for { - rows, err := l.db.QueryContext(ctx, query, append(slices.Clone(args), after, dueBlockedBatch)...) - if err != nil { - return nil, fmt.Errorf("connector: list blocked records due for a retry: %w", err) - } - records, err := scanRecords(rows) - if err != nil { +// ScheduledBlockedIDs is every record in scope the retry schedule still has +// something to do about, due now or later, oldest first. +// +// 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 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. +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+` ORDER BY id`, args...) + if err != nil { + return nil, fmt.Errorf("connector: list blocked records on the retry schedule: %w", err) + } + defer func() { _ = rows.Close() }() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { return nil, err } - for _, record := range records { - after = record.ID - if !blockedRetryDue(record, scope.Now) { - continue - } - due = append(due, record) - if len(due) == scope.Limit { - return due, nil - } - } - if len(records) < dueBlockedBatch { - return due, nil - } + ids = append(ids, id) } + return ids, rows.Err() } -// blockedRetryDue reports whether record's next scheduled retry is at or -// before now. A record with no blocked_at or decided_at is not one this -// schedule can date, so it waits for a person. -func blockedRetryDue(record Record, now time.Time) bool { - d := record.Decision - if d.BlockedAt == nil || d.DecidedAt == nil { - return false - } - var notBefore time.Time - if d.RetryAt != nil { - notBefore = *d.RetryAt +// scheduledBlockedWhere selects the blocked records the schedule still runs, +// narrowed to buckets. Both queries are built from it so neither can read a +// different set from the other. +func scheduledBlockedWhere(buckets []int64) (string, []any) { + where := ` WHERE state = ? AND next_retry_at IS NOT NULL` + args := []any{string(StateBlocked)} + if len(buckets) > 0 { + scoped := slices.Compact(slices.Sorted(slices.Values(buckets))) + where += ` AND bucket_id IN (` + placeholders(len(scoped)) + `)` + for _, bucket := range scoped { + args = append(args, bucket) + } } - next, ok := admission.NextBlockedRetry(admission.Reason(record.Reason), *d.BlockedAt, *d.DecidedAt, notBefore) - return ok && !next.After(now) + return where, args } // lifecycle is the ledger's state machine: for each state, the states a record @@ -460,18 +461,25 @@ func (l *Ledger) move(ctx context.Context, db dbtx, t transition) (bool, error) // retry_at holds only for the move that set it. And a record moving to // blocked or discarded loses its snapshot: only a record on its way to a // worker carries content. - now := l.timestamp() + at := l.now() + now := stamp(at) var retryAt any if !t.retryAt.IsZero() { retryAt = stamp(t.retryAt) } + retrySince, nextRetry, err := l.blockedSchedule(ctx, db, t, at) + if err != nil { + return false, err + } var query strings.Builder query.WriteString(`UPDATE events SET state = ?, reason = ?, revision = revision + 1, updated_at = CASE WHEN state = ? THEN updated_at ELSE ? END, blocked_at = CASE WHEN ? <> 'blocked' THEN NULL ELSE COALESCE(blocked_at, ?) END, retry_at = ?, + retry_since = ?, + next_retry_at = ?, snapshot = CASE WHEN ? IN ('blocked', 'discarded') THEN NULL ELSE snapshot END`) - args := []any{string(t.state), t.reason, string(t.state), now, string(t.state), now, retryAt, string(t.state)} + args := []any{string(t.state), t.reason, string(t.state), now, string(t.state), now, retryAt, retrySince, nextRetry, string(t.state)} for _, a := range t.set { query.WriteString(", " + a.column + " = ?") args = append(args, a.value) @@ -516,6 +524,55 @@ func (l *Ledger) move(ctx context.Context, db dbtx, t transition) (bool, error) return affected > 0, nil } +// blockedSchedule is the retry schedule this move writes: when the record +// entered the reason it will be blocked on, and when it is next owed an +// attempt. Both are nil for any state but blocked, and next is nil for a +// blocked record the schedule does not run — an untimed reason, or a window +// that has passed. +// +// The window start is per REASON, not per block. blocked_at survives every +// blocked-to-blocked verdict, which is right for the person's authorization +// that reads it and wrong here: a record that spent a week as the unbounded +// config_unreadable and then blocks read_failed would be measured against a +// week-old clock and get none of read_failed's twenty-four hours (Copilot on +// #770). +// +// It reads the row it is about to write, which every caller that writes a +// verdict does inside its own transaction. Outside one the read could in +// principle see a row another writer is changing — but AcquireInstanceLock is +// a flock keyed on account and agent, so there is no other writer, and the +// UPDATE's own revision and state guards are what keep the move correct +// either way. Only the schedule stamp could be stale, and only in a case the +// lock does not allow. +func (l *Ledger) blockedSchedule(ctx context.Context, db dbtx, t transition, at time.Time) (since, next any, err error) { + if t.state != StateBlocked { + return nil, nil, nil + } + var ( + wasState, wasReason string + wasSince sql.NullString + ) + switch err := db.QueryRowContext(ctx, + `SELECT state, reason, retry_since FROM events WHERE id = ?`, t.id).Scan(&wasState, &wasReason, &wasSince); { + case errors.Is(err, sql.ErrNoRows): + // No row to move. The UPDATE matches nothing and says so. + return nil, nil, nil + case err != nil: + return nil, nil, fmt.Errorf("connector: read the retry schedule of %d: %w", t.id, err) + } + sinceAt := at + if wasState == string(StateBlocked) && wasReason == t.reason && wasSince.Valid { + if sinceAt, err = parseStamp(wasSince.String); err != nil { + return nil, nil, err + } + } + since = stamp(sinceAt) + if due, ok := admission.NextBlockedRetry(admission.Reason(t.reason), sinceAt, at, t.retryAt); ok { + next = stamp(due) + } + return since, next, nil +} + // heldByWorker is true of a dispatched events row a worker was handed and // has not reported on: a delivery exposed or delivered, on any task, and not // withdrawn because the spawn failed before any worker existed. @@ -610,7 +667,8 @@ const selectRecords = ` SELECT id, state, reason, lane, event_type, kind, action, bucket_id, creator_id, performed_by_id, recording_id, details, actor_type, visible_to_clients, created_at, seen_at, updated_at, content_dropped, revision, decided_at, - blocked_at, retry_at, trigger_name, acknowledge, conversation_key, + blocked_at, retry_at, retry_since, next_retry_at, + trigger_name, acknowledge, conversation_key, reply_kind, reply_recording_id, served, class, recording_url, requester_id, snapshot FROM events` @@ -629,7 +687,8 @@ func scanRecords(rows *sql.Rows) ([]Record, error) { performedBy sql.NullInt64 visibleToClients sql.NullBool decidedAt, blockedAt sql.NullString - retryAt sql.NullString + retryAt, retrySince sql.NullString + nextRetryAt sql.NullString acknowledge, served int snapshot []byte d = &r.Decision @@ -638,7 +697,7 @@ func scanRecords(rows *sql.Rows) ([]Record, error) { &r.Action, &r.BucketID, &r.CreatorID, &performedBy, &r.RecordingID, &details, &r.ActorType, &visibleToClients, &createdAt, &seenAt, &updatedAt, &contentDropped, &r.Revision, &decidedAt, &blockedAt, - &retryAt, &d.Trigger, &acknowledge, &d.ConversationKey, &d.ReplyKind, + &retryAt, &retrySince, &nextRetryAt, &d.Trigger, &acknowledge, &d.ConversationKey, &d.ReplyKind, &d.ReplyRecordingID, &served, &d.Class, &d.RecordingURL, &d.RequesterID, &snapshot); err != nil { return nil, fmt.Errorf("connector: scan event record: %w", err) @@ -674,7 +733,8 @@ func scanRecords(rows *sql.Rows) ([]Record, error) { for _, stamped := range []struct { raw sql.NullString to **time.Time - }{{decidedAt, &d.DecidedAt}, {blockedAt, &d.BlockedAt}, {retryAt, &d.RetryAt}} { + }{{decidedAt, &d.DecidedAt}, {blockedAt, &d.BlockedAt}, {retryAt, &d.RetryAt}, + {retrySince, &d.RetrySince}, {nextRetryAt, &d.NextRetryAt}} { if !stamped.raw.Valid { continue } From 9a2886ed12358ae9a97472f1485bd9a727e3f0f2 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 09:53:05 +0200 Subject: [PATCH 3/5] Nothing asks before a deadline a server named MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot on #770, the migration finding and the one it found in code the first round did not change. Both proven with a test that failed first. The backfill honoured no throttle. A migration runs once against a real ledger and leaves the rows behind, so a deadline it ignores is wrong from then on and no later fix reaches those rows — and the specific harm is retrying into an active throttle, which is how a rate limit becomes a harder one. It now writes MAX(decided_at, retry_at): a row whose server deadline is still ahead waits for it, a spent deadline is due like any other, and a row with no retry_at at all is due at the next sweep, which is the attempt nothing had ever computed for it. The migration is not reversible and does not pretend to be — a newer ledger is refused at open — and it is run-once rather than idempotent, which the runner guarantees; both are now written above it. Nothing decides a blocked record before its own schedule says so, whoever handed the id over. The queue carries an id and not the revision it was claimed at, and `basecamp connect redispatch` runs beside a live connector (openConnectLedger takes the instance lock only for an import), so a person could re-decide a record between the sweep's offer and admission taking it — and admission would then load the newer revision and decide it at once, inside the interval that re-decision had just written. LoadUndecided drops the stale hand-off instead. A person's redispatch is not a timer and does not wait for one: authorizeBlocked marks the record due as it authorizes it, so the rerun the command runs straight afterwards is not refused as early. It writes "due now" rather than clearing the schedule, so a rerun that never happened — the command died between the two — stays on the sweep's list instead of being stranded. --- internal/connector/blocked_retry_test.go | 102 ++++++++++++++++++++ internal/connector/ledger.go | 30 +++++- internal/connector/ledger_admission.go | 12 +++ internal/connector/ledger_admission_test.go | 11 ++- internal/connector/ledger_decisions.go | 11 ++- 5 files changed, 158 insertions(+), 8 deletions(-) diff --git a/internal/connector/blocked_retry_test.go b/internal/connector/blocked_retry_test.go index ea4bd8952..0d23577e9 100644 --- a/internal/connector/blocked_retry_test.go +++ b/internal/connector/blocked_retry_test.go @@ -426,3 +426,105 @@ VALUES (?, 'blocked', ?, 'poll', 'comment.created', 'comment_created', 'created' }), "read_failed is owed the attempt nothing ever ran; no_route still waits for the operator") assert.Nil(t, getRecord(t, ledger, 2).Decision.NextRetryAt) } + +// Copilot on #770: a backfill runs once against a real ledger and leaves +// state behind, so a throttle it ignores is wrong from then on and no later +// fix reaches those rows. throttled means a server told us to wait; retrying +// into an active throttle is how a rate limit becomes a harder one. +func TestTheUpgradeNeverSchedulesAThrottledRowBeforeItsDeadline(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "state", "connector.db") + old := applyMigrationsThrough(t, path, migrationsBeforeBlockedRetrySchedule) + blockedAt := "2026-09-18T12:00:00.000000000Z" + deadline := "2026-09-18T12:45:00.000000000Z" + for _, row := range []struct { + id int64 + reason string + retryAt any + }{{1, string(admission.ReasonThrottled), deadline}, {2, string(admission.ReasonReadFailed), nil}} { + _, err := old.ExecContext(ctx, ` +INSERT INTO events (id, state, reason, lane, event_type, kind, action, bucket_id, creator_id, + recording_id, created_at, seen_at, updated_at, decided_at, blocked_at, retry_at) +VALUES (?, 'blocked', ?, 'poll', 'comment.created', 'comment_created', 'created', ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + row.id, row.reason, adapterBucketID, adapterOperatorID, 10304028972, + blockedAt, blockedAt, blockedAt, blockedAt, blockedAt, row.retryAt) + require.NoError(t, err) + } + require.NoError(t, old.Close()) + + ledger := openUpgraded(t, path) + throttled := getRecord(t, ledger, 1) + require.NotNil(t, throttled.Decision.NextRetryAt) + assert.Equal(t, mustStamp(t, deadline), throttled.Decision.NextRetryAt.UTC(), + "the server named the moment; the upgrade does not ask sooner") + + // At the deadline it is due, and not one tick before it. + justBefore := BlockedRetryScope{Now: mustStamp(t, deadline).Add(-time.Second), Limit: 10} + assert.Equal(t, []int64{2}, dueIDs(t, ledger, justBefore), "only the row with no deadline to respect") + assert.Equal(t, []int64{1, 2}, dueIDs(t, ledger, BlockedRetryScope{Now: mustStamp(t, deadline), Limit: 10})) +} + +func mustStamp(t *testing.T, s string) time.Time { + t.Helper() + at, err := parseStamp(s) + require.NoError(t, err) + return at.UTC() +} + +// Copilot on #770, in code the first round did not change: the queue carries +// an id, not the revision it was claimed at. `basecamp connect redispatch` +// runs beside a live connector (openConnectLedger takes the instance lock +// only for an import), so a person can re-decide the record after the sweep +// offered it and before admission took it. Admission would then load the +// newer revision and decide it at once, inside the interval the re-decision +// just wrote. +// +// Nothing decides a blocked record before its own schedule says so, whoever +// offered it — so the stale hand-off is dropped where it is loaded. +func TestAStaleRetryHandoffIsNotDecidedInsideTheInterval(t *testing.T) { + ledger, clock := retryLedger(t) + ctx := context.Background() + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + clock.Advance(admission.BlockedRetryInterval) + + // The sweep reads it as due and offers the id. + require.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10})) + + // Before admission takes it, something else decides the record and it + // blocks again: a fresh interval starts now. + reblock(t, ledger, 1, admission.ReasonReadFailed) + + _, ok, err := ledger.Admission().LoadUndecided(ctx, 1) + require.NoError(t, err) + assert.False(t, ok, "its next attempt is ten minutes away; this hand-off is stale") + + clock.Advance(admission.BlockedRetryInterval) + _, ok, err = ledger.Admission().LoadUndecided(ctx, 1) + require.NoError(t, err) + assert.True(t, ok, "and at the moment the schedule names, it loads") +} + +// A person's redispatch is not a timer, and does not wait for one. It marks +// the record due, so the rerun it asks for runs at once — and so a rerun that +// never happened (the command died between the authorization and the +// admission run) is picked up by the next sweep rather than stranded. +func TestARedispatchMakesABlockedRecordDueAtOnce(t *testing.T) { + ledger, clock := retryLedger(t) + ctx := context.Background() + ledger.SetHooks(LifecycleHooks(ledger, LifecycleOptions{})) + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + + _, ok, err := ledger.Admission().LoadUndecided(ctx, 1) + require.NoError(t, err) + require.False(t, ok, "nine minutes early, on the timer's account") + + out, err := ledger.Redispatch(ctx, 1, "jorge", []int64{adapterBucketID}) + require.NoError(t, err) + require.True(t, out.Rerun) + + _, ok, err = ledger.Admission().LoadUndecided(ctx, 1) + require.NoError(t, err) + assert.True(t, ok, "a person asked for it now") + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "and if the rerun never happens, the sweep picks it up") +} diff --git a/internal/connector/ledger.go b/internal/connector/ledger.go index e7e9d1a95..fcd592181 100644 --- a/internal/connector/ledger.go +++ b/internal/connector/ledger.go @@ -917,10 +917,29 @@ END; // blocked row in the ledger from the beginning of its history. // // The backfill gives every blocked record on a timed reason one attempt - // now, and the ordinary schedule takes over from its verdict. The reason - // list is written out here rather than shared with the Go one: a - // migration says what was true at its own version, and it must keep - // saying that after the code moves on. + // now, and the ordinary schedule takes over from its verdict — except + // that it never asks before a deadline a server already named. + // MAX(decided_at, retry_at) is what "one attempt now, and not before the + // throttle is over" means as one expression; a throttled row whose + // retry_at has passed is due like any other, and a row with no retry_at + // at all — every reason but throttled, and any throttled row written + // before the column existed — is due at the next sweep, which is the + // attempt nothing had computed for it. Copying decided_at alone would + // have retried into an active throttle, which is how a rate limit + // becomes a harder one (Copilot on #770). + // + // The reason list is written out here rather than shared with the Go + // one: a migration says what was true at its own version, and it must + // keep saying that after the code moves on. + // + // It is not reversible and does not try to be. A ledger newer than the + // running build is refused at open (ErrLedgerSchema), so a downgrade + // never reads these rows rather than reading them wrongly; the columns + // are additive, so an older build that did open it would ignore them. + // It is run-once rather than idempotent: the migration runner applies it + // inside a transaction only when schema_migrations is short of 14, and + // its ALTER TABLE would refuse a second run outright rather than + // backfill twice. migrationBlockedRetrySchedule, } @@ -928,7 +947,8 @@ END; const migrationBlockedRetrySchedule = ` ALTER TABLE events ADD COLUMN retry_since TEXT; ALTER TABLE events ADD COLUMN next_retry_at TEXT; -UPDATE events SET retry_since = blocked_at, next_retry_at = decided_at +UPDATE events SET retry_since = blocked_at, + next_retry_at = MAX(decided_at, COALESCE(retry_at, decided_at)) WHERE state = 'blocked' AND decided_at IS NOT NULL AND reason IN ( 'read_failed', 'read_unresolved', 'delta_unverified', 'trust_unverified', 'throttled', 'config_unreadable' diff --git a/internal/connector/ledger_admission.go b/internal/connector/ledger_admission.go index 82f7e9746..2bb7fe089 100644 --- a/internal/connector/ledger_admission.go +++ b/internal/connector/ledger_admission.go @@ -42,6 +42,15 @@ var undecided = []RecordState{StateSeen, StateBlocked} // LoadUndecided loads a seen or blocked record as the event admission decides, // with the revision it was loaded at. An unknown id, or a record past // deciding, is not ok and is skipped. +// +// A blocked record is skipped until its own schedule says so, whoever handed +// the id over. The queue carries an id and not the revision it was claimed +// at, and `basecamp connect redispatch` runs beside a live connector, so a +// person can re-decide a record between the sweep's offer and admission +// taking it; without this, admission would load the newer revision and decide +// it at once, inside the interval that re-decision had just written (Copilot +// on #770). A redispatch does not wait on the timer: it marks the record due +// as it authorizes it (authorizeBlocked). func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event, bool, error) { record, ok, err := a.ledger.Get(ctx, id) if err != nil || !ok { @@ -50,6 +59,9 @@ func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event if record.State != StateSeen && record.State != StateBlocked { return admission.Event{}, false, nil } + if next := record.Decision.NextRetryAt; record.State == StateBlocked && next != nil && next.After(a.ledger.now()) { + return admission.Event{}, false, nil + } return admission.Event{ ID: record.ID, EventType: record.EventType, diff --git a/internal/connector/ledger_admission_test.go b/internal/connector/ledger_admission_test.go index aea8d2462..84c60beec 100644 --- a/internal/connector/ledger_admission_test.go +++ b/internal/connector/ledger_admission_test.go @@ -99,9 +99,18 @@ func TestAdmissionLoadsAnUndecidedRecordAtItsRevision(t *testing.T) { }, ev) assert.False(t, ev.SeenAt.IsZero(), "admission dates membership refusals from SeenAt") - // A blocked record is decided again, at the revision its verdict left. + // A blocked record is decided again, at the revision its verdict left — + // once its own schedule says so, and not before, whoever hands the id + // over (LoadUndecided). + at := ledger.now() + ledger.now = func() time.Time { return at } _, err = store.Commit(ctx, blockedVerdict(1, 0, admission.ReasonReadFailed)) require.NoError(t, err) + _, ok, err = store.LoadUndecided(ctx, 1) + require.NoError(t, err) + require.False(t, ok, "its next attempt is ten minutes away") + + ledger.now = func() time.Time { return at.Add(admission.BlockedRetryInterval) } ev, ok, err = store.LoadUndecided(ctx, 1) require.NoError(t, err) require.True(t, ok) diff --git a/internal/connector/ledger_decisions.go b/internal/connector/ledger_decisions.go index cd6361561..eb0003236 100644 --- a/internal/connector/ledger_decisions.go +++ b/internal/connector/ledger_decisions.go @@ -510,10 +510,17 @@ func pendingNote(task eventTask) string { // authorization counts for that block only when it is not older than it // (AuthorizedBlocked), and neither a move's own later stamp nor a clock that // stepped back may make a fresh one look stale. +// It also makes the record due now. A person asking for a rerun is not a +// timer and does not wait for one: the redispatch command runs admission +// itself straight after this, which the blocked schedule would otherwise +// refuse as early (LoadUndecided). Writing "due now" rather than clearing the +// schedule is what keeps a rerun that never happened — the command died +// between the two — on the sweep's list instead of stranding it. func authorizeBlocked(ctx context.Context, tx *sql.Tx, eventID int64, now, by string) error { if _, err := tx.ExecContext(ctx, ` -UPDATE events SET authorized_at = MAX(?, COALESCE(blocked_at, '')), authorized_by = ? -WHERE id = ? AND state = 'blocked'`, now, by, eventID); err != nil { +UPDATE events SET authorized_at = MAX(?, COALESCE(blocked_at, '')), authorized_by = ?, + next_retry_at = CASE WHEN next_retry_at IS NULL THEN NULL ELSE ? END +WHERE id = ? AND state = 'blocked'`, now, by, now, eventID); err != nil { return fmt.Errorf("connector: authorize event %d: %w", eventID, err) } return nil From 8f8ea41f7f1db1d3a4c195fbb64039d22929b208 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 10:05:24 +0200 Subject: [PATCH 4/5] A limit counts work the sweep can do, and nil is not permission MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot on #770, both proven with a test that failed first. The batch counted records read, not records offered. A row already claimed is due and stays due until admission decides it, so a batch filled with claims came back full of the same claims tick after tick and never reached the rows behind — the window starvation the claim exists to prevent, arriving through the limit instead of through the offer. DueBlockedRetries takes an AfterID and returns a page; the sweep pages, filters its own claims out of each page, and stops when it has offered a batch. The claim set lives in memory, where the filtering is free, rather than being sent to SQLite as an IN list that after an outage is the whole backlog. The paging walks the due index, not every blocked row. Nil meant "no attempt is owed" and was read as permission to run now. A blocked record with no next_retry_at — an untimed reason, or a window that has passed — was loaded immediately and on every tick after it. LoadUndecided now requires a non-nil and due stamp, and authorizeBlocked stamps one unconditionally rather than preserving nil: without that second half, no_route and every other untimed reason could not be redispatched at all, which is the case a person is most likely to redispatch. The two columns do not mean the same thing by nil, and both now say so where they are read. retry_at is a server's deadline, and nil is the absence of a constraint — the ten-minute interval still governs. next_retry_at is the schedule's answer, and nil is a refusal. The backfill agrees: it never leaves a scheduled record nil. --- internal/connector/blocked_retry_test.go | 54 ++++++++++++++++++++++ internal/connector/intake.go | 59 +++++++++++++++++------- internal/connector/ledger_admission.go | 26 +++++++---- internal/connector/ledger_decisions.go | 16 ++++--- internal/connector/ledger_events.go | 16 +++++-- 5 files changed, 134 insertions(+), 37 deletions(-) diff --git a/internal/connector/blocked_retry_test.go b/internal/connector/blocked_retry_test.go index 0d23577e9..c0413f6fd 100644 --- a/internal/connector/blocked_retry_test.go +++ b/internal/connector/blocked_retry_test.go @@ -528,3 +528,57 @@ func TestARedispatchMakesABlockedRecordDueAtOnce(t *testing.T) { assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), "and if the rerun never happens, the sweep picks it up") } + +// Copilot on #770, and it is the starvation the card named, reintroduced +// through the back door. The claim stops a row being offered twice; it does +// nothing if the claimed rows still spend the sweep's budget. A limit filled +// entirely by rows already claimed offers nothing at all, tick after tick, +// while the rows behind them are never reached. +func TestASweepsLimitCountsWorkItCanActuallyDo(t *testing.T) { + intake, ledger, queue, clock := retryIntake(t) + ctx := context.Background() + intake.retryBatch = 2 + for id := int64(1); id <= 3; id++ { + blockRecord(t, ledger, id, adapterBucketID, admission.ReasonReadFailed) + } + clock.Advance(admission.BlockedRetryInterval) + + intake.sweepBlockedRetries(ctx) + require.Equal(t, 2, queue.Depth(), "the batch is two") + assert.Equal(t, int64(1), takeID(t, queue)) + assert.Equal(t, int64(2), takeID(t, queue)) + + // Admission has not got to them, so 1 and 2 are still blocked, still due + // and still claimed. They are not work this sweep can do, and must not + // spend its budget. + intake.sweepBlockedRetries(ctx) + require.Equal(t, 1, queue.Depth(), "the record behind them is reached") + assert.Equal(t, int64(3), takeID(t, queue)) +} + +// Copilot on #770: nil is documented as "the schedule owes no attempt", and +// the load treated it as permission to run now. Absent information is not +// consent — a blocked record with no next attempt stays blocked until +// something gives it one. +func TestABlockedRecordWithNoScheduledAttemptIsNotRunnable(t *testing.T) { + ledger, clock := retryLedger(t) + ctx := context.Background() + ledger.SetHooks(LifecycleHooks(ledger, LifecycleOptions{})) + // no_route waits for the operator to serve the project: untimed, so the + // verdict writes no next attempt at all. + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonNoRoute) + require.Nil(t, getRecord(t, ledger, 1).Decision.NextRetryAt) + + _, ok, err := ledger.Admission().LoadUndecided(ctx, 1) + require.NoError(t, err) + assert.False(t, ok, "nothing has given this record an attempt to make") + + // A person is what gives an untimed record one. + _, err = ledger.Redispatch(ctx, 1, "jorge", []int64{adapterBucketID}) + require.NoError(t, err) + _, ok, err = ledger.Admission().LoadUndecided(ctx, 1) + require.NoError(t, err) + assert.True(t, ok, "the redispatch is the attempt") + assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), + "and if the rerun never happens, the sweep picks it up rather than stranding it") +} diff --git a/internal/connector/intake.go b/internal/connector/intake.go index e178100fc..ff46c6edf 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -209,9 +209,11 @@ type Intake struct { repairs sync.WaitGroup repairQueue chan Loss - // repairQueueSize and repairSweep override the pool's defaults in tests. + // repairQueueSize, repairSweep and retryBatch override the defaults in + // tests. repairQueueSize int repairSweep time.Duration + retryBatch int // inFlight is the losses a worker is walking or the queue is holding, so // the sweeper does not offer one twice. inFlight map[int64]bool @@ -907,11 +909,23 @@ const blockedRetryBatch = 100 // 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 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 +// counted those would come back full of them tick after tick and never reach +// the rows behind — the window starvation the claim is there to prevent, +// through the limit instead of through the offer (Copilot on #770). So it +// pages, and the claims are filtered out of each page before the budget is +// spent. func (in *Intake) sweepBlockedRetries(ctx context.Context) { + batch := in.retryBatch + if batch <= 0 { + batch = blockedRetryBatch + } scope := BlockedRetryScope{ Buckets: in.opts.Filters.Buckets, Now: in.now(), - Limit: blockedRetryBatch, + Limit: batch, } // 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 @@ -921,23 +935,36 @@ func (in *Intake) sweepBlockedRetries(ctx context.Context) { } else { in.pruneBlockedClaims(scheduled) } - records, err := in.ledger.DueBlockedRetries(ctx, scope) - if err != nil { - in.log.Warn("could not read the blocked records due for a retry", "error", err) - return - } - for _, record := range records { - release, claimed := in.claimBlockedRetry(record.ID, record.Revision) - if !claimed { - continue + for offered := 0; offered < batch; { + records, err := in.ledger.DueBlockedRetries(ctx, scope) + if err != nil { + in.log.Warn("could not read the blocked records due for a retry", "error", err) + return + } + if len(records) == 0 { + return + } + for _, record := range records { + scope.AfterID = record.ID + release, claimed := in.claimBlockedRetry(record.ID, record.Revision) + if !claimed { + continue + } + if err := in.queue.Offer(ctx, record.ID); err != nil { + release() + in.log.Warn("a blocked record due for a retry could not be offered; it stays for the next sweep", + "event_id", record.ID, "reason", record.Reason, "error", err) + return + } + in.log.Debug("a blocked record was offered for its timed retry", "event_id", record.ID, "reason", record.Reason) + if offered++; offered == batch { + return + } } - if err := in.queue.Offer(ctx, record.ID); err != nil { - release() - in.log.Warn("a blocked record due for a retry could not be offered; it stays for the next sweep", - "event_id", record.ID, "reason", record.Reason, "error", err) + if len(records) < scope.Limit { + // The last page: there is nothing behind it to reach. return } - in.log.Debug("a blocked record was offered for its timed retry", "event_id", record.ID, "reason", record.Reason) } } diff --git a/internal/connector/ledger_admission.go b/internal/connector/ledger_admission.go index 2bb7fe089..62f41f0f4 100644 --- a/internal/connector/ledger_admission.go +++ b/internal/connector/ledger_admission.go @@ -43,14 +43,22 @@ var undecided = []RecordState{StateSeen, StateBlocked} // with the revision it was loaded at. An unknown id, or a record past // deciding, is not ok and is skipped. // -// A blocked record is skipped until its own schedule says so, whoever handed -// the id over. The queue carries an id and not the revision it was claimed -// at, and `basecamp connect redispatch` runs beside a live connector, so a -// person can re-decide a record between the sweep's offer and admission -// taking it; without this, admission would load the newer revision and decide -// it at once, inside the interval that re-decision had just written (Copilot -// on #770). A redispatch does not wait on the timer: it marks the record due -// as it authorizes it (authorizeBlocked). +// A blocked record is loaded only when something has given it an attempt to +// make, and only once that attempt is due — whoever handed the id over. The +// queue carries an id and not the revision it was claimed at, and `basecamp +// connect redispatch` runs beside a live connector, so a person can re-decide +// a record between the sweep's offer and admission taking it; without this, +// admission would load the newer revision and decide it at once, inside the +// interval that re-decision had just written (Copilot on #770). +// +// No next_retry_at means no attempt is owed — an untimed reason, or a window +// that has passed — and that is a reason to stay blocked, not permission to +// run. Absent information is not consent: read the other way, a record with +// no schedule would be eligible immediately and on every tick after it. What +// gives an untimed record an attempt is a person: a redispatch stamps +// next_retry_at as it authorizes (authorizeBlocked), so the rerun it asks for +// runs at once and a rerun that never happened is picked up by the sweep +// rather than stranded. func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event, bool, error) { record, ok, err := a.ledger.Get(ctx, id) if err != nil || !ok { @@ -59,7 +67,7 @@ func (a Admission) LoadUndecided(ctx context.Context, id int64) (admission.Event if record.State != StateSeen && record.State != StateBlocked { return admission.Event{}, false, nil } - if next := record.Decision.NextRetryAt; record.State == StateBlocked && next != nil && next.After(a.ledger.now()) { + if next := record.Decision.NextRetryAt; record.State == StateBlocked && (next == nil || next.After(a.ledger.now())) { return admission.Event{}, false, nil } return admission.Event{ diff --git a/internal/connector/ledger_decisions.go b/internal/connector/ledger_decisions.go index eb0003236..9a10d0e20 100644 --- a/internal/connector/ledger_decisions.go +++ b/internal/connector/ledger_decisions.go @@ -510,16 +510,18 @@ func pendingNote(task eventTask) string { // authorization counts for that block only when it is not older than it // (AuthorizedBlocked), and neither a move's own later stamp nor a clock that // stepped back may make a fresh one look stale. -// It also makes the record due now. A person asking for a rerun is not a -// timer and does not wait for one: the redispatch command runs admission -// itself straight after this, which the blocked schedule would otherwise -// refuse as early (LoadUndecided). Writing "due now" rather than clearing the -// schedule is what keeps a rerun that never happened — the command died -// between the two — on the sweep's list instead of stranding it. +// It also makes the record due now, whatever it was blocked on. A person +// asking for a rerun is not a timer and does not wait for one: the redispatch +// command runs admission itself straight after this, and a blocked record +// with no attempt owed it is not one admission will load (LoadUndecided) — +// which is every untimed reason, no_route above all, the one a person is +// most likely to redispatch. Writing "due now" rather than leaving the +// schedule as it was is also what keeps a rerun that never happened — the +// command died between the two — on the sweep's list instead of stranding it. func authorizeBlocked(ctx context.Context, tx *sql.Tx, eventID int64, now, by string) error { if _, err := tx.ExecContext(ctx, ` UPDATE events SET authorized_at = MAX(?, COALESCE(blocked_at, '')), authorized_by = ?, - next_retry_at = CASE WHEN next_retry_at IS NULL THEN NULL ELSE ? END + next_retry_at = ? WHERE id = ? AND state = 'blocked'`, now, by, now, eventID); err != nil { return fmt.Errorf("connector: authorize event %d: %w", eventID, err) } diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index f04dedc4e..3e3caad1a 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -220,12 +220,18 @@ type BlockedRetryScope struct { Buckets []int64 // Now is the moment the sweep is asking about. Now time.Time - // Limit is the most records one sweep takes. + // Limit is the most records one page of the query returns. Limit int + // AfterID pages: only records above it. The caller pages because only it + // knows which rows it has already claimed, and a page that came back full + // of claims would otherwise spend a sweep's whole budget on work nothing + // can do. + AfterID int64 } -// DueBlockedRetries returns the blocked records whose next retry has come, -// in scope, oldest first, at most Limit of them. +// DueBlockedRetries returns one page of the blocked records whose next retry +// has come, in scope, oldest first, at most Limit of them and all above +// AfterID. // // The schedule is not re-derived here. next_retry_at is what // admission.NextBlockedRetry answered when the verdict was written, so this @@ -238,8 +244,8 @@ func (l *Ledger) DueBlockedRetries(ctx context.Context, scope BlockedRetryScope) return nil, nil } where, args := scheduledBlockedWhere(scope.Buckets) - where += ` AND next_retry_at <= ?` - args = append(args, stamp(scope.Now)) + where += ` AND next_retry_at <= ? AND id > ?` + args = append(args, stamp(scope.Now), scope.AfterID) //nolint:gosec // G202: the clauses are this package's constants and placeholders, never values rows, err := l.db.QueryContext(ctx, selectRecords+where+` ORDER BY id LIMIT ?`, append(args, scope.Limit)...) if err != nil { From 6cd607f7f718fbb1d4b39f0cdecbb1b68a5c5724 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Sat, 19 Sep 2026 10:24:29 +0200 Subject: [PATCH 5/5] The due-time index the queries were not using MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot on #770, from the Previously missed section of a review whose header said Findings: None. EXPLAIN QUERY PLAN, before: DueBlockedRetries SEARCH events USING INDEX events_state_id (state=? AND id>?) ScheduledBlockedIDs SEARCH events USING INDEX events_state_id (state=?) state=? AND id>? is every blocked row the ledger has ever retained, walked once a minute, with next_retry_at applied afterwards — the scan the stored due time was added to remove, happening anyway. The index was there, the schema was right, the tests were green, and nothing said the plan ignored all of it. After: DueBlockedRetries SEARCH events USING INDEX events_next_retry (state=? AND next_retry_at>? AND next_retry_at?) No schema change was needed. events.id is INTEGER PRIMARY KEY, so it is the rowid and every events_next_retry entry is already (state, next_retry_at, id): ordering by (next_retry_at, id) is the index's own order and sorts nothing. The due page takes a row-value cursor and the sweep carries both of its values; the prune query takes the same ordering and its doc no longer promises an id order its caller does not use. Records now come back longest-overdue first, which is the better service order as well as the indexed one. The assertion is the plan. TestTheSweepsQueriesUseTheDueTimeIndex runs EXPLAIN QUERY PLAN over the SQL both methods build: the due-time index present, the state/id index absent, nothing sorted. A timing test could not tell the two apart — at test row counts a full scan is instant — so it would have passed over the thing it claimed to check, which is this defect in test form. timedBlockedReasons said the ledger's query narrows by reason. It never has: the reason reaches the sweep once, through the next_retry_at that blockedSchedule persists. The comment now says that, and why a second filter would be a copy of the schedule that could fall out of step with it. --- internal/connector/admission/commit.go | 9 ++-- internal/connector/blocked_retry_test.go | 60 +++++++++++++++++++++++- internal/connector/intake.go | 2 +- internal/connector/ledger_events.go | 50 +++++++++++++++----- 4 files changed, 104 insertions(+), 17 deletions(-) diff --git a/internal/connector/admission/commit.go b/internal/connector/admission/commit.go index e22a422c9..95fab4a86 100644 --- a/internal/connector/admission/commit.go +++ b/internal/connector/admission/commit.go @@ -157,9 +157,12 @@ const ( // timedBlockedReasons is every blocked reason the schedule re-runs, and for // each whether its retries stop at BlockedRetryWindow. // -// One list, read by both halves of the join: NextBlockedRetry decides whether -// a record is due, and the ledger's query narrows to these reasons so the -// sweep never reads the rows that wait for a person. +// NextBlockedRetry is its only reader, and that is the whole of how a reason +// reaches the sweep: the answer is computed once, when the verdict is +// written, and stored on the row (events.next_retry_at). The ledger's queries +// never see a reason — they compare that stored moment. A second filter by +// reason down there could only ever repeat this one, and would be a copy of +// the schedule that could fall out of step with it (Copilot on #770). var timedBlockedReasons = map[Reason]bool{ ReasonReadFailed: true, ReasonReadUnresolved: true, diff --git a/internal/connector/blocked_retry_test.go b/internal/connector/blocked_retry_test.go index c0413f6fd..cf3a262e4 100644 --- a/internal/connector/blocked_retry_test.go +++ b/internal/connector/blocked_retry_test.go @@ -4,6 +4,7 @@ import ( "context" "path/filepath" "strconv" + "strings" "testing" "time" @@ -461,7 +462,9 @@ VALUES (?, 'blocked', ?, 'poll', 'comment.created', 'comment_created', 'created' // At the deadline it is due, and not one tick before it. justBefore := BlockedRetryScope{Now: mustStamp(t, deadline).Add(-time.Second), Limit: 10} assert.Equal(t, []int64{2}, dueIDs(t, ledger, justBefore), "only the row with no deadline to respect") - assert.Equal(t, []int64{1, 2}, dueIDs(t, ledger, BlockedRetryScope{Now: mustStamp(t, deadline), Limit: 10})) + // The longest overdue first, which is the schedule's order and not the + // id's: 2 has been due since noon, 1 only since its deadline passed. + assert.Equal(t, []int64{2, 1}, dueIDs(t, ledger, BlockedRetryScope{Now: mustStamp(t, deadline), Limit: 10})) } func mustStamp(t *testing.T, s string) time.Time { @@ -582,3 +585,58 @@ func TestABlockedRecordWithNoScheduledAttemptIsNotRunnable(t *testing.T) { assert.Equal(t, []int64{1}, dueIDs(t, ledger, BlockedRetryScope{Now: clock.Now(), Limit: 10}), "and if the rerun never happens, the sweep picks it up rather than stranding it") } + +// queryPlan is what SQLite says it will actually do, which is the only +// evidence that distinguishes an index being present from an index being +// used. +func queryPlan(t *testing.T, ledger *Ledger, query string, args ...any) string { + t.Helper() + rows, err := ledger.db.QueryContext(context.Background(), "EXPLAIN QUERY PLAN "+query, args...) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + var plan []string + for rows.Next() { + var id, parent, notUsed int + var detail string + require.NoError(t, rows.Scan(&id, &parent, ¬Used, &detail)) + plan = append(plan, detail) + } + require.NoError(t, rows.Err()) + return strings.Join(plan, "\n") +} + +// Copilot on #770, and the night's defect in a new place: the index was +// there, the schema was right, the tests were green, and the queries used +// neither. Ordered by id, SQLite plans both against events_state_id and walks +// every blocked row the ledger has ever retained — the scan the stored due +// time exists to remove, happening anyway with nothing saying so. +// +// A timing test cannot tell these apart: at the row counts a test uses, a +// full scan is instant. The plan is the assertion. +func TestTheSweepsQueriesUseTheDueTimeIndex(t *testing.T) { + ledger, clock := retryLedger(t) + ctx := context.Background() + blockRecord(t, ledger, 1, adapterBucketID, admission.ReasonReadFailed) + scope := BlockedRetryScope{Now: clock.Now(), Limit: 10} + + // The queries the sweep actually runs, planned as they are built. The + // ledger runs them through these two methods and nothing else, so the + // assertion is on the same SQL the connector executes. + _, err := ledger.DueBlockedRetries(ctx, scope) + require.NoError(t, err) + _, err = ledger.ScheduledBlockedIDs(ctx, scope) + require.NoError(t, err) + + where, args := scheduledBlockedWhere(nil) + due := queryPlan(t, ledger, + selectRecords+where+` AND next_retry_at <= ? AND (next_retry_at, id) > (?, ?)`+blockedRetryOrder+` LIMIT ?`, + append(append([]any{}, args...), stamp(scope.Now), stamp(time.Time{}), int64(0), scope.Limit)...) + assert.Contains(t, due, "events_next_retry", "the due page is found through the due-time index") + 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") +} diff --git a/internal/connector/intake.go b/internal/connector/intake.go index ff46c6edf..63a15b445 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -945,7 +945,7 @@ func (in *Intake) sweepBlockedRetries(ctx context.Context) { return } for _, record := range records { - scope.AfterID = record.ID + scope.AfterRetryAt, scope.AfterID = *record.Decision.NextRetryAt, record.ID release, claimed := in.claimBlockedRetry(record.ID, record.Revision) if !claimed { continue diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index 3e3caad1a..2a4e40176 100644 --- a/internal/connector/ledger_events.go +++ b/internal/connector/ledger_events.go @@ -222,16 +222,22 @@ type BlockedRetryScope struct { Now time.Time // Limit is the most records one page of the query returns. Limit int - // AfterID pages: only records above it. The caller pages because only it - // knows which rows it has already claimed, and a page that came back full - // of claims would otherwise spend a sweep's whole budget on work nothing - // can do. - AfterID int64 + // AfterRetryAt and AfterID are one cursor, in the order the query returns + // rows: the next page is what sorts after them. The caller pages because + // only it knows which rows it has already claimed, and a page that came + // back full of claims would otherwise spend a sweep's whole budget on + // work nothing can do. + // + // The cursor is the due time first and the id second because that is the + // order events_next_retry holds, and paging in any other order is what + // makes the query ignore it. Their zero values are before every row. + AfterRetryAt time.Time + AfterID int64 } // DueBlockedRetries returns one page of the blocked records whose next retry -// has come, in scope, oldest first, at most Limit of them and all above -// AfterID. +// has come, in scope, the longest overdue first, at most Limit of them and +// all sorting after the cursor. // // The schedule is not re-derived here. next_retry_at is what // admission.NextBlockedRetry answered when the verdict was written, so this @@ -239,23 +245,38 @@ type BlockedRetryScope struct { // dates every blocked row the ledger has ever held. A record the schedule is // finished with has no next_retry_at and is not read at all (Copilot on // #770). +// +// That only holds if the query is ordered the way events_next_retry is. +// Ordered by id instead, SQLite planned it against events_state_id and +// walked every blocked row the ledger has ever held, filtering next_retry_at +// after the fact — the index present, the schema right, the tests green, and +// the scan the stored due time exists to remove happening anyway (Copilot on +// #770). The plan is the thing that has to be checked, and +// TestTheSweepsQueriesUseTheDueTimeIndex checks it. func (l *Ledger) DueBlockedRetries(ctx context.Context, scope BlockedRetryScope) ([]Record, error) { if scope.Limit <= 0 { return nil, nil } where, args := scheduledBlockedWhere(scope.Buckets) - where += ` AND next_retry_at <= ? AND id > ?` - args = append(args, stamp(scope.Now), scope.AfterID) + where += ` AND next_retry_at <= ? AND (next_retry_at, id) > (?, ?)` + args = append(args, stamp(scope.Now), stamp(scope.AfterRetryAt), scope.AfterID) //nolint:gosec // G202: the clauses are this package's constants and placeholders, never values - rows, err := l.db.QueryContext(ctx, selectRecords+where+` ORDER BY id LIMIT ?`, append(args, scope.Limit)...) + rows, err := l.db.QueryContext(ctx, selectRecords+where+blockedRetryOrder+` LIMIT ?`, append(args, scope.Limit)...) if err != nil { return nil, fmt.Errorf("connector: list blocked records due for a retry: %w", err) } return scanRecords(rows) } +// blockedRetryOrder is the order events_next_retry holds, and the only order +// either query may ask for: id is the rowid, so an index entry is +// (state, next_retry_at, id) and this needs no sort at all. Asking for any +// other order sends both queries to events_state_id and the whole blocked +// history. +const blockedRetryOrder = ` ORDER BY next_retry_at, id` + // ScheduledBlockedIDs is every record in scope the retry schedule still has -// something to do about, due now or later, oldest first. +// 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 @@ -266,10 +287,15 @@ func (l *Ledger) DueBlockedRetries(ctx context.Context, scope BlockedRetryScope) // 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. +// +// 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+` ORDER BY id`, args...) + 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) }