From a675701afffb12e04d0d24ede3cee92e708c4737 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Thu, 21 May 2026 21:53:35 +0530 Subject: [PATCH 1/3] fix(restore): stream-hash + stuck-row recovery in customer_restore_runner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit D26-F4 — Replace io.ReadAll(s3Reader) with io.TeeReader → temp-file + sha256.New() hasher pattern. Multi-GB pro/team backups previously OOM-killed the worker pod because the SHA-256 integrity gate must precede pg_restore (its --clean --if-exists is destructive — cannot stream-verify in-band). Memory footprint drops from O(backup-size) to O(io.Copy buffer ≈ 32 KiB). Temp file lives under os.TempDir(), removed via defer. D26-F5 — Add recoverStuckRestores mirroring customer_backup_runner's recoverStuckRows pass. A pod kill mid-pg_restore previously orphaned the resource_restores row at status='running' forever (the pending-row sweep only selects status='pending'). Stuck rows are now marked 'failed' with error_summary='worker_killed_during_restore: ...' once started_at exceeds restorePerRunTimeout. Deliberately NOT re-queued (unlike the backup path) — pg_restore is destructive, silent retry of a half-completed restore could compound partial-state damage; customer must manually re-issue. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/jobs/customer_restore_runner.go | 126 +++++++++++++++++++---- 1 file changed, 104 insertions(+), 22 deletions(-) diff --git a/internal/jobs/customer_restore_runner.go b/internal/jobs/customer_restore_runner.go index 5487b11..0068ccb 100644 --- a/internal/jobs/customer_restore_runner.go +++ b/internal/jobs/customer_restore_runner.go @@ -25,7 +25,6 @@ package jobs import ( - "bytes" "compress/gzip" "context" "crypto/sha256" @@ -36,6 +35,7 @@ import ( "fmt" "io" "log/slog" + "os" "os/exec" "strings" "time" @@ -129,6 +129,24 @@ func (w *CustomerRestoreRunnerWorker) Work(ctx context.Context, job *river.Job[C return nil } + // D26-F5 (2026-05-21): stuck-row recovery — mirrors the equivalent + // pass in customer_backup_runner.go (P2-W4). A restore row is + // atomically claimed by flipping status 'pending' → 'running'. If the + // worker pod is killed mid-pg_restore (rolling deploy, OOM, node + // drain) AFTER the claim but BEFORE finalize/markRestoreFailed, the + // row is orphaned at 'running' forever — the pending-row sweep below + // only selects status='pending', so no runner ever picks it up again. + // Customer support's "why is my restore stuck for 6h" ticket is the + // user-visible symptom. Recovery: any 'running' row whose started_at + // is older than restorePerRunTimeout could not still be a live + // in-flight restore (the per-run context would have fired), so it is + // marked 'failed' with an explicit error_summary. We deliberately do + // NOT re-queue these (unlike the backup runner) — restore is a + // destructive operation gated on user intent, and a silent retry of + // a half-completed pg_restore could compound database corruption. + // The user can re-issue the POST /resources/:id/restore manually. + w.recoverStuckRestores(ctx) + rows, err := w.db.QueryContext(ctx, ` SELECT rr.id::text, rr.resource_id::text, rr.backup_id::text, rb.s3_key, rb.sha256, @@ -212,6 +230,40 @@ func (w *CustomerRestoreRunnerWorker) Work(ctx context.Context, job *river.Job[C return nil } +// recoverStuckRestores marks restore rows orphaned at status='running' as +// 'failed' with a worker_killed_during_restore reason. A row qualifies only +// when started_at is older than restorePerRunTimeout — a genuinely +// in-flight restore is bounded by that per-run context, so anything older +// is a casualty of a pod kill, not a live job. Best-effort: a failure here +// is logged and the sweep proceeds. +// +// Unlike customer_backup_runner.recoverStuckRows (which re-queues to +// 'pending'), this path TERMINATES the row at 'failed'. pg_restore is +// destructive — a half-completed restore that re-queues for retry could +// compound the partial-state damage. The customer can manually re-issue +// POST /resources/:id/restore after inspecting the resource state. +func (w *CustomerRestoreRunnerWorker) recoverStuckRestores(ctx context.Context) { + res, err := w.db.ExecContext(ctx, ` + UPDATE resource_restores + SET status = 'failed', + finished_at = now(), + error_summary = 'worker_killed_during_restore: runner pod lost before finalize — pg_restore may have partially applied; re-issue restore after inspecting resource state' + WHERE status = 'running' + AND started_at IS NOT NULL + AND started_at < now() - ($1::int * INTERVAL '1 second') + `, int(w.timeout.Seconds())) + if err != nil { + slog.Warn("jobs.customer_restore_runner.stuck_row_recovery_failed", "error", err) + return + } + if n, raErr := res.RowsAffected(); raErr == nil && n > 0 { + slog.Warn("jobs.customer_restore_runner.recovered_stuck_rows", + "count", n, + "note", "rows orphaned at status='running' past the per-run timeout — marked failed (NOT re-queued; pg_restore is destructive)", + ) + } +} + // processRestore runs a single restore row. Returns true on success. func (w *CustomerRestoreRunnerWorker) processRestore(parentCtx context.Context, p struct { restoreID string @@ -277,31 +329,55 @@ func (w *CustomerRestoreRunnerWorker) processRestore(parentCtx context.Context, return false } - // Download from S3. We buffer the WHOLE object into memory rather than - // streaming it straight into gunzip→pg_restore, because the SHA-256 - // integrity check below must hash the exact bytes BEFORE pg_restore's - // `--clean --if-exists` DROPs every table. A streaming verify-as-you-go - // would only detect a mismatch after the destructive restore already ran. - // Memory cost is bounded by the per-tier backup size (the same gzipped - // object the backup runner already round-trips through io.Pipe). + // Download from S3 → temp file via streaming TeeReader. We MUST verify + // the SHA-256 BEFORE invoking pg_restore (its `--clean --if-exists` + // would otherwise DROP every table from an unverified — potentially + // truncated / bit-rotted — archive before the mismatch was detected). + // + // D26-F4 (2026-05-21): the previous implementation `io.ReadAll`-ed the + // whole object into RAM to satisfy the "hash before restore" ordering. + // A multi-GB pro/team backup OOM-killed the worker pod. Fix: tee the + // S3 read stream into BOTH a sha256.New() hasher AND a bounded temp + // file. Memory footprint stays at the io.Copy buffer (~32 KiB) while + // the digest is computed inline. The temp file is rewound to offset 0 + // for the gunzip + pg_restore step once the SHA gate has passed. + // + // Temp file lives under os.TempDir() (the pod's emptyDir / tmpfs); + // caller MUST `defer os.Remove` to avoid leaks across worker restarts. obj, dlErr := w.store.Download(ctx, w.bucket, p.s3Key.String) if dlErr != nil { w.markRestoreFailed(ctx, p.restoreID, fmt.Sprintf("S3 download failed: %v", dlErr), start, p) return false } - objBytes, readErr := io.ReadAll(obj) - obj.Close() - if readErr != nil { - w.markRestoreFailed(ctx, p.restoreID, fmt.Sprintf("S3 read failed: %v", readErr), start, p) + + tmpFile, tmpErr := os.CreateTemp("", "instant-restore-*.dump.gz") + if tmpErr != nil { + obj.Close() + w.markRestoreFailed(ctx, p.restoreID, fmt.Sprintf("create temp file: %v", tmpErr), start, p) return false } + // Always clean up — the temp file is destructively rewritten on every + // restore, and a pod restart would otherwise leak the file on /tmp. + defer func() { + _ = tmpFile.Close() + _ = os.Remove(tmpFile.Name()) + }() + + hasher := sha256.New() + tee := io.TeeReader(obj, hasher) + if _, copyErr := io.Copy(tmpFile, tee); copyErr != nil { + obj.Close() + w.markRestoreFailed(ctx, p.restoreID, fmt.Sprintf("S3 read failed: %v", copyErr), start, p) + return false + } + obj.Close() - // Integrity gate. The backup runner (customer_backup_runner.go, FIX-H - // #59) hashes the COMPRESSED (gzipped) object — its SHA-256 hasher sits - // in an io.MultiWriter fed by the gzip writer's output, i.e. the exact - // bytes uploaded to S3. So we hash objBytes here (still gzipped, BEFORE - // gunzip) to compare against the stored digest. A mismatch means the - // object bit-rotted or was truncated in transit → refuse to restore. + // Integrity gate (same semantics as before — only the hashing path + // changed). The backup runner (customer_backup_runner.go, FIX-H #59) + // hashes the COMPRESSED (gzipped) object — its SHA-256 hasher sits in + // an io.MultiWriter fed by the gzip writer's output, i.e. the exact + // bytes uploaded to S3. We hash the same gzipped stream here (BEFORE + // gunzip) to compare against the stored digest. // // Fail-open on a NULL/empty stored digest: rows predating migration // 043_backup_sha256.sql have no sha256, and the documented contract is @@ -314,8 +390,7 @@ func (w *CustomerRestoreRunnerWorker) processRestore(parentCtx context.Context, "backup_id", p.backupID, ) } else { - sum := sha256.Sum256(objBytes) - actualDigest := hex.EncodeToString(sum[:]) + actualDigest := hex.EncodeToString(hasher.Sum(nil)) if !strings.EqualFold(actualDigest, storedDigest) { w.markRestoreFailed(ctx, p.restoreID, fmt.Sprintf( "%s: sha256 mismatch (stored %s, downloaded %s) — backup object is corrupt or truncated; pg_restore NOT run", @@ -324,8 +399,15 @@ func (w *CustomerRestoreRunnerWorker) processRestore(parentCtx context.Context, } } - // Verified (or fail-open legacy) → gunzip the buffered bytes and restore. - gzReader, gzErr := gzip.NewReader(bytes.NewReader(objBytes)) + // Verified (or fail-open legacy) → rewind temp file, gunzip, and + // stream into pg_restore. The gunzip reader holds at most one block + // of decompressed data at a time, so the pg_restore stdin pipe never + // buffers the whole archive in memory. + if _, seekErr := tmpFile.Seek(0, io.SeekStart); seekErr != nil { + w.markRestoreFailed(ctx, p.restoreID, fmt.Sprintf("rewind temp file: %v", seekErr), start, p) + return false + } + gzReader, gzErr := gzip.NewReader(tmpFile) if gzErr != nil { w.markRestoreFailed(ctx, p.restoreID, fmt.Sprintf("gunzip header: %v", gzErr), start, p) return false From 99d92217d60b28e93e9181f8405a39ef87d07a69 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Thu, 21 May 2026 21:53:44 +0530 Subject: [PATCH 2/3] fix(platform-backup): emit 'backup OK' log line for NR staleness alert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit D26-F7 — infra/newrelic/alerts/backup-stale-36h.json keys on the literal string 'backup OK' (predicate: message LIKE '%backup OK%'). The k8s CronJob-based backups (postgres-customers-backup, mongodb-backup, redis-provision-backup) emit '[] backup OK' per shell convention; the River-based platform_db_backup job previously emitted only the structured 'jobs.platform_db_backup.succeeded' slog line and was invisible to the 36h/60h staleness alert. Add a second slog.Info('backup OK', kind='platform_db', ...) line alongside the existing structured emit. Both surfaces matter: the structured line stays for dashboard/query, the literal line triggers the alert. Rule 25 — every alert needs its trigger to actually fire. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/jobs/platform_db_backup.go | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/internal/jobs/platform_db_backup.go b/internal/jobs/platform_db_backup.go index e0d5caf..865e80e 100644 --- a/internal/jobs/platform_db_backup.go +++ b/internal/jobs/platform_db_backup.go @@ -455,6 +455,27 @@ func (w *PlatformDBBackupWorker) Work(ctx context.Context, job *river.Job[Platfo "s3_key", key, "swept_objects", deleted, ) + // D26-F7 (2026-05-21): emit the literal "backup OK" success line that + // the infra/newrelic/alerts/backup-stale-36h.json NR alert keys on. + // The CronJob-based backups (postgres-customers-backup, mongodb-backup, + // redis-provision-backup) emit `[] backup OK` per the alert's + // `message LIKE '%backup OK%'` predicate; this River job previously + // emitted only the structured `jobs.platform_db_backup.succeeded` + // line and was invisible to the staleness alert. Operator-visible + // symptom pre-fix: a silent platform_db_backup failure would persist + // until someone manually inspected audit_log for + // platform_backup.failed. + // + // Keep both lines: the structured one is the dashboard/query surface, + // the literal one is the alert trigger. Rule 25 ("every metric + // ships with its alert") — same principle applies to log-based + // alerts. + slog.Info("backup OK", + "kind", "platform_db", + "size_bytes", size, + "duration_seconds", duration.Seconds(), + "s3_key", key, + ) return nil } From 19ff581b444673a46ac5ade14e12561f9cb26725 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Thu, 21 May 2026 21:53:53 +0530 Subject: [PATCH 3/3] fix(propagation): crash-safe lease bump in pickEligible tx MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit D22-P3 — pickEligible's tx committed after the SELECT FOR UPDATE SKIP LOCKED, leaving next_attempt_at unchanged. A pod crash between commit and the handler's terminal markApplied/markRetry left the row eligible for re-pick on the next 30s tick with attempts unchanged — a flaky pod could spin the same row through dispatch every 30s without ever incurring backoff, eventually exhausting maxAttempts against transient pod failures rather than genuine downstream outages. Add a propagationLeaseDuration (5m, ≥ propagationDispatchTimeout) push on next_attempt_at INSIDE the same picker tx. Healthy run: markApplied or markRetry overwrites the lease as before. Crash run: the row falls back to the lease window — re-picked after 5m, attempts unchanged (lease is a crash-recovery mechanism, not a dispatch failure → should not consume a retry attempt). Hand-rolled pgUUIDArray formatter so the worker stays driver-agnostic (no lib/pq pq.Array import). Lease bump is best-effort — a bump failure falls back to pre-fix behavior + logs WARN; the dispatch loop remains correct. Co-Authored-By: Claude Opus 4.7 (1M context) --- internal/jobs/propagation_runner.go | 99 +++++++++++++++++++ .../propagation_runner_integration_test.go | 4 + internal/jobs/propagation_runner_test.go | 15 +++ 3 files changed, 118 insertions(+) diff --git a/internal/jobs/propagation_runner.go b/internal/jobs/propagation_runner.go index 4e7a447..3b502bc 100644 --- a/internal/jobs/propagation_runner.go +++ b/internal/jobs/propagation_runner.go @@ -474,6 +474,29 @@ func (w *PropagationRunnerWorker) Work(ctx context.Context, job *river.Job[Propa return nil } +// propagationLeaseDuration is how far forward we push next_attempt_at on +// pick, BEFORE dispatching. The value MUST be >= propagationDispatchTimeout +// — otherwise a long-running handler could see its row become eligible +// again under a sibling pod's picker while the original handler is still +// mid-flight, leading to redundant double-dispatch (idempotent, but +// wastes provisioner cycles). +// +// D22-P3 (2026-05-21): a crash between pickEligible's COMMIT and the +// handler's markRetry call previously left next_attempt_at unchanged. +// On the next tick (30s later), the picker re-selected the same row +// with attempts unchanged — a crashing pod could spin the same row +// through the dispatch loop every 30s without ever incurring backoff, +// effectively burning the maxAttempts budget against transient pod +// failures rather than genuine downstream outages. The lease bump +// inside the pick transaction pre-emptively schedules the row out by +// propagationLeaseDuration; the normal dispatch path's markApplied / +// markRetry overwrites this value on completion, so a healthy run is +// unaffected. A crashed dispatch falls back to the lease — the row +// re-enters the picker after propagationLeaseDuration with attempts +// unchanged (the actual retry counter is bumped only by markRetry, so +// a crashed dispatch correctly does NOT consume an attempt). +const propagationLeaseDuration = 5 * time.Minute + // pickEligible runs the SELECT … FOR UPDATE SKIP LOCKED that the runner // uses to claim a batch. Each picked row is implicitly locked for the // duration of the surrounding transaction; we COMMIT inside the picker @@ -486,6 +509,15 @@ func (w *PropagationRunnerWorker) Work(ctx context.Context, job *river.Job[Propa // We use the FOR UPDATE SKIP LOCKED clause to keep two concurrent picks // from claiming the SAME rows (otherwise both pods would dispatch the // same handler twice in the same tick window). +// +// D22-P3 (2026-05-21): before COMMIT, we UPDATE the just-picked rows to +// push next_attempt_at forward by propagationLeaseDuration. This is the +// crash-safe lease: if the pod dies between COMMIT and the handler's +// terminal markApplied/markRetry, the row is no longer eligible for +// re-pick for propagationLeaseDuration. On the happy path the lease is +// overwritten by markApplied/markRetry as before; on the crash path the +// row gets a free backoff penalty (the same lease window) without +// burning a retry attempt. func (w *PropagationRunnerWorker) pickEligible(ctx context.Context) ([]propagationRow, error) { tx, err := w.db.BeginTx(ctx, &sql.TxOptions{}) if err != nil { @@ -527,6 +559,47 @@ func (w *PropagationRunnerWorker) pickEligible(ctx context.Context) ([]propagati } rows.Close() + // D22-P3 lease bump (2026-05-21). Push next_attempt_at on the + // just-picked, still-locked rows so a pod crash between COMMIT below + // and the per-row dispatch can't immediately re-stage the same rows + // on the next 30s tick. Done inside the SAME tx as the SELECT FOR + // UPDATE so the bump is atomic with the pick — no window where a + // sibling pod could SELECT before the bump. + // + // We do NOT bump attempts here — attempts is the failed-dispatch + // counter and a crashed pod's row shouldn't burn an attempt against + // the maxAttempts budget. last_error stays unchanged for the same + // reason; a crash is not a dispatch failure. + if len(out) > 0 { + ids := make([]uuid.UUID, 0, len(out)) + for _, r := range out { + ids = append(ids, r.id) + } + // pq array binding via pq.Array would require importing lib/pq + // here; the picker already runs against pgx-compatible drivers + // where = ANY($1) accepts a []uuid.UUID. We pass the slice + // directly — both pgx and lib/pq honour this when the column + // type is uuid[]. + nextAttempt := w.now().Add(propagationLeaseDuration) + if _, leaseErr := tx.ExecContext(ctx, ` + UPDATE pending_propagations + SET next_attempt_at = $1 + WHERE id = ANY($2::uuid[]) + AND applied_at IS NULL + AND failed_at IS NULL + `, nextAttempt, pgUUIDArray(ids)); leaseErr != nil { + // Lease bump failed — log + proceed. The downside is the + // pre-D22-P3 behavior (crash → immediate re-pick) which is + // the SAME as before this fix shipped; the dispatch loop + // remains correct. We deliberately do NOT abort the tick. + slog.Warn("jobs.propagation_runner.lease_bump_failed", + "error", leaseErr, + "picked_count", len(out), + "note", "rows will dispatch but crash-resume window may re-pick on next tick", + ) + } + } + if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit pick tx: %w", err) } @@ -534,6 +607,32 @@ func (w *PropagationRunnerWorker) pickEligible(ctx context.Context) ([]propagati return out, nil } +// pgUUIDArray formats a []uuid.UUID as a Postgres array literal suitable +// for `$N::uuid[]` parameter binding via the standard database/sql driver. +// Returns the literal as a string (e.g. `{a-b-c-...,d-e-f-...}`) — the +// driver round-trips this through the array text protocol. Empty slice +// returns `{}` (a valid empty PG array). +// +// We hand-roll this rather than depend on lib/pq's Array() helper so the +// worker stays driver-agnostic — the propagation_runner_integration_test +// runs against TEST_DATABASE_URL with whichever driver the operator +// configures (pgx or lib/pq). +func pgUUIDArray(ids []uuid.UUID) string { + if len(ids) == 0 { + return "{}" + } + var b strings.Builder + b.WriteByte('{') + for i, id := range ids { + if i > 0 { + b.WriteByte(',') + } + b.WriteString(id.String()) + } + b.WriteByte('}') + return b.String() +} + // propagationBackoffFor returns the delay to apply BEFORE the next attempt // given the row's PRE-increment attempts count. attempts is the failed-attempt // counter that will be UPDATEd to attempts+1; the index into the schedule is diff --git a/internal/jobs/propagation_runner_integration_test.go b/internal/jobs/propagation_runner_integration_test.go index c89aa96..c2e2d80 100644 --- a/internal/jobs/propagation_runner_integration_test.go +++ b/internal/jobs/propagation_runner_integration_test.go @@ -237,6 +237,10 @@ func TestPropagation_UnknownKindIntegration_BoundedRetries(t *testing.T) { mock.ExpectQuery(`SELECT id, kind, team_id, target_tier, payload, attempts\s+FROM pending_propagations`). WillReturnRows(sqlmock.NewRows(propagationSweepCols). AddRow(propID, "garbage_kind_nobody_handles", teamID, "pro", []byte(`{}`), 0)) + // D22-P3 lease bump (2026-05-21). + mock.ExpectExec(`UPDATE pending_propagations\s+SET next_attempt_at\s*=\s*\$1\s+WHERE id = ANY\(\$2::uuid\[\]\)`). + WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectCommit() // Expected: markRetry fires (attempts++ + audit row). On master diff --git a/internal/jobs/propagation_runner_test.go b/internal/jobs/propagation_runner_test.go index 611d9de..4e60665 100644 --- a/internal/jobs/propagation_runner_test.go +++ b/internal/jobs/propagation_runner_test.go @@ -125,6 +125,13 @@ func TestPropagationRunner_AppliesEligibleRow(t *testing.T) { mock.ExpectQuery(`SELECT id, kind, team_id, target_tier, payload, attempts\s+FROM pending_propagations`). WillReturnRows(sqlmock.NewRows(propagationSweepCols). AddRow(propID, propagationKindTierElevation, teamID, "pro", []byte(`{}`), 0)) + // D22-P3 lease bump (2026-05-21): inside the pick tx, push + // next_attempt_at on the just-picked rows by propagationLeaseDuration + // so a crash before markApplied/markRetry can't immediately re-stage + // the row on the next 30s tick. + mock.ExpectExec(`UPDATE pending_propagations\s+SET next_attempt_at\s*=\s*\$1\s+WHERE id = ANY\(\$2::uuid\[\]\)`). + WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectCommit() // Handler queries the team's resources. @@ -172,6 +179,10 @@ func TestPropagationRunner_RetryOnFailure_PersistsBackoff(t *testing.T) { mock.ExpectQuery(`SELECT id, kind, team_id, target_tier, payload, attempts\s+FROM pending_propagations`). WillReturnRows(sqlmock.NewRows(propagationSweepCols). AddRow(propID, propagationKindTierElevation, teamID, "pro", []byte(`{}`), 0)) + // D22-P3 lease bump (2026-05-21). + mock.ExpectExec(`UPDATE pending_propagations\s+SET next_attempt_at\s*=\s*\$1\s+WHERE id = ANY\(\$2::uuid\[\]\)`). + WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectCommit() mock.ExpectQuery(`SELECT r\.id, r\.token, r\.provider_resource_id, r\.tier, r\.resource_type`). @@ -225,6 +236,10 @@ func TestPropagationRunner_DeadLettersAfterMaxAttempts(t *testing.T) { mock.ExpectQuery(`SELECT id, kind, team_id, target_tier, payload, attempts\s+FROM pending_propagations`). WillReturnRows(sqlmock.NewRows(propagationSweepCols). AddRow(propID, propagationKindTierElevation, teamID, "pro", []byte(`{}`), propagationMaxAttempts-1)) + // D22-P3 lease bump (2026-05-21). + mock.ExpectExec(`UPDATE pending_propagations\s+SET next_attempt_at\s*=\s*\$1\s+WHERE id = ANY\(\$2::uuid\[\]\)`). + WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectCommit() mock.ExpectQuery(`SELECT r\.id, r\.token, r\.provider_resource_id, r\.tier, r\.resource_type`).