fix(reliability): restore OOM, stuck-row recovery, backup-OK log, propagation lease

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

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 `[<ts>] 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
 }
 

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,13 +559,80 @@ 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)
 	}
 	committed = true
 	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

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

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`).