Skip to content

feat(cloud-backups): retention-gated drop for pg audit-rotate #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
logicflakes merged 1 commit into from
Jul 21, 2026
Merged

feat(cloud-backups): retention-gated drop for pg audit-rotate #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 9 additions & 7 deletions cloud-backups/cmd/pg.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,16 +20,17 @@ func init() {
pgCmd.PersistentFlags().String("pg-port", "5432", "PostgreSQL port (ENV: PG_PORT)")
pgCmd.PersistentFlags().String("pg-database", "", "Database name (ENV: PG_DATABASE)")
pgCmd.PersistentFlags().String("pg-user", "", "PostgreSQL username (ENV: PG_USER)")
pgCmd.PersistentFlags().String("exclude-table", "", "pg backup: comma-separated pg_dump --exclude-table patterns (wildcards ok, e.g. 'rearm.audit_archive_*') to omit from a whole-DB backup -- e.g. the retained audit archive tables, which have their own permanent-bucket backups (ENV: EXCLUDE_TABLE)")

// audit-rotate specific flags
pgCmd.PersistentFlags().String("pg-schema", "rearm", "Schema containing the audit table, for audit-rotate (ENV: PG_SCHEMA)")
pgCmd.PersistentFlags().String("audit-table", "audit", "Audit table name, for audit-rotate (ENV: AUDIT_TABLE)")
pgCmd.PersistentFlags().Int("keep-tail-days", 0, "audit-rotate: also keep audit rows newer than N days in the live table (0 = readers only) (ENV: KEEP_TAIL_DAYS)")
pgCmd.PersistentFlags().Int("audit-retention-days", 30, "audit-rotate: keep each sealed archive on disk (queryable by name for ops inspection) until it is older than N days, then DROP it whole on a later run (0 = drop on the next run) (ENV: AUDIT_RETENTION_DAYS)")
pgCmd.PersistentFlags().String("lock-timeout", "5s", "audit-rotate: lock_timeout for the rename step; on contention the rotate rolls back and retries next run (ENV: LOCK_TIMEOUT)")
pgCmd.PersistentFlags().Bool("allow-unencrypted", false, "audit-rotate: allow writing an UNENCRYPTED dump to the permanent bucket when no --encryption-password is set (ENV: ALLOW_UNENCRYPTED)")
pgCmd.PersistentFlags().Bool("no-drop", false, "audit-rotate: rotate + back up + verify, but do NOT drop the archive (leave it for manual confirmation, then a later run drops it) (ENV: NO_DROP)")
pgCmd.PersistentFlags().Bool("verify-restore", false, "audit-rotate: before dropping, re-download the archive, decrypt it, run pg_restore -l (proves it's a restorable dump), and match its SHA-256 (full re-download) (ENV: VERIFY_RESTORE)")
pgCmd.PersistentFlags().Bool("drop-pending", false, "audit-rotate: do NOT rotate; instead verify each already-backed-up leftover archive against its stored .sha256 sidecar (+ pg_restore -l) and drop it. The confirm step after a --no-drop run. (ENV: DROP_PENDING)")
pgCmd.PersistentFlags().Bool("verify-restore", false, "audit-rotate: before an aged-out drop, re-download the archive, decrypt it, run pg_restore -l (proves it's a restorable dump), and match its SHA-256 (full re-download). Default is a cheap existence gate (ENV: VERIFY_RESTORE)")
pgCmd.PersistentFlags().Bool("drain-backlog", false, "audit-rotate: back up + drop the archive created THIS run immediately, regardless of retention age. Set only on the first/cutover run to reclaim the historical backlog now; retention accumulates from the next run. Keep false for the recurring cron. (ENV: DRAIN_BACKLOG)")
pgCmd.PersistentFlags().Bool("drop-instance-rows", false, "audit-rotate: proceed even if the audit table holds frozen entity_name='instances' rows (still read by the app but never re-written). Without this the run refuses when such rows exist. Setting it does NOT lose data (the rows are backed up to the permanent bucket like any archive) but the app's instance-revision reads return empty once those rows age out of the DB -- a conscious cutover choice. (ENV: DROP_INSTANCE_ROWS)")

