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..95fab4a86 100644 --- a/internal/connector/admission/commit.go +++ b/internal/connector/admission/commit.go @@ -135,33 +135,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. +// +// 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, + ReasonDeltaUnverified: true, + ReasonTrustUnverified: true, + ReasonThrottled: true, + ReasonConfigUnreadable: false, +} + // 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. -func NextBlockedRetry(reason Reason, blockedAt, lastAttempt, notBefore time.Time) (time.Time, bool) { - switch reason { - case ReasonReadFailed, ReasonReadUnresolved, ReasonDeltaUnverified, ReasonTrustUnverified, ReasonThrottled: - default: +// +// 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, since, lastAttempt, notBefore time.Time) (time.Time, bool) { + 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(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 9473b7e7b..f7028f7ff 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) +} + +// 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, + 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, []Reason{ + ReasonConfigUnreadable, ReasonDeltaUnverified, ReasonReadFailed, + ReasonReadUnresolved, ReasonThrottled, ReasonTrustUnverified, + }, scheduled, "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..cf3a262e4 --- /dev/null +++ b/internal/connector/blocked_retry_test.go @@ -0,0 +1,642 @@ +package connector + +import ( + "context" + "path/filepath" + "strconv" + "strings" + "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, 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) + 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 +// 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") +} + +// 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) +} + +// 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") + // 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 { + 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") +} + +// 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") +} + +// 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 21708f154..63a15b445 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -178,11 +178,42 @@ 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. + // + // 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. 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 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 @@ -367,6 +398,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: @@ -849,6 +886,133 @@ 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. 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: 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 + // 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) + } + 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.AfterRetryAt, scope.AfterID = *record.Decision.NextRetryAt, 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 len(records) < scope.Limit { + // The last page: there is nothing behind it to reach. + return + } + } +} + +// 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. +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 +1148,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 +1176,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) @@ -1396,3 +1561,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 1b340899d..fcd592181 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" @@ -891,8 +893,69 @@ 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 — 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, } +// 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 = 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' +); +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_admission.go b/internal/connector/ledger_admission.go index 82f7e9746..62f41f0f4 100644 --- a/internal/connector/ledger_admission.go +++ b/internal/connector/ledger_admission.go @@ -42,6 +42,23 @@ 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 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 { @@ -50,6 +67,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 d008ca56d..9a10d0e20 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 { @@ -507,10 +510,19 @@ 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, 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 = ? -WHERE id = ? AND state = 'blocked'`, now, by, eventID); err != nil { +UPDATE events SET authorized_at = MAX(?, COALESCE(blocked_at, '')), authorized_by = ?, + next_retry_at = ? +WHERE id = ? AND state = 'blocked'`, now, by, now, eventID); err != nil { return fmt.Errorf("connector: authorize event %d: %w", eventID, err) } return nil diff --git a/internal/connector/ledger_events.go b/internal/connector/ledger_events.go index fd96d27c7..2a4e40176 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. @@ -54,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 @@ -192,6 +203,130 @@ 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 page of the query returns. + Limit int + // 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, 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 +// 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). +// +// 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 (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+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. +// +// 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. +// +// It comes back in the schedule's order rather than by id, for the reason +// DueBlockedRetries does: asked for id order, SQLite reads every blocked row +// the ledger holds instead of only the scheduled ones. The caller reads it as +// a set, so the order is the index's to choose. +func (l *Ledger) ScheduledBlockedIDs(ctx context.Context, scope BlockedRetryScope) ([]int64, error) { + where, args := scheduledBlockedWhere(scope.Buckets) + //nolint:gosec // G202: the clauses are this package's constants and placeholders, never values + rows, err := l.db.QueryContext(ctx, `SELECT id FROM events`+where+blockedRetryOrder, args...) + if err != nil { + return nil, fmt.Errorf("connector: list blocked records on the retry schedule: %w", err) + } + defer func() { _ = rows.Close() }() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// 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) + } + } + return where, args +} + // lifecycle is the ledger's state machine: for each state, the states a record // may move to from it. // @@ -202,7 +337,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 @@ -357,18 +493,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) @@ -413,6 +556,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. @@ -507,7 +699,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` @@ -526,7 +719,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 @@ -535,7 +729,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) @@ -571,7 +765,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 } 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