From 222494d1c5ddcf520c458acb19b4f10b0ef1d68b Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Wed, 3 Jun 2026 10:44:34 +0530 Subject: [PATCH 1/3] fix(backup): stop leaking internals on backup failure + classify + meter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Incident: a customer received a "Backup failed" notification containing raw pg_dump stderr — the internal host/IP (pg.instanode.dev / 152.42.154.144:5432) and the per-tenant role name (usr_…). The message was also non-actionable: the backup credential is platform-managed, so there's nothing the user can do. And because markFailed emitted no metric, a clean failure slipped past both existing backup alerts (backup-stale fires only after 36h; no-followup only on STUCK rows) — a paid-tier durability/SLA breach with no prompt page. Root cause of *that* failure: crypto.Decrypt fails CLOSED, and the error was "password authentication failed" (not a decrypt error) — so the credential decrypted fine but the live DB role rejected it (stored connection_url drifted from the role's password). That per-resource drift needs an ops fix; this change makes the *class* of failure safe, classified, and observable: - markFailed now persists a SANITIZED, customer-safe summary to resource_backups.error_summary + the backup.failed audit (what the email and backup-health surface read). Raw stderr stays in the worker log only. - Classify failures by reason (auth|decrypt|config|dump|upload) via backupFailReason(); auth = credential drift (won't self-heal, SLA-relevant, paged) vs transient dump/upload (retried next run). - New metrics instant_customer_backup_failed_total{reason} + instant_customer_backup_succeeded_total for an alert + success-ratio tile (rule 25 — NR alert + Prom rule + dashboard tile land in the infra PR). Retry semantics already correct: Work() returns nil after markFailed, so auth failures aren't hammered. Tests: table tests for backupFailReason (incl. the exact prod stderr) and an anti-leak guard asserting the sanitized summary never contains the host/IP/ role/pg_dump/password tokens. make gate green; new funcs 100% covered. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/jobs/customer_backup_reason_test.go | 78 +++++++++++++++++++ internal/jobs/customer_backup_runner.go | 82 +++++++++++++++++--- internal/metrics/metrics.go | 25 ++++++ 3 files changed, 176 insertions(+), 9 deletions(-) create mode 100644 internal/jobs/customer_backup_reason_test.go diff --git a/internal/jobs/customer_backup_reason_test.go b/internal/jobs/customer_backup_reason_test.go new file mode 100644 index 0000000..201702a --- /dev/null +++ b/internal/jobs/customer_backup_reason_test.go @@ -0,0 +1,78 @@ +package jobs + +// customer_backup_reason_test.go — unit tests for the backup-failure reason +// classifier and the customer-safe message sanitizer (2026-06-03 backup +// observability fix). These pin two invariants: +// 1. A credential/auth failure is classified "auth" (SLA-relevant, paged) +// and distinguished from a transient "dump"/"upload" failure. +// 2. The customer-facing summary NEVER leaks internal detail (host/IP, +// per-tenant role name, raw pg_dump stderr) — the incident that +// triggered this fix forwarded exactly that to a user. + +import ( + "errors" + "strings" + "testing" +) + +func TestBackupFailReason(t *testing.T) { + cases := []struct { + name string + err error + want string + }{ + // The exact prod stderr that triggered this fix. + {"prod password auth", errors.New(`pg_dump: error: connection to server at "pg.instanode.dev" (152.42.154.144), port 5432 failed: FATAL: password authentication failed for user "usr_96edf9eed8ed42929036b63298ec5b2b"`), "auth"}, + {"generic auth failed", errors.New("authentication failed"), "auth"}, + {"no password supplied", errors.New("pg_dump: error: no password supplied"), "auth"}, + {"role does not exist", errors.New(`FATAL: role "usr_abc" does not exist`), "auth"}, + {"permission denied", errors.New("permission denied for table users"), "auth"}, + {"server unavailable is transient", errors.New("pg_dump: server unavailable"), "dump"}, + {"connection refused is transient", errors.New("connection refused"), "dump"}, + {"nil err defaults to dump", nil, "dump"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := backupFailReason(c.err); got != c.want { + t.Fatalf("backupFailReason(%q) = %q, want %q", c.err, got, c.want) + } + }) + } +} + +func TestSanitizedBackupFailure_NeverLeaksInternals(t *testing.T) { + // Tokens from the real leaked error that must NEVER appear in a + // customer-facing summary, regardless of reason. + leaks := []string{ + "pg.instanode.dev", "152.42.154.144", "5432", + "usr_96edf9eed8ed42929036b63298ec5b2b", "pg_dump", "password", "FATAL", + } + for _, reason := range []string{"auth", "decrypt", "config", "dump", "upload", "other", ""} { + msg := sanitizedBackupFailure(reason) + if strings.TrimSpace(msg) == "" { + t.Fatalf("sanitizedBackupFailure(%q) is empty", reason) + } + low := strings.ToLower(msg) + for _, leak := range leaks { + if strings.Contains(low, strings.ToLower(leak)) { + t.Errorf("sanitizedBackupFailure(%q) leaks %q: %s", reason, leak, msg) + } + } + } +} + +func TestSanitizedBackupFailure_PerReasonCopy(t *testing.T) { + // auth → reassuring + "no action needed"; transient → "try again". + if !strings.Contains(strings.ToLower(sanitizedBackupFailure("auth")), "no action") { + t.Error("auth message should reassure the user no action is needed") + } + for _, r := range []string{"dump", "upload"} { + if !strings.Contains(strings.ToLower(sanitizedBackupFailure(r)), "try again") { + t.Errorf("%q message should say we'll retry", r) + } + } + // decrypt/config are internal config issues — surfaced as such, no leak. + if !strings.Contains(strings.ToLower(sanitizedBackupFailure("config")), "internal configuration") { + t.Error("config message should name an internal configuration issue") + } +} diff --git a/internal/jobs/customer_backup_runner.go b/internal/jobs/customer_backup_runner.go index e85525b..c53657d 100644 --- a/internal/jobs/customer_backup_runner.go +++ b/internal/jobs/customer_backup_runner.go @@ -62,6 +62,7 @@ import ( "instant.dev/common/crypto" "instant.dev/worker/internal/apiclient" "instant.dev/worker/internal/circuit" + "instant.dev/worker/internal/metrics" ) // CustomerBackupRunnerArgs holds no fields — periodic job. @@ -410,17 +411,17 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p // or malformed ciphertext is a hard failure since we can't safely // dump from a guess. if !p.connURL.Valid || p.connURL.String == "" { - w.markFailed(ctx, p.backupID, "resource.connection_url is empty", start, p) + w.markFailed(ctx, p.backupID, "config", "resource.connection_url is empty", start, p) return false } aesKey, keyErr := crypto.ParseAESKey(w.aesKey) if keyErr != nil { - w.markFailed(ctx, p.backupID, fmt.Sprintf("AES key invalid: %v", keyErr), start, p) + w.markFailed(ctx, p.backupID, "config", fmt.Sprintf("AES key invalid: %v", keyErr), start, p) return false } plainConn, decErr := crypto.Decrypt(aesKey, p.connURL.String) if decErr != nil { - w.markFailed(ctx, p.backupID, fmt.Sprintf("decrypt connection_url: %v", decErr), start, p) + w.markFailed(ctx, p.backupID, "decrypt", fmt.Sprintf("decrypt connection_url: %v", decErr), start, p) return false } @@ -479,7 +480,7 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p // actionable: "pg_dump: connection refused" vs "pipe: io: read/write // on closed pipe"). if dumpErr != nil { - w.markFailed(ctx, p.backupID, fmt.Sprintf("pg_dump failed: %v", dumpErr), start, p) + w.markFailed(ctx, p.backupID, backupFailReason(dumpErr), fmt.Sprintf("pg_dump failed: %v", dumpErr), start, p) // Best-effort cleanup of a half-written object so we don't pay // for orphan bytes; failure to delete is logged but not fatal. if delErr := w.store.DeleteObject(parentCtx, w.bucket, objectKey); delErr != nil { @@ -489,7 +490,7 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p return false } if upErr != nil { - w.markFailed(ctx, p.backupID, fmt.Sprintf("S3 upload failed: %v", upErr), start, p) + w.markFailed(ctx, p.backupID, "upload", fmt.Sprintf("S3 upload failed: %v", upErr), start, p) return false } @@ -540,6 +541,7 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p }) } + metrics.CustomerBackupSucceededTotal.Inc() slog.Info("jobs.customer_backup_runner.succeeded", "backup_id", p.backupID, "resource_id", p.resourceID, @@ -558,8 +560,55 @@ func (w *CustomerBackupRunnerWorker) processBackup(parentCtx context.Context, p // the api's internal refund endpoint so the team's daily counter is // credited. Scheduled backups don't burn the manual-counter so no // refund is needed. +// backupFailReason classifies a pg_dump failure into "auth" (the credential +// was rejected — password auth failed, missing role, no password supplied: +// credential drift that will NOT self-heal and is SLA-relevant) vs "dump" (any +// other pg_dump failure — DB briefly unreachable, timeout: transient, retried +// next run). The match is on Postgres' own error text, lower-cased so it's +// resilient to surrounding formatting. +func backupFailReason(err error) string { + if err == nil { + return "dump" + } + s := strings.ToLower(err.Error()) + switch { + case strings.Contains(s, "password authentication failed"), + strings.Contains(s, "authentication failed"), + strings.Contains(s, "no password supplied"), + strings.Contains(s, "role") && strings.Contains(s, "does not exist"), + strings.Contains(s, "permission denied for"): + return "auth" + default: + return "dump" + } +} + +// sanitizedBackupFailure maps an internal failure reason to a customer-safe, +// actionable message. It deliberately contains NO internal host/IP, per-tenant +// role name, or raw pg_dump stderr — that detail stays in the worker log only. +// This string is what lands in resource_backups.error_summary, the failure +// email, and the customer-visible backup-health surface. +func sanitizedBackupFailure(reason string) string { + switch reason { + case "auth": + return "We couldn't authenticate to your database to take this backup. " + + "Our team has been alerted and is investigating — no action is needed from you." + case "decrypt", "config": + return "This backup couldn't run due to an internal configuration issue. " + + "Our team has been alerted — no action is needed from you." + case "dump": + return "We couldn't read your database for this backup (it may have been " + + "briefly unreachable). We'll automatically try again on the next scheduled run." + case "upload": + return "The backup was created but couldn't be stored. " + + "We'll automatically try again on the next scheduled run." + default: + return "This backup didn't complete. Our team has been alerted." + } +} + func (w *CustomerBackupRunnerWorker) markFailed( - ctx context.Context, backupID, errSummary string, start time.Time, + ctx context.Context, backupID, reason, internalDetail string, start time.Time, p struct { backupID string resourceID string @@ -571,6 +620,18 @@ func (w *CustomerBackupRunnerWorker) markFailed( teamID uuid.NullUUID }, ) { + // Observability: count by reason so an SLA-relevant credential/auth drift + // (reason="auth", PAGE) is distinguishable from a transient dump/upload + // failure (retried next run). NR alert: customer-backup-failed.json. + metrics.CustomerBackupFailedTotal.WithLabelValues(reason).Inc() + + // Two summaries: a SANITIZED, user-safe one persisted to the DB + audit + // (it surfaces on the customer's failure email and the backup-health + // dashboard), and the FULL internalDetail kept only in the worker log + // (NR Logs, ops-only). Never persist raw pg_dump stderr — it leaks the + // internal host/IP and per-tenant role name to the customer. + publicSummary := sanitizedBackupFailure(reason) + // Use a fresh ctx with a small timeout so a parentCtx-already-cancelled // path still gets the row updated. dbCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) @@ -582,7 +643,7 @@ func (w *CustomerBackupRunnerWorker) markFailed( finished_at = now(), error_summary = $2 WHERE id = $1 - `, backupID, errSummary); err != nil { + `, backupID, publicSummary); err != nil { slog.Error("jobs.customer_backup_runner.mark_failed_db_error", "backup_id", backupID, "error", err) } @@ -592,15 +653,18 @@ func (w *CustomerBackupRunnerWorker) markFailed( w.writeAudit(dbCtx, p.teamID.UUID, p.resourceID, p.resourceType, auditKindBackupFailed, "Backup failed", map[string]any{ "backup_id": backupID, - "error_summary": errSummary, + "error_summary": publicSummary, + "reason": reason, "duration_seconds": int(duration.Seconds()), "tier": p.tier.String, }) } + // Full internal detail — including raw pg_dump stderr — stays HERE only. slog.Error("jobs.customer_backup_runner.failed", "backup_id", backupID, - "error_summary", errSummary, + "reason", reason, + "internal_detail", internalDetail, "duration_ms", duration.Milliseconds(), ) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index e87061b..6820173 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -68,6 +68,31 @@ var ( Help: "Resource heartbeat probe attempts by type and outcome", }, []string{"resource_type", "outcome"}) + // CustomerBackupFailedTotal counts customer (per-tenant) backup runs that + // failed, labelled by reason so an SLA-relevant credential/auth drift is + // distinguishable from a transient timeout: + // auth — pg_dump rejected the credential (password auth failed / role + // missing). Credential drift between the stored connection_url + // and the live DB role. SLA-relevant; PAGE (P1) — won't self-heal. + // decrypt — connection_url could not be decrypted (AES key mismatch). + // config — empty connection_url / invalid AES key (misconfiguration). + // dump — pg_dump failed for a non-auth reason (DB briefly unreachable); + // transient, retried on the next scheduled run. + // upload — snapshot produced but S3 upload failed; transient. + // NR alert: customer-backup-failed.json. Prom: instant-backups group. + CustomerBackupFailedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "instant_customer_backup_failed_total", + Help: "Customer backup runs that failed, by reason (auth|decrypt|config|dump|upload). auth = credential drift, SLA-relevant.", + }, []string{"reason"}) + + // CustomerBackupSucceededTotal counts customer backup runs that completed + // and were durably stored in S3. Paired with CustomerBackupFailedTotal to + // compute a per-window success ratio on the backup-health dashboard. + CustomerBackupSucceededTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "instant_customer_backup_succeeded_total", + Help: "Customer backup runs that completed and were stored in S3.", + }) + // ResourceDegradedGauge is sampled at the end of each heartbeat run. // Labelled by resource_type so the dashboard can break down "how many // of my Postgres instances are unreachable right now". From 0dfc8ff09d0f42ea5e305fc2ed1d8e0715bbf8c9 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Wed, 3 Jun 2026 11:43:53 +0530 Subject: [PATCH 2/3] feat(backup): retain only the last 5 healthy backups per resource MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Operator request (2026-06-03): cap retained successful backups at 5 per resource. Adds runKeepLastNSweep — retires every status='ok' backup that is NOT among the keepHealthyBackupsPerResource (=5) most-recent for its resource, even if still within the tier's time-based retention window. Reuses runRetentionSweep's retire mechanism (delete S3 object, soft-flag the row via s3_key=NULL; error_summary='retained:count-cap' distinguishes it from the time-based 'retained:expired'). Runs at the end of each backup tick alongside the existing retention sweep; fail-soft per victim so one S3/DB blip never blocks the rest. Tests: retire-beyond-cap, within-cap no-op, query-error, scan-error, delete-error (skips DB update), db-update-error (fails soft). New method 100% covered. make gate green. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/jobs/customer_backup_keepn_test.go | 149 ++++++++++++++++++++ internal/jobs/customer_backup_runner.go | 66 +++++++++ 2 files changed, 215 insertions(+) create mode 100644 internal/jobs/customer_backup_keepn_test.go diff --git a/internal/jobs/customer_backup_keepn_test.go b/internal/jobs/customer_backup_keepn_test.go new file mode 100644 index 0000000..d2b8da0 --- /dev/null +++ b/internal/jobs/customer_backup_keepn_test.go @@ -0,0 +1,149 @@ +package jobs + +// customer_backup_keepn_test.go — count-based retention ("keep last N healthy +// backups per resource", 2026-06-03 operator request). Covers runKeepLastNSweep: +// every status='ok' backup beyond the newest keepHealthyBackupsPerResource for +// its resource is retired (S3 object deleted + row soft-flagged via s3_key=NULL). + +import ( + "context" + "errors" + "testing" + + sqlmock "github.com/DATA-DOG/go-sqlmock" +) + +// deleteErrStore wraps the in-memory store but always errors on DeleteObject, +// exercising the per-victim S3-failure (skip) branch. +type deleteErrStore struct{ *fakeBackupStore } + +func (d deleteErrStore) DeleteObject(_ context.Context, _, _ string) error { + return errors.New("s3 unavailable") +} + +func TestRunKeepLastNSweep_RetiresBeyondCap(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + + // Window query returns the over-cap victims (rn > keep). Two here. + mock.ExpectQuery(`row_number\(\) OVER`). + WithArgs(keepHealthyBackupsPerResource). + WillReturnRows(sqlmock.NewRows([]string{"id", "s3_key"}). + AddRow("b6", "backups/r/b6.dump.gz"). + AddRow("b7", "backups/r/b7.dump.gz")) + mock.ExpectExec(`UPDATE resource_backups`). + WithArgs("b6").WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(`UPDATE resource_backups`). + WithArgs("b7").WillReturnResult(sqlmock.NewResult(0, 1)) + + store := newFakeBackupStore() + w := &CustomerBackupRunnerWorker{db: db, store: store, bucket: "instant-shared"} + w.runKeepLastNSweep(context.Background()) + + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations: %v", err) + } + if len(store.deletes) != 2 { + t.Fatalf("expected 2 S3 objects retired, got %d (%v)", len(store.deletes), store.deletes) + } +} + +func TestRunKeepLastNSweep_NoVictimsWithinCap(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + + // Resource has <= keep backups → window query returns nothing → no deletes. + mock.ExpectQuery(`row_number\(\) OVER`). + WithArgs(keepHealthyBackupsPerResource). + WillReturnRows(sqlmock.NewRows([]string{"id", "s3_key"})) + + store := newFakeBackupStore() + w := &CustomerBackupRunnerWorker{db: db, store: store, bucket: "instant-shared"} + w.runKeepLastNSweep(context.Background()) + + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations: %v", err) + } + if len(store.deletes) != 0 { + t.Fatalf("expected 0 retired when within cap, got %d", len(store.deletes)) + } +} + +func TestRunKeepLastNSweep_QueryError_FailsSoft(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + mock.ExpectQuery(`row_number\(\) OVER`). + WithArgs(keepHealthyBackupsPerResource). + WillReturnError(errors.New("db blip")) + w := &CustomerBackupRunnerWorker{db: db, store: newFakeBackupStore(), bucket: "b"} + w.runKeepLastNSweep(context.Background()) // must not panic; fails soft + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations: %v", err) + } +} + +func TestRunKeepLastNSweep_ScanError_SkipsRow(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + // NULL s3_key cannot scan into a string → scan error → row skipped. + mock.ExpectQuery(`row_number\(\) OVER`). + WithArgs(keepHealthyBackupsPerResource). + WillReturnRows(sqlmock.NewRows([]string{"id", "s3_key"}).AddRow("b6", nil)) + store := newFakeBackupStore() + w := &CustomerBackupRunnerWorker{db: db, store: store, bucket: "b"} + w.runKeepLastNSweep(context.Background()) + if len(store.deletes) != 0 { + t.Fatalf("scan-failed row must not be retired, got %d deletes", len(store.deletes)) + } +} + +func TestRunKeepLastNSweep_DeleteError_SkipsUpdate(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + mock.ExpectQuery(`row_number\(\) OVER`). + WithArgs(keepHealthyBackupsPerResource). + WillReturnRows(sqlmock.NewRows([]string{"id", "s3_key"}).AddRow("b6", "backups/r/b6.dump.gz")) + // No ExpectExec(UPDATE): an S3 delete failure must skip the row's DB update. + w := &CustomerBackupRunnerWorker{db: db, store: deleteErrStore{newFakeBackupStore()}, bucket: "b"} + w.runKeepLastNSweep(context.Background()) + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations (UPDATE must NOT run on delete error): %v", err) + } +} + +func TestRunKeepLastNSweep_DBUpdateError_FailsSoft(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + mock.ExpectQuery(`row_number\(\) OVER`). + WithArgs(keepHealthyBackupsPerResource). + WillReturnRows(sqlmock.NewRows([]string{"id", "s3_key"}).AddRow("b6", "backups/r/b6.dump.gz")) + mock.ExpectExec(`UPDATE resource_backups`). + WithArgs("b6").WillReturnError(errors.New("update blip")) + store := newFakeBackupStore() + w := &CustomerBackupRunnerWorker{db: db, store: store, bucket: "b"} + w.runKeepLastNSweep(context.Background()) // soft-fails on the update error + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet sqlmock expectations: %v", err) + } + if len(store.deletes) != 1 { + t.Fatalf("object should still be deleted before the failed DB update, got %d", len(store.deletes)) + } +} diff --git a/internal/jobs/customer_backup_runner.go b/internal/jobs/customer_backup_runner.go index c53657d..e5364c1 100644 --- a/internal/jobs/customer_backup_runner.go +++ b/internal/jobs/customer_backup_runner.go @@ -303,6 +303,9 @@ func (w *CustomerBackupRunnerWorker) Work(ctx context.Context, job *river.Job[Cu // Retention sweep at end of run. A failure here doesn't unwind the // successful uploads from the same tick. w.runRetentionSweep(ctx) + // Count cap: keep only the last N healthy backups per resource (retire + // older ones even if within the tier's time window). 2026-06-03 request. + w.runKeepLastNSweep(ctx) // T21 P1-1 (BugBash 2026-05-20): idle-tick demoted INFO→DEBUG. The // runner is invoked per River batch; the steady state in prod is @@ -799,6 +802,69 @@ func (w *CustomerBackupRunnerWorker) runRetentionSweep(ctx context.Context) { } } +// keepHealthyBackupsPerResource caps how many successful backups are retained +// per resource. Beyond this count, the OLDEST ok backups are retired (S3 +// object deleted + row soft-flagged) even if still inside the tier's +// time-based retention window — "only keep the last N healthy backups" +// (2026-06-03 operator request). +const keepHealthyBackupsPerResource = 5 + +// runKeepLastNSweep retires every status='ok' backup that is NOT among the +// keepHealthyBackupsPerResource most-recent ok backups for its resource. It +// reuses runRetentionSweep's retire mechanism: delete the S3 object, then +// soft-flag the row (s3_key=NULL so the api list/restore surfaces drop it). +// The error_summary marker distinguishes count-cap retirement from the +// time-based 'retained:expired'. Fail-soft per victim — an S3 or DB blip on +// one row never blocks the rest of the sweep. +func (w *CustomerBackupRunnerWorker) runKeepLastNSweep(ctx context.Context) { + rows, err := w.db.QueryContext(ctx, ` + SELECT id::text, s3_key FROM ( + SELECT id, s3_key, + row_number() OVER (PARTITION BY resource_id ORDER BY created_at DESC) AS rn + FROM resource_backups + WHERE status = 'ok' AND s3_key IS NOT NULL + ) ranked + WHERE rn > $1 + LIMIT 500 + `, keepHealthyBackupsPerResource) + if err != nil { + slog.Warn("jobs.customer_backup_runner.keep_last_n_query_failed", "error", err) + return + } + type victim struct{ id, s3Key string } + var victims []victim + for rows.Next() { + var v victim + if scanErr := rows.Scan(&v.id, &v.s3Key); scanErr != nil { + slog.Warn("jobs.customer_backup_runner.keep_last_n_scan_failed", "error", scanErr) + continue + } + victims = append(victims, v) + } + _ = rows.Close() + + for _, v := range victims { + if delErr := w.store.DeleteObject(ctx, w.bucket, v.s3Key); delErr != nil { + slog.Warn("jobs.customer_backup_runner.keep_last_n_s3_delete_failed", + "s3_key", v.s3Key, "error", delErr) + continue + } + if _, updErr := w.db.ExecContext(ctx, ` + UPDATE resource_backups + SET s3_key = NULL, + error_summary = 'retained:count-cap' + WHERE id = $1 + `, v.id); updErr != nil { + slog.Warn("jobs.customer_backup_runner.keep_last_n_db_update_failed", + "backup_id", v.id, "error", updErr) + } + } + if len(victims) > 0 { + slog.Info("jobs.customer_backup_runner.keep_last_n_swept", + "retired", len(victims), "keep", keepHealthyBackupsPerResource) + } +} + // limitedBuffer is a tiny bytes.Buffer wrapper that caps growth at 4KiB so // a chatty pg_dump can't blow up worker RAM with stderr. The error_summary // column is TEXT so we'll truncate at write-site anyway. From e0cc56b331b83d244560526cbb0d77034d422f89 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Wed, 3 Jun 2026 12:12:12 +0530 Subject: [PATCH 3/3] =?UTF-8?q?build(worker):=20bump=20toolchain=20go1.25.?= =?UTF-8?q?10=20=E2=86=92=20go1.25.11=20(stdlib=20CVE=20fixes)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit govulncheck flagged GO-2026-5039 (net/textproto arbitrary input in errors) and GO-2026-5038 (mime WordDecoder quadratic complexity), both fixed in go1.25.11. The version-file-based CI jobs (govulncheck/coverage/codeql/lint) read the go.mod toolchain, so they were pinning the vulnerable 1.25.10. Bumping the toolchain patches the stdlib and unblocks this PR (and all worker PRs). Pre-existing — not introduced by the backup changes in this PR. Co-Authored-By: Claude Opus 4.8 (1M context) --- go.mod | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go.mod b/go.mod index b28fb71..95315cd 100644 --- a/go.mod +++ b/go.mod @@ -2,7 +2,7 @@ module instant.dev/worker go 1.25.0 -toolchain go1.25.10 +toolchain go1.25.11 require ( github.com/DATA-DOG/go-sqlmock v1.5.2