mustBindPFlag := func(key, flagName string) {
if err := viper.BindPFlag(key, pgCmd.PersistentFlags().Lookup(flagName)); err != nil {
Expand All @@ -42,10 +43,11 @@ func init() {
mustBindPFlag("pg-user", "pg-user")
mustBindPFlag("pg-schema", "pg-schema")
mustBindPFlag("audit-table", "audit-table")
mustBindPFlag("keep-tail-days", "keep-tail-days")
mustBindPFlag("audit-retention-days", "audit-retention-days")
mustBindPFlag("lock-timeout", "lock-timeout")
mustBindPFlag("allow-unencrypted", "allow-unencrypted")
mustBindPFlag("no-drop", "no-drop")
mustBindPFlag("verify-restore", "verify-restore")
mustBindPFlag("drop-pending", "drop-pending")
mustBindPFlag("drain-backlog", "drain-backlog")
mustBindPFlag("drop-instance-rows", "drop-instance-rows")
mustBindPFlag("exclude-table", "exclude-table")
}
482 changes: 287 additions & 195 deletions cloud-backups/cmd/pg_audit_rotate.go
View file
Open in desktop

Large diffs are not rendered by default.

246 changes: 157 additions & 89 deletions cloud-backups/cmd/pg_audit_rotate_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,23 +3,26 @@ package cmd
import (
"context"
"errors"
"fmt"
"io"
"strings"
"testing"
"time"

"github.com/relizaio/cloud-backup/internal/config"
"github.com/relizaio/cloud-backup/internal/stats"
"github.com/relizaio/cloud-backup/internal/storage"
)

// fakeStore is a storage.Provider that records sidecar uploads and returns a
// configurable Head, for unit-testing the post-upload verification gate.
// fakeStore is a storage.Provider for unit-testing the post-upload verification and
// the pre-drop gate. If objects is non-nil, Head answers per-key (present -> size,
// absent -> ErrNotFound), which the drop-gate tests use. Otherwise it falls back to
// the single headSize/headErr (the verifyUploadedObject test).
type fakeStore struct {
headSize int64
headErr error
uploadErr error
uploaded map[string]bool
objects map[string]int64
}

func (f *fakeStore) UploadStream(_ context.Context, path string, r io.Reader) error {
Expand All @@ -33,7 +36,13 @@ func (f *fakeStore) UploadStream(_ context.Context, path string, r io.Reader) er
return f.uploadErr
}
func (f *fakeStore) DownloadStream(_ context.Context, _ string, _ io.Writer) error { return nil }
func (f *fakeStore) Head(_ context.Context, _ string) (*storage.ObjectInfo, error) {
func (f *fakeStore) Head(_ context.Context, path string) (*storage.ObjectInfo, error) {
if f.objects != nil {
if sz, ok := f.objects[path]; ok {
return &storage.ObjectInfo{Size: sz}, nil
}
return nil, fmt.Errorf("head %q: %w", path, storage.ErrNotFound)
}
if f.headErr != nil {
return nil, f.headErr
}
Expand Down Expand Up @@ -87,22 +96,18 @@ func TestRotateSQL(t *testing.T) {

// The rename-aside suffix MUST derive from the (unique) archive name, not the
// constant original constraint/index name -- otherwise two coexisting un-dropped
// archives (e.g. a --no-drop staging re-run) collide on `audit_pkey_<sfx>` and the
// second rotation fails with `relation "audit_pkey_..." already exists`.
// archives (the norm under retention) collide on `audit_pkey_<sfx>` and the second
// rotation fails with `relation "audit_pkey_..." already exists`.
func TestRotateSQL_RenameSuffixIsPerArchive(t *testing.T) {
a1 := rotateSQL("rearm", "audit", "audit_archive_20260720t100100z_05196fcc", "5s")
a2 := rotateSQL("rearm", "audit", "audit_archive_20260720t112820z_d2337e60", "5s")

// The suffix is md5(archive-name), so it must reference the archive name, not
// the original constraint name.
if !strings.Contains(a1, "substr(md5('audit_archive_20260720t100100z_05196fcc'), 1, 8)") {
t.Errorf("rename suffix not derived from archive name in:\n%s", a1)
}
if strings.Contains(a1, "md5(r.conname)") || strings.Contains(a1, "md5(r.relname)") {
t.Errorf("rename suffix still derived from the (constant) constraint/index name:\n%s", a1)
}
// Two different archives must produce different DECLARE-d suffixes so their
// renamed constraints/indexes never collide schema-wide.
sfx := func(s string) string {
const marker = "DECLARE sfx text := "
i := strings.Index(s, marker)
Expand All @@ -116,33 +121,6 @@ func TestRotateSQL_RenameSuffixIsPerArchive(t *testing.T) {
}
}

func TestKeepCopySQL_InstancesOnly(t *testing.T) {
cols := []string{"uuid", "entity_name", "revision_record_data"}
got := keepCopySQL("rearm", "audit", "audit_archive_x", 0, "5s", cols)
for _, want := range []string{
"SET statement_timeout = 0;",
"SET lock_timeout = '5s';",
`INSERT INTO rearm.audit ("uuid", "entity_name", "revision_record_data")`,
`SELECT "uuid", "entity_name", "revision_record_data" FROM rearm.audit_archive_x`,
"entity_name = 'instances'",
"ON CONFLICT DO NOTHING;",
} {
if !strings.Contains(got, want) {
t.Errorf("keepCopySQL(0) missing %q in:\n%s", want, got)
}
}
if strings.Contains(got, "revision_created_date") {
t.Errorf("keepCopySQL(0) should be INSTANCES-only, but references revision_created_date:\n%s", got)
}
}

func TestKeepCopySQL_WithTail(t *testing.T) {
got := keepCopySQL("rearm", "audit", "a", 30, "5s", []string{"uuid"})
if !strings.Contains(got, "make_interval(days => 30)") {
t.Errorf("keepCopySQL(30) did not honor keepTailDays:\n%s", got)
}
}

func TestListArchivesSQL_Anchored(t *testing.T) {
got := listArchivesSQL("rearm", "audit")
if !strings.Contains(got, `tablename ~ '^audit_archive_[0-9]{8}t[0-9]{6}z(_[0-9a-f]+)?$'`) {
Expand All @@ -162,15 +140,39 @@ func TestDropArchiveSQL(t *testing.T) {
}
}

func TestAssertHasUniqueSQL(t *testing.T) {
got := assertHasUniqueSQL("rearm", "audit")
for _, want := range []string{"table_constraints", "table_name = 'audit'", "'PRIMARY KEY', 'UNIQUE'"} {
func TestCountInstancesSQL(t *testing.T) {
got := countInstancesSQL("rearm", "audit")
for _, want := range []string{"count(*)", "rearm.audit", "entity_name = 'instances'"} {
if !strings.Contains(got, want) {
t.Errorf("assertHasUniqueSQL missing %q in:\n%s", want, got)
t.Errorf("countInstancesSQL missing %q in:\n%s", want, got)
}
}
}

func TestNonOwnerGrantsSQL(t *testing.T) {
got := nonOwnerGrantsSQL("rearm", "audit")
for _, want := range []string{"aclexplode(c.relacl)", "'rearm.audit'::regclass", "acl.grantee <> c.relowner", "'PUBLIC'"} {
if !strings.Contains(got, want) {
t.Errorf("nonOwnerGrantsSQL missing %q in:\n%s", want, got)
}
}
}

func TestOldestArchiveAgeDays(t *testing.T) {
now := time.Date(2026, 7, 20, 12, 0, 0, 0, time.UTC)
archives := []string{
"audit_archive_20260719t120000z_a", // 1 day
"audit_archive_20260601t120000z_b", // 49 days
"audit_archive_notatimestamp_c", // unparseable -> skipped
}
if got := oldestArchiveAgeDays(archives, "audit", now); got != 49 {
t.Errorf("oldestArchiveAgeDays = %d, want 49", got)
}
if got := oldestArchiveAgeDays(nil, "audit", now); got != 0 {
t.Errorf("oldestArchiveAgeDays(nil) = %d, want 0", got)
}
}

func TestNewArchiveName(t *testing.T) {
ts := time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC)
a, err := newArchiveName("audit", ts)
Expand All @@ -185,68 +187,134 @@ func TestNewArchiveName(t *testing.T) {
}
}

// --- drop-gate: the one irreversible step, unit-tested via the archiveBackend seam ---
// --- the retention drop-gate oracle: parsed from the name, must fail SAFE ---

type fakeBackend struct {
cols []string
queryErr error
execErr error
backupErr error // BackupAndVerify result: nil = fully verified
execs []string
func TestArchiveRotationTime(t *testing.T) {
cases := []struct {
name string
archive string
want time.Time
wantErr bool
}{
{"valid", "audit_archive_20260719t120000z_deadbeef", time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC), false},
{"valid no hex suffix", "audit_archive_20260719t120000z", time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC), false},
{"not an archive name", "audit", time.Time{}, true},
{"wrong prefix", "other_archive_20260719t120000z_x", time.Time{}, true},
{"impossible date (month 13) -> parse error, NOT zero-time-as-ancient", "audit_archive_20261301t120000z_x", time.Time{}, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got, err := archiveRotationTime(tc.archive, "audit")
if (err != nil) != tc.wantErr {
t.Fatalf("err = %v, wantErr = %v", err, tc.wantErr)
}
if !tc.wantErr && !got.Equal(tc.want) {
t.Errorf("got %v, want %v", got, tc.want)
}
})
}
}

func (f *fakeBackend) QueryRows(_ context.Context, _ string) ([]string, error) {
return f.cols, f.queryErr
}
func (f *fakeBackend) Exec(_ context.Context, sql string) error {
f.execs = append(f.execs, sql)
return f.execErr
}
func (f *fakeBackend) BackupAndVerify(_ context.Context, _ string, _ *stats.Tracker) error {
return f.backupErr
}
func (f *fakeBackend) dropped() bool {
for _, s := range f.execs {
if strings.Contains(s, "DROP TABLE") {
return true
}
func TestAgedOut(t *testing.T) {
now := time.Date(2026, 7, 20, 12, 0, 0, 0, time.UTC)
cases := []struct {
name string
archive string
retentionDays int
wantAged bool
wantErr bool
}{
{"1 day old, 30d window -> retain", "audit_archive_20260719t120000z_a", 30, false, false},
{"49 days old, 30d window -> aged", "audit_archive_20260601t120000z_a", 30, true, false},
{"exactly at boundary (30d ago) -> not yet strictly past -> retain", "audit_archive_20260620t120000z_a", 30, false, false},
{"retention 0 -> any prior archive aged", "audit_archive_20260719t120000z_a", 0, true, false},
{"unparseable name -> error, caller must not drop", "audit_archive_99999999t999999z_a", 30, false, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
aged, err := agedOut(tc.archive, "audit", now, tc.retentionDays)
if (err != nil) != tc.wantErr {
t.Fatalf("err = %v, wantErr = %v", err, tc.wantErr)
}
if !tc.wantErr && aged != tc.wantAged {
t.Errorf("aged = %v, want %v", aged, tc.wantAged)
}
})
}
return false
}

func TestBackupAndDropArchive_Gate(t *testing.T) {
// --- the pre-drop gate: the decision guarding the sole irreversible step ---

func TestBackupIsDroppable_CheapGate(t *testing.T) {
const archive = "audit_archive_20260720t100100z_dead"
// cfg without encryption -> suffix ".dump"
cfg := &config.AppConfig{PGSchema: "rearm", DumpPrefix: "p"}
dumpKey := "p-" + archive + ".dump"
sidecarKey := dumpKey + ".sha256"

cases := []struct {
name string
cols []string
queryErr error
backupErr error
noDrop bool
wantErr bool
wantDropped bool
name string
objects map[string]int64
wantErr bool
}{
{"verified backup -> drop", []string{"uuid"}, nil, nil, false, false, true},
{"backup/verify failed -> no drop", []string{"uuid"}, nil, errors.New("upload failed"), false, true, false},
{"verified but --no-drop -> no drop, no error", []string{"uuid"}, nil, nil, true, false, false},
{"empty shared columns -> no drop, no backup", nil, nil, nil, false, true, false},
{"query error -> no drop", []string{"uuid"}, errors.New("boom"), nil, false, true, false},
{"dump + sidecar present -> droppable", map[string]int64{dumpKey: 100, sidecarKey: 65}, false},
{"dump present, sidecar absent -> NOT droppable", map[string]int64{dumpKey: 100}, true},
{"dump absent -> NOT droppable", map[string]int64{sidecarKey: 65}, true},
{"nothing present -> NOT droppable", map[string]int64{}, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
cfg := &config.AppConfig{PGSchema: "rearm", AuditTable: "audit", LockTimeout: "5s", NoDrop: tc.noDrop}
b := &fakeBackend{cols: tc.cols, queryErr: tc.queryErr, backupErr: tc.backupErr}
err := backupAndDropArchive(context.Background(), b, cfg, "audit_archive_x", stats.New())
b := &pgArchiveBackend{store: &fakeStore{objects: tc.objects}, cfg: cfg}
err := b.backupIsDroppable(context.Background(), archive, false)
if (err != nil) != tc.wantErr {
t.Errorf("err = %v, wantErr = %v", err, tc.wantErr)
}
if b.dropped() != tc.wantDropped {
t.Errorf("dropped = %v, want %v (execs: %v)", b.dropped(), tc.wantDropped, b.execs)
}
// When a drop happens, keep-copy must have run first.
if tc.wantDropped {
if len(b.execs) < 2 || !strings.Contains(b.execs[0], "INSERT INTO") || !strings.Contains(b.execs[len(b.execs)-1], "DROP TABLE") {
t.Errorf("keep-copy must precede drop; execs: %v", b.execs)
}
}
})
}
}

// A transient (non-NotFound) Head error must NOT be read as "safe to drop".
func TestBackupIsDroppable_TransientHeadErrorDoesNotDrop(t *testing.T) {
cfg := &config.AppConfig{PGSchema: "rearm", DumpPrefix: "p"}
b := &pgArchiveBackend{store: &fakeStore{headErr: errors.New("throttled")}, cfg: cfg}
if err := b.backupIsDroppable(context.Background(), "audit_archive_20260720t100100z_dead", false); err == nil {
t.Error("transient Head error must fail the gate (not droppable), got nil")
}
}

// hasBackup keys on the sidecar (written last) and must map a definitive ErrNotFound
// to "not backed up" while propagating a transient error.
func TestHasBackup(t *testing.T) {
const archive = "audit_archive_20260720t100100z_dead"
cfg := &config.AppConfig{PGSchema: "rearm", DumpPrefix: "p"}
dumpKey := "p-" + archive + ".dump"
sidecarKey := dumpKey + ".sha256"

t.Run("dump + sidecar present -> backed up", func(t *testing.T) {
b := &pgArchiveBackend{store: &fakeStore{objects: map[string]int64{dumpKey: 100, sidecarKey: 65}}, cfg: cfg}
ok, err := b.hasBackup(context.Background(), archive)
if err != nil || !ok {
t.Errorf("ok=%v err=%v, want true,nil", ok, err)
}
})
t.Run("sidecar present but dump missing -> not backed up (self-heal re-dump)", func(t *testing.T) {
b := &pgArchiveBackend{store: &fakeStore{objects: map[string]int64{sidecarKey: 65}}, cfg: cfg}
ok, err := b.hasBackup(context.Background(), archive)
if err != nil || ok {
t.Errorf("ok=%v err=%v, want false,nil", ok, err)
}
})
t.Run("both absent -> not backed up, no error", func(t *testing.T) {
b := &pgArchiveBackend{store: &fakeStore{objects: map[string]int64{}}, cfg: cfg}
ok, err := b.hasBackup(context.Background(), archive)
if err != nil || ok {
t.Errorf("ok=%v err=%v, want false,nil", ok, err)
}
})
t.Run("transient error -> propagated, not treated as absence", func(t *testing.T) {
b := &pgArchiveBackend{store: &fakeStore{headErr: errors.New("throttled")}, cfg: cfg}
if _, err := b.hasBackup(context.Background(), archive); err == nil {
t.Error("transient Head error must propagate, got nil")
}
})
}
10 changes: 10 additions & 0 deletions cloud-backups/cmd/pg_backup.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"net"
"os"
"os/signal"
"strings"
"syscall"
"time"

Expand Down Expand Up @@ -49,6 +50,7 @@ func runPGBackup() error {
PGPort: pgPort,
PGDatabase: viper.GetString("pg-database"),
PGUser: viper.GetString("pg-user"),
ExcludeTable: viper.GetString("exclude-table"),
StorageType: viper.GetString("backup-storage-type"),
EncryptionPassword: viper.GetString("encryption-password"),
DumpPrefix: viper.GetString("dump-prefix"),
Expand Down Expand Up @@ -99,6 +101,14 @@ func runPGBackup() error {
Database: cfg.PGDatabase,
User: cfg.PGUser,
}
// Comma-separated --exclude-table patterns (e.g. the retained audit archive
// tables, which have their own permanent-bucket backups) omitted from the
// whole-DB dump.
for _, pat := range strings.Split(cfg.ExcludeTable, ",") {
if pat = strings.TrimSpace(pat); pat != "" {
pgClient.ExcludeTables = append(pgClient.ExcludeTables, pat)
}
}
slog.Info("running_preflight_check", "host", cfg.PGHost, "port", cfg.PGPort)
if err := pgClient.PreflightCheck(ctx, cfg.PGDatabase); err != nil {
slog.Error("preflight_check_failed", "error", err.Error())
Expand Down
Loading
Loading