From 3462e57a3baa93b555d71690440d10d736f5f2ce Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 6 Sep 2026 05:06:33 +0000 Subject: [PATCH] fix(distributed): back a failed claim off instead of respinning it The claim queue's attempts counter grew without bound and nothing read it. At the default two-second poll a permanently undispatchable row cost about 43000 UPDATEs a day, and it cost more than writes: rows are claimed oldest first, so the oldest stuck row was re-claimed ahead of every newer one on every tick and held a dispatch slot while it failed. One poison row starved the queue behind it. No dead letter, and that is the decision rather than the omission. Read settleClaim: the only outcome that releases a claim is one where NOTHING was learned about the work. No agent worker was connected, the tunnel broke, a peer could not be reached, the stream was refused before the request body left this replica. Not one of those is a worker saying it ran the job and it failed, and an attempt ceiling would turn "the fleet was away long enough" into a job failure nobody reported, which is the collapse this whole design exists to prevent pointed at work instead of at nodes. The one verdict available here, that no build of any worker serves this kind, is already settled as an answer. So the retry stays unbounded and the RATE does not. Each release stamps the row with the earliest it may be claimed again, doubling from two seconds to a cap of sixty, computed in the release statement from the row's own attempts count and stamped on the DATABASE clock, because that is the clock competing replicas order the queue on. Queued work becomes claimable again within one cap of the fleet returning, and a stuck row no longer holds the head of the queue. A claim released by the reap carries no delay at all: that work was never handed to anyone, so there is nothing to back off from. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/services/jobs/claim.go | 83 +++++++++++-- core/services/jobs/claim_test.go | 145 ++++++++++++++++++++++ core/services/jobs/dispatch_loop.go | 15 ++- docs/content/features/distributed-mode.md | 6 + 4 files changed, 237 insertions(+), 12 deletions(-) diff --git a/core/services/jobs/claim.go b/core/services/jobs/claim.go index baa7f85f8..7cbf6bc2b 100644 --- a/core/services/jobs/claim.go +++ b/core/services/jobs/claim.go @@ -46,7 +46,12 @@ type WorkClaim struct { ClaimedBy string `gorm:"size:64;index"` ClaimedAt *time.Time `gorm:"index"` Attempts int - CreatedAt time.Time `gorm:"index"` + // NotBefore is the earliest this row may be claimed again, stamped by + // ReleaseClaim from the DATABASE clock. NULL means "now", which is what + // every freshly enqueued row carries and what a reaped row goes back to. + // See claimBackoff. + NotBefore *time.Time `gorm:"index"` + CreatedAt time.Time `gorm:"index"` } // TableName pins the table this maps onto, because ReapAbandoned and ClaimNext @@ -116,6 +121,7 @@ func EnqueueClaim(ctx context.Context, db *gorm.DB, kind ClaimKind, payload any) "claimed_by": "", "claimed_at": nil, "attempts": 0, + "not_before": nil, "created_at": gorm.Expr("now()"), }).Error; err != nil { return "", fmt.Errorf("enqueueing a %s claim: %w", kind, err) @@ -143,11 +149,12 @@ SET claimed_by = ?, claimed_at = now() WHERE id = ( SELECT id FROM ` + claimsTable + ` WHERE claimed_at IS NULL AND kind IN ? + AND (not_before IS NULL OR not_before <= now()) ORDER BY created_at, id FOR UPDATE SKIP LOCKED LIMIT 1 ) -RETURNING id, kind, payload, claimed_by, claimed_at, attempts, created_at` +RETURNING id, kind, payload, claimed_by, claimed_at, attempts, not_before, created_at` // ClaimNext takes at most one unclaimed row of any kind in kinds, marking it // claimed by owner. It returns ErrNoWork when there is none. @@ -180,19 +187,69 @@ func ClaimNext(ctx context.Context, db *gorm.DB, owner string, kinds []ClaimKind return &claim, nil } -// ReleaseClaim returns a row to the pool, incrementing Attempts. +// The retry schedule a released claim comes back on. +// +// There is no dead letter here, and that is a decision rather than an omission. +// Read settleClaim: the only outcome that reaches ReleaseClaim is one where +// NOTHING was learned about the work. No agent worker was connected, the tunnel +// broke, a peer could not be reached, the stream was refused before the request +// body left this replica. Not one of those is the worker saying it ran the job +// and it failed, and failing a job on any of them would report an absent +// connection as a worker's verdict, which is the one collapse this whole design +// exists to prevent. A deployment whose fleet is down for a day must run its +// queued work when the fleet comes back, not find it failed. +// +// What IS a defect is retrying at the poll interval for ever. At the default +// two seconds a permanently undispatchable row costs ~43000 UPDATEs a day, and +// it costs more than writes: rows are claimed oldest-first, so the oldest +// stuck row is re-claimed ahead of every newer one on every single tick and +// holds a dispatch slot while it fails. One poison row starves the queue behind +// it. +// +// So the retry is unbounded and the RATE is not. Each release stamps the row +// with the earliest it may be claimed again, doubling per attempt from +// claimBackoffBase up to claimBackoffCap. A first failure costs the base delay, +// a fleet that has been down for a minute retries at the cap, and work becomes +// claimable again within one cap of the fleet returning. The exponent is +// clamped so the arithmetic cannot overflow however long a row has been stuck. +const ( + // claimBackoffBase is the delay after the first failed dispatch. One poll + // interval: a row that failed because nothing was connected should not be + // re-tried before the loop would have looked again anyway. + claimBackoffBase = 2 * time.Second + // claimBackoffCap bounds the delay, and with it how long queued work waits + // after a fleet comes back. Kept short for that reason: the point of the + // backoff is to stop a stuck row from spinning, not to give up on it. + claimBackoffCap = 60 * time.Second + // claimBackoffMaxShift clamps the exponent. 2^16 base seconds is already + // far past the cap, so this only keeps power() away from infinity for a row + // that has been retried for months. + claimBackoffMaxShift = 16 +) + +// releaseClaimSQL returns a row to the pool and stamps its next eligibility. +// +// One statement, and the delay computed IN the statement, because it reads the +// row's own attempts count and stamps now() from the DATABASE clock. Competing +// replicas order the queue on that clock; a Go-side deadline would make how +// long a row waits depend on the clock skew of whichever replica happened to +// fail it. +const releaseClaimSQL = `UPDATE ` + claimsTable + ` +SET claimed_by = '', claimed_at = NULL, attempts = attempts + 1, + not_before = now() + make_interval(secs => LEAST(?, ? * power(2, LEAST(attempts, ?)))) +WHERE id = ?` + +// ReleaseClaim returns a row to the pool, incrementing Attempts and stamping +// the backoff described above. // // It is what a TRANSPORT failure does: nothing was learned about the work, so // it must be retried, possibly by another replica against another worker. func ReleaseClaim(ctx context.Context, db *gorm.DB, id string) error { - if db == nil { - return errors.New("releasing a claim: no database handle") + if err := requirePostgres(db, "releasing a claim"); err != nil { + return err } - res := db.WithContext(ctx).Model(&WorkClaim{}).Where("id = ?", id).Updates(map[string]any{ - "claimed_by": "", - "claimed_at": nil, - "attempts": gorm.Expr("attempts + 1"), - }) + res := db.WithContext(ctx).Exec(releaseClaimSQL, + claimBackoffCap.Seconds(), claimBackoffBase.Seconds(), claimBackoffMaxShift, id) if res.Error != nil { return fmt.Errorf("releasing claim %q: %w", id, res.Error) } @@ -227,8 +284,12 @@ func CompleteClaim(ctx context.Context, db *gorm.DB, id string) error { // so this reap and every other reader of replica absence in the deployment move // together, and it runs as ONE statement so a replica cannot die, or come back, // between deciding who is live and acting on it. +// It stamps NO backoff, unlike ReleaseClaim. A claim abandoned because its +// replica died was never dispatched anywhere: there is nothing to back off +// from, and delaying it would punish the work for the death of the process that +// held it. It becomes claimable at once, by whichever replica polls next. const reapAbandonedSQL = `UPDATE ` + claimsTable + ` -SET claimed_by = '', claimed_at = NULL, attempts = attempts + 1 +SET claimed_by = '', claimed_at = NULL, attempts = attempts + 1, not_before = NULL WHERE claimed_at IS NOT NULL AND claimed_by NOT IN (` + cluster.LiveInstanceIDsSQL + `)` diff --git a/core/services/jobs/claim_test.go b/core/services/jobs/claim_test.go index bb931b887..99f8e10af 100644 --- a/core/services/jobs/claim_test.go +++ b/core/services/jobs/claim_test.go @@ -103,6 +103,15 @@ var _ = Describe("The claim queue", func() { return got } + // expireBackoff brings a released row's next-eligible stamp into the past, + // so a spec can assert what happens AFTER the backoff without waiting out a + // real one. Sleeping for the delay would make every one of these specs a + // spec that fails one run in ten. + expireBackoff := func(id string) { + GinkgoHelper() + Expect(db.Exec(`UPDATE `+claimsTable+` SET not_before = now() - interval '1 hour' WHERE id = ?`, id).Error).To(Succeed()) + } + Describe("taking work", func() { It("reports an empty queue distinguishably rather than as a failure", func() { _, err := ClaimNext(ctx, db, "inst-a", []ClaimKind{ClaimKindMCPCI}) @@ -273,6 +282,10 @@ var _ = Describe("The claim queue", func() { Expect(row.ClaimedBy).To(BeEmpty()) Expect(row.Attempts).To(Equal(1)) + // Claimable again, once the backoff this release stamped has + // passed. The wait is brought forward rather than slept through: + // see "backing a failed dispatch off". + expireBackoff(id) again, err := ClaimNext(ctx, db, "inst-b", []ClaimKind{ClaimKindMCPCI}) Expect(err).ToNot(HaveOccurred()) Expect(again.ID).To(Equal(id)) @@ -293,6 +306,112 @@ var _ = Describe("The claim queue", func() { }) }) + Describe("backing a failed dispatch off", func() { + // A release means NOTHING was learned about the work: no worker was + // connected, the tunnel broke, the stream was refused before the + // request left. So the work is retried for ever and is never failed. + // What is bounded is the RATE, because at the poll interval a + // permanently undispatchable row costs an UPDATE every two seconds and, + // worse, is re-claimed ahead of every newer row on every tick. + It("does not offer a just-released row again immediately", func() { + id := enqueue(ClaimKindMCPCI, JobEvent{JobID: "j1"}) + _, err := ClaimNext(ctx, db, "inst-a", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + + Expect(ReleaseClaim(ctx, db, id)).To(Succeed()) + + _, err = ClaimNext(ctx, db, "inst-b", []ClaimKind{ClaimKindMCPCI}) + Expect(err).To(MatchError(ErrNoWork), + "a row re-claimed on the very next tick spins at the poll interval for as long as the fleet is away") + }) + + It("stamps a longer delay on each successive failure", func() { + id := enqueue(ClaimKindMCPCI, JobEvent{JobID: "j1"}) + + delays := make([]time.Duration, 0, 3) + for i := 0; i < 3; i++ { + expireBackoff(id) + _, err := ClaimNext(ctx, db, "inst-a", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + before := time.Now() + Expect(ReleaseClaim(ctx, db, id)).To(Succeed()) + row := rowOf(id) + Expect(row.NotBefore).ToNot(BeNil()) + delays = append(delays, row.NotBefore.Sub(before)) + } + + Expect(delays[1]).To(BeNumerically(">", delays[0])) + Expect(delays[2]).To(BeNumerically(">", delays[1])) + }) + + It("never delays a retry past the cap, however long the row has been stuck", func() { + // The retry is unbounded and the wait is not: work has to become + // claimable again within one cap of the fleet coming back. + id := enqueue(ClaimKindMCPCI, JobEvent{JobID: "j1"}) + Expect(db.Model(&WorkClaim{}).Where("id = ?", id). + Update("attempts", 100000).Error).To(Succeed()) + + _, err := ClaimNext(ctx, db, "inst-a", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + before := time.Now() + Expect(ReleaseClaim(ctx, db, id)).To(Succeed()) + + row := rowOf(id) + Expect(row.NotBefore).ToNot(BeNil()) + Expect(row.NotBefore.Sub(before)).To(BeNumerically("<=", claimBackoffCap+time.Second)) + }) + + It("keeps a stuck row from starving the newer work behind it", func() { + // Rows are claimed oldest first. Before the backoff, the oldest + // undispatchable row was taken again on every tick and held a + // dispatch slot while it failed, so nothing behind it ever ran. + stuck := enqueue(ClaimKindMCPCI, JobEvent{JobID: "stuck"}) + _, err := ClaimNext(ctx, db, "inst-a", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + Expect(ReleaseClaim(ctx, db, stuck)).To(Succeed()) + + behind := enqueue(ClaimKindMCPCI, JobEvent{JobID: "behind"}) + + got, err := ClaimNext(ctx, db, "inst-b", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + Expect(got.ID).To(Equal(behind)) + }) + + It("retries a stuck row for ever rather than failing work nobody refused", func() { + // The decision this Describe records. A release carries no verdict, + // so there is no attempt count after which the claim is discarded: + // a deployment whose fleet was down for a day runs its queued work + // when the fleet comes back. + id := enqueue(ClaimKindMCPCI, JobEvent{JobID: "j1"}) + for i := 0; i < 25; i++ { + expireBackoff(id) + _, err := ClaimNext(ctx, db, "inst-a", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred(), "the row was discarded after %d attempts", i) + Expect(ReleaseClaim(ctx, db, id)).To(Succeed()) + } + + Expect(rowOf(id).Attempts).To(Equal(25)) + }) + + It("does not delay a claim its replica died holding", func() { + // A reap is not a failed dispatch: the work was never handed to + // anyone, so there is nothing to back off from and delaying it + // would punish the work for the death of the process holding it. + id := enqueue(ClaimKindMCPCI, JobEvent{JobID: "j1"}) + _, err := ClaimNext(ctx, db, "dead-replica", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + + released, err := ReapAbandoned(ctx, db, time.Minute) + Expect(err).ToNot(HaveOccurred()) + Expect(released).To(Equal(int64(1))) + + Expect(rowOf(id).NotBefore).To(BeNil()) + again, err := ClaimNext(ctx, db, "inst-b", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + Expect(again.ID).To(Equal(id)) + }) + }) + Describe("reaping what a departed replica left", func() { // The distinction the whole programme rests on, applied to work: a // claim held by a replica that is GONE becomes claimable again, and a @@ -404,6 +523,24 @@ var _ = Describe("The claim queue", func() { "created_at must not be a literal timestamp from this process's clock") }) + It("releases in ONE statement that stamps the backoff on the database clock", func() { + id := enqueue(ClaimKindMCPCI, JobEvent{JobID: "j1"}) + _, err := ClaimNext(ctx, db, "inst-a", []ClaimKind{ClaimKindMCPCI}) + Expect(err).ToNot(HaveOccurred()) + + rec := newClaimSQLRecorder() + Expect(ReleaseClaim(ctx, db.Session(&gorm.Session{Logger: rec}), id)).To(Succeed()) + + sql := strings.ToLower(rec.only()) + // Read-then-compute-then-update would show up as two statements, + // and would compute the delay from this process's clock. + Expect(sql).To(ContainSubstring("make_interval")) + Expect(sql).To(ContainSubstring("now()")) + Expect(sql).To(ContainSubstring("attempts")) + Expect(sql).ToNot(MatchRegexp(`not_before\s*=\s*'`), + "the next-eligible stamp must not be a literal timestamp from this process's clock") + }) + It("reaps in ONE statement whose predicate is replica liveness on the database clock, not claim age", func() { enqueue(ClaimKindMCPCI, JobEvent{JobID: "j1"}) rec := newClaimSQLRecorder() @@ -452,6 +589,13 @@ var _ = Describe("The claim queue", func() { Expect(err).To(MatchError(ContainSubstring("requires PostgreSQL"))) }) + It("refuses to release", func() { + // make_interval and power() are the backoff's, and they fail on + // SQLite with a parse error that reads like a missing column. + Expect(ReleaseClaim(ctx, lite, "some-claim")).To( + MatchError(ContainSubstring("requires PostgreSQL"))) + }) + It("runs no statement at all when it refuses, so the failure cannot be read as a missing table", func() { rec := newClaimSQLRecorder() guarded := lite.Session(&gorm.Session{Logger: rec}) @@ -459,6 +603,7 @@ var _ = Describe("The claim queue", func() { _, _ = EnqueueClaim(ctx, guarded, ClaimKindMCPCI, JobEvent{JobID: "j1"}) _, _ = ReapAbandoned(ctx, guarded, time.Minute) _, _ = OwnerIsLive(ctx, guarded, "inst-a", time.Minute) + _ = ReleaseClaim(ctx, guarded, "some-claim") Expect(rec.count()).To(BeZero()) }) }) diff --git a/core/services/jobs/dispatch_loop.go b/core/services/jobs/dispatch_loop.go index 616025cc6..df49f857d 100644 --- a/core/services/jobs/dispatch_loop.go +++ b/core/services/jobs/dispatch_loop.go @@ -372,6 +372,14 @@ func (l *DispatchLoop) rebroadcast(nodeType, subject string, raw json.RawMessage // | | CompleteClaim | | // | anything else | ReleaseClaim | an unreachable peer, a lost tunnel or a refused stream is not a verdict | // +// The second row has no attempt ceiling, and there is deliberately no dead +// letter on it. Nothing that reaches it is the worker refusing the work: the +// row exists precisely for the outcomes where nothing was learned, and a +// ceiling would turn "the fleet was away long enough" into a job failure +// nobody reported. What bounds the retry is its RATE and not its count; see +// claimBackoff in claim.go, which is also what stops one stuck row from being +// re-claimed ahead of every newer one on every tick. +// // The line between the two rows is whether a REPLY LINE was decoded, and it is // deliberately NOT cluster.IsWorkerAnswer, though the plan for this task said // it should be. IsWorkerAnswer accepts the tunnel's stream-refusal vocabulary, @@ -390,7 +398,12 @@ func (l *DispatchLoop) rebroadcast(nodeType, subject string, raw json.RawMessage // which is the dropped-result defect the bus carrier had. func (l *DispatchLoop) settleClaim(ctx context.Context, claim *WorkClaim, reply *ClaimReply, callErr error) error { if callErr != nil { - xlog.Warn("Releasing a claim whose dispatch obtained no answer", "claim", claim.ID, "kind", claim.Kind, "error", callErr) + // attempts is the count BEFORE this release, so the line names the + // attempt that just failed. A row that has been stuck for a while + // prints this at the backoff cap rather than at the poll interval, + // which is what makes it readable as a growing number. + xlog.Warn("Releasing a claim whose dispatch obtained no answer; it will be retried after a backoff", + "claim", claim.ID, "kind", claim.Kind, "attempt", claim.Attempts+1, "error", callErr) if err := ReleaseClaim(ctx, l.db, claim.ID); err != nil { return fmt.Errorf("%w (and the claim could not be released: %w)", callErr, err) } diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index 72023cccd..13a25269f 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -369,6 +369,12 @@ This changes three behaviours an operator can see: - **A plain task job (a task whose model configures no MCP servers) is now failed with a reason.** No agent worker has ever served that kind of job, and it used to be published into silence. It is now marked `failed` with `no worker in this deployment serves plain task jobs`. This surfaces a pre-existing gap rather than introducing one. - **Dispatch is at-least-once instead of at-most-once.** A transport failure (a lost tunnel, an unreachable peer, a replica that died mid-dispatch) returns the work to the pool and increments the row's `attempts`; only the worker's own answer, success or failure, removes it. +**A claim that could not be dispatched is retried for ever, at a backing-off rate.** There is no attempt limit and no dead-letter queue, on purpose. Everything that returns a claim to the pool is a case where nothing was learned about the work: no agent worker was connected, the tunnel broke, a peer could not be reached, the stream was refused before the request body left the frontend. None of those is the worker saying it ran the job and it failed, so failing the job on any of them would report an absent connection as a verdict. A deployment whose agent workers are down for a day runs its queued work when they come back. + +What is bounded is the retry RATE. Each failed dispatch stamps the row with the earliest it may be claimed again, doubling from **2 seconds** to a cap of **60 seconds**, measured on the database clock. Two things follow. Queued work becomes claimable again within a minute of the fleet returning, and one permanently undispatchable row no longer starves the queue: rows are claimed oldest-first, so before the backoff the oldest stuck row was re-claimed ahead of every newer one on every tick and held a dispatch slot while it failed. A claim released by the reap because its replica died carries no delay at all, since that work was never handed to anyone. + +A row that keeps failing logs `Releasing a claim whose dispatch obtained no answer` with a growing `attempt` count, at most once per backoff interval. `SELECT id, kind, attempts, not_before FROM work_claims ORDER BY attempts DESC` is what tells you a job is stuck rather than merely queued. + **A claim held by a replica that has gone becomes claimable again; a claim held by a replica that is merely slow is never taken away from it.** The reap asks whether the claim's owner is still a live replica in the `instances` table, on the database clock, and never how long the claim has been held: a job that legitimately runs for an hour on a heartbeating replica is left alone, and a claim whose owner stopped heartbeating is released on the next tick (within `30s`, the replica-liveness window). **A frontend replica with no advertised peer address claims no work.** Such a replica has no row in the `instances` table, so no peer can tell its claims from ones a dead replica left, and another replica would take the work away from it mid-run. It logs an error naming `LOCALAI_DISTRIBUTED_ADVERTISE_ADDR` and starts claiming as soon as it registers. This is the same configuration that already makes a replica's workers unroutable from its peers.