diff --git a/cloud-backups/cmd/pg.go b/cloud-backups/cmd/pg.go
index fdc36f8..c84b60c 100644
--- a/cloud-backups/cmd/pg.go
+++ b/cloud-backups/cmd/pg.go
@@ -26,6 +26,7 @@ func init() {
 	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("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().Int("rotation-interval-days", 0, "audit-rotate: rotate (cut a new archive) only when the newest existing archive is >= N days old, decoupling rotation from the cron cadence so a fast cron (per-minute..daily) still yields ~retention/N coexisting archives. 0 = OFF = rotate every run. Set = --audit-retention-days for a single archive at a time. Must be <= --audit-retention-days. (ENV: ROTATION_INTERVAL_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("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)")
@@ -44,6 +45,7 @@ func init() {
 	mustBindPFlag("pg-schema", "pg-schema")
 	mustBindPFlag("audit-table", "audit-table")
 	mustBindPFlag("audit-retention-days", "audit-retention-days")
+	mustBindPFlag("rotation-interval-days", "rotation-interval-days")
 	mustBindPFlag("lock-timeout", "lock-timeout")
 	mustBindPFlag("allow-unencrypted", "allow-unencrypted")
 	mustBindPFlag("verify-restore", "verify-restore")
diff --git a/cloud-backups/cmd/pg_audit_rotate.go b/cloud-backups/cmd/pg_audit_rotate.go
index 08c9ced..f375c69 100644
--- a/cloud-backups/cmd/pg_audit_rotate.go
+++ b/cloud-backups/cmd/pg_audit_rotate.go
@@ -7,6 +7,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"io"
 	"log/slog"
 	"net"
@@ -76,6 +77,7 @@ func runPGAuditRotate() error {
 		PGSchema:            viper.GetString("pg-schema"),
 		AuditTable:          viper.GetString("audit-table"),
 		RetentionDays:       viper.GetInt("audit-retention-days"),
+		RotationInterval:    viper.GetInt("rotation-interval-days"),
 		LockTimeout:         viper.GetString("lock-timeout"),
 		AllowUnencrypted:    viper.GetBool("allow-unencrypted"),
 		VerifyRestore:       viper.GetBool("verify-restore"),
@@ -216,48 +218,95 @@ func runPGAuditRotate() error {
 		}
 	}
 
-	// Pass 2: rotate -- rename the live table aside and stand up a fresh EMPTY one
-	// (fail-safe on lock contention). The new archive is backed up + verified and then
-	// RETAINED for the retention window; a later run drops it once aged. Exception:
-	// --drain-backlog drops it THIS run, to reclaim the historical backlog immediately
-	// on the one-off cutover run (the recurring cron never sets it).
-	archive, err := newArchiveName(cfg.AuditTable, now)
+	// Rotation gate: decide whether to cut a new archive THIS run. With
+	// rotation-interval-days == 0 (default) we rotate every run (unchanged behavior).
+	// Otherwise rotation is decoupled from the cron cadence -- we rotate only when the
+	// newest existing archive is >= the interval old (or none exists), so a fast cron
+	// reconciles (Pass 1) every run but cuts archives only every interval. --drain-backlog
+	// always rotates (the one-off cutover). Re-query the archive set AFTER Pass 1's drops
+	// so the decision reflects reality. newestSeen feeds the rotate guard below.
+	current, err := pgClient.QueryRows(ctx, listArchivesSQL(cfg.PGSchema, cfg.AuditTable))
 	if err != nil {
+		slog.Error("list_current_archives_failed", "error", err.Error())
 		return err
 	}
-	slog.Info("rotating_audit_table", "schema", cfg.PGSchema, "table", cfg.AuditTable, "archive", archive)
-	if err := pgClient.Exec(ctx, rotateSQL(cfg.PGSchema, cfg.AuditTable, archive, cfg.LockTimeout)); err != nil {
-		slog.Error("rotate_failed_will_retry_next_run", "error", err.Error())
-		return err
-	}
-	if err := backend.BackupAndVerify(ctx, archive, tracker); err != nil {
-		slog.Error("backup_and_verify_failed", "archive", archive, "error", err.Error())
-		return err
-	}
-	if cfg.DrainBacklog {
-		slog.Info("drain_backlog_dropping_new_archive", "schema", cfg.PGSchema, "archive", archive)
-		// deepVerify=false: BackupAndVerify above already did the restore-verify this
-		// run (when --verify-restore), so skip a redundant full re-download.
-		if err := backend.verifyAndDrop(ctx, archive, false); err != nil {
-			slog.Error("drain_backlog_drop_failed", "archive", archive, "error", err.Error())
+	newestSeen, newestRot, haveNewest := newestArchive(current, cfg.AuditTable)
+	newestAgeDays := 0
+	if haveNewest {
+		newestAgeDays = int(now.UTC().Sub(newestRot).Hours() / 24)
+	}
+	rotate, skipReason := rotationDecision(cfg, now, newestRot, haveNewest)
+
+	rotated := false
+	if rotate {
+		// Pass 2: rotate -- rename the live table aside and stand up a fresh EMPTY one
+		// (fail-safe on lock contention). The new archive is backed up + verified and then
+		// RETAINED for the retention window; a later run drops it once aged. Exception:
+		// --drain-backlog drops it THIS run, to reclaim the historical backlog immediately
+		// on the one-off cutover run (the recurring cron never sets it).
+		archive, err := newArchiveName(cfg.AuditTable, now)
+		if err != nil {
 			return err
 		}
-		dropped++
+		slog.Info("rotating_audit_table", "schema", cfg.PGSchema, "table", cfg.AuditTable, "archive", archive)
+		// The rotate transaction is self-guarding against a concurrent rotation (a manual
+		// job racing the cron): it takes a transaction advisory lock and aborts if a newer
+		// archive already exists, so two overlapping runs can't both cut an archive (the
+		// second would otherwise rotate the fresh EMPTY table into a stray archive that
+		// squats for a whole retention window). A guarded abort is a benign skip, not a
+		// failure -- another run already did the rotation.
+		if err := pgClient.Exec(ctx, rotateSQL(cfg.PGSchema, cfg.AuditTable, archive, cfg.LockTimeout, advisoryLockKey(cfg.PGSchema, cfg.AuditTable), newestSeen)); err != nil {
+			if isRotateSkip(err) {
+				slog.Info("rotation_skipped_concurrent", "archive", archive, "reason", "another run rotated concurrently (advisory-lock/supersession guard)")
+				skipReason = "concurrent rotation by another run"
+			} else {
+				slog.Error("rotate_failed_will_retry_next_run", "error", err.Error())
+				return err
+			}
+		} else {
+			rotated = true
+			if err := backend.BackupAndVerify(ctx, archive, tracker); err != nil {
+				slog.Error("backup_and_verify_failed", "archive", archive, "error", err.Error())
+				return err
+			}
+			if cfg.DrainBacklog {
+				slog.Info("drain_backlog_dropping_new_archive", "schema", cfg.PGSchema, "archive", archive)
+				// deepVerify=false: BackupAndVerify above already did the restore-verify this
+				// run (when --verify-restore), so skip a redundant full re-download.
+				if err := backend.verifyAndDrop(ctx, archive, false); err != nil {
+					slog.Error("drain_backlog_drop_failed", "archive", archive, "error", err.Error())
+					return err
+				}
+				dropped++
+			} else {
+				retained++
+			}
+		}
 	} else {
-		retained++
+		slog.Info("rotation_skipped_not_due", "reason", skipReason, "newest_archive_age_days", newestAgeDays, "rotation_interval_days", cfg.RotationInterval)
 	}
 
-	// A single at-a-glance signal for alerting: an operator/monitor can tell a healthy
-	// retain-only run from a "reclaimed nothing / archives piling up" run (e.g. a
-	// clock-skewed pod that never drops). oldest_archive_age_days climbing past
-	// retention_days, or archives_found growing run over run, means retention isn't
-	// keeping up.
+	// A single at-a-glance signal for alerting. rotated_this_run + newest_archive_age_days
+	// let a monitor page on "was due but did not rotate" (rotated_this_run=false AND
+	// newest_archive_age_days >= rotation_interval_days + grace) -- the failure that, under
+	// interval rotation, would otherwise read as healthy (a stalled rotation leaves a young
+	// or zero archive set, so the old oldest>retention signal can't see it). A reconcile-
+	// only run reads clearly as "skipped: not due yet". oldest_archive_age_days is retained
+	// for the retention-health signal (climbing past retention_days = drops not keeping up).
+	rotationSkippedReason := ""
+	if !rotated {
+		rotationSkippedReason = skipReason
+	}
 	slog.Info("audit_rotate_summary",
 		"archives_found", len(leftovers),
 		"archives_recovered", recovered,
 		"archives_dropped", dropped,
 		"archives_retained", retained,
 		"archives_quarantined", quarantined,
+		"rotated_this_run", rotated,
+		"rotation_interval_days", cfg.RotationInterval,
+		"rotation_skipped_reason", rotationSkippedReason,
+		"newest_archive_age_days", newestAgeDays,
 		"oldest_archive_age_days", oldestArchiveAgeDays(leftovers, cfg.AuditTable, now),
 		"retention_days", cfg.RetentionDays)
 
@@ -705,15 +754,71 @@ func agedOut(archive, audit string, now time.Time, retentionDays int) (bool, err
 // dropArchiveSQL drops the archive inside a txn bounded by lock_timeout (so a
 // concurrent ACCESS SHARE holder -- e.g. the full-DB backup's pg_dump -- makes the
 // drop fail fast and defer to next-run recovery, not hang) with no statement_timeout.
+// DROP TABLE IF EXISTS makes a concurrent double-drop benign: two overlapping runs can
+// both list the same aged archive in Pass 1 and both try to drop it; the loser's drop of
+// the already-gone table is a no-op success, not a spurious quarantine + non-zero exit.
 func dropArchiveSQL(schema, archive, lockTimeout string) string {
 	return fmt.Sprintf(`BEGIN;
 SET LOCAL lock_timeout = '%[3]s';
 SET LOCAL statement_timeout = 0;
-DROP TABLE %[1]s.%[2]s;
+DROP TABLE IF EXISTS %[1]s.%[2]s;
 COMMIT;
 `, schema, archive, lockTimeout)
 }
 
+// advisoryLockKey derives the transaction advisory-lock key that serializes rotation for
+// a given audit table (distinct tables don't block each other). Stable across runs and
+// hosts (a pure hash of schema.table), 64-bit to match pg_try_advisory_xact_lock(bigint).
+func advisoryLockKey(schema, audit string) int64 {
+	h := fnv.New64a()
+	_, _ = h.Write([]byte("cloud-backup:audit-rotate:" + schema + "." + audit))
+	return int64(h.Sum64())
+}
+
+// rotateSkipToken marks a rotate transaction that aborted because another run already
+// rotated (advisory-lock contention or a superseding newer archive). It's a benign skip,
+// not a failure -- the caller detects it via isRotateSkip and continues without error.
+const rotateSkipToken = "AUDIT_ROTATE_SKIP"
+
+func isRotateSkip(err error) bool {
+	return err != nil && strings.Contains(err.Error(), rotateSkipToken)
+}
+
+// newestArchive returns the newest (latest rotation time) archive in the set and whether
+// any exists. Names from listArchivesSQL all carry a well-formed timestamp (the SQL regex
+// matches exactly the generated shape), so each parses; the skip-on-parse-error is
+// defensive. Used to decide whether the interval has elapsed since the last rotation.
+func newestArchive(archives []string, audit string) (name string, rot time.Time, ok bool) {
+	for _, a := range archives {
+		t, err := archiveRotationTime(a, audit)
+		if err != nil {
+			continue
+		}
+		if !ok || t.After(rot) {
+			name, rot, ok = a, t, true
+		}
+	}
+	return name, rot, ok
+}
+
+// rotationDecision decides whether Pass 2 cuts a new archive this run. --drain-backlog and
+// the OFF setting (rotation-interval-days == 0) always rotate (today's every-run behavior).
+// Otherwise rotate only when no archive exists (bootstrap / all aged out) OR the newest one
+// is older than the interval -- using the SAME precise cutoff as the retention drop
+// (agedOut). Sharing the threshold is what makes interval == retention hold EXACTLY one
+// archive: on the run the lone archive crosses the line, Pass 1 drops it, so this re-queried
+// set is empty and we rotate a fresh one -- never a transient second archive.
+func rotationDecision(cfg *config.AppConfig, now, newestRot time.Time, haveNewest bool) (rotate bool, skipReason string) {
+	if cfg.DrainBacklog || cfg.RotationInterval == 0 || !haveNewest {
+		return true, ""
+	}
+	cutoff := now.UTC().Add(-time.Duration(cfg.RotationInterval) * 24 * time.Hour)
+	if newestRot.Before(cutoff) {
+		return true, ""
+	}
+	return false, fmt.Sprintf("newest archive rotated %s; rotation interval %dd not yet elapsed", newestRot.UTC().Format(time.RFC3339), cfg.RotationInterval)
+}
+
 // rotateSQL renames the live table aside and creates a fresh identical one in a
 // single transaction. lock_timeout keeps it fail-safe: on contention the whole
 // statement rolls back (table untouched) and the run retries next cycle. The
@@ -728,10 +833,37 @@ COMMIT;
 // (`relation "audit_pkey_..." already exists`); md5(archive) makes it per-archive
 // unique. The renamed names are throwaway (the archive is dropped later); only
 // uniqueness matters. left(name,54)+'_'+8 stays within the 63-byte identifier limit.
-func rotateSQL(schema, audit, archive, lockTimeout string) string {
+//
+// The transaction is self-guarding against a CONCURRENT rotation (a manual job racing
+// the cron, which k8s concurrencyPolicy can't prevent for distinct jobs): it takes a
+// transaction advisory lock (lockKey) and aborts if any archive newer than newestSeen
+// already exists. Without this, two runs that both passed the rotation gate would each
+// rename -- the second renaming the fresh EMPTY table into a stray archive that squats
+// for a whole retention window. Both checks run BEFORE the rename, inside the lock, so
+// they're atomic w.r.t. another rotation; the abort raises rotateSkipToken, which the
+// caller treats as a benign skip. newestSeen is "" when no archive existed at decision
+// time (then any archive => superseded); otherwise it's the newest name the gate saw.
+// The supersession test is lexical (tablename > newestSeen), which equals chronological
+// because the embedded timestamp is fixed-width UTC -- sound unless the wall clock steps
+// backward far enough that a concurrent winner's name sorts below newestSeen (needs
+// multiple rotations inside one wall-second plus a clock step; not operationally reachable,
+// same NTP assumption newArchiveName already notes).
+func rotateSQL(schema, audit, archive, lockTimeout string, lockKey int64, newestSeen string) string {
 	return fmt.Sprintf(`BEGIN;
 SET LOCAL lock_timeout = '%[4]s';
 SET LOCAL statement_timeout = 0;
+DO $GUARD$
+BEGIN
+  IF NOT pg_try_advisory_xact_lock(%[5]d) THEN
+    RAISE EXCEPTION '%[6]s: another rotation holds the advisory lock';
+  END IF;
+  IF EXISTS (SELECT 1 FROM pg_tables WHERE schemaname = '%[1]s'
+             AND tablename ~ '^%[2]s_archive_[0-9]{8}t[0-9]{6}z(_[0-9a-f]+)?$'
+             AND tablename > '%[7]s') THEN
+    RAISE EXCEPTION '%[6]s: a newer archive already exists (superseded by a concurrent run)';
+  END IF;
+END
+$GUARD$;
 ALTER TABLE %[1]s.%[2]s RENAME TO %[3]s;
 DO $ROT$
 DECLARE r record;
@@ -749,7 +881,7 @@ END
 $ROT$;
 CREATE TABLE %[1]s.%[2]s (LIKE %[1]s.%[3]s INCLUDING ALL);
 COMMIT;
-`, schema, audit, archive, lockTimeout)
+`, schema, audit, archive, lockTimeout, lockKey, rotateSkipToken, newestSeen)
 }
 
 func init() {
diff --git a/cloud-backups/cmd/pg_audit_rotate_test.go b/cloud-backups/cmd/pg_audit_rotate_test.go
index 5bd9f7b..9aabf0c 100644
--- a/cloud-backups/cmd/pg_audit_rotate_test.go
+++ b/cloud-backups/cmd/pg_audit_rotate_test.go
@@ -80,18 +80,26 @@ func TestVerifyUploadedObject(t *testing.T) {
 }
 
 func TestRotateSQL(t *testing.T) {
-	got := rotateSQL("rearm", "audit", "audit_archive_20260719t120000z_deadbeef", "5s")
+	got := rotateSQL("rearm", "audit", "audit_archive_20260719t120000z_deadbeef", "5s", 1234567890, "audit_archive_20260601t120000z_old")
 	for _, want := range []string{
 		"SET LOCAL lock_timeout = '5s';",
 		"ALTER TABLE rearm.audit RENAME TO audit_archive_20260719t120000z_deadbeef;",
 		"'rearm.audit_archive_20260719t120000z_deadbeef'::regclass",
 		"CREATE TABLE rearm.audit (LIKE rearm.audit_archive_20260719t120000z_deadbeef INCLUDING ALL);",
 		"COMMIT;",
+		// concurrency guard: advisory lock + supersession check, before the rename
+		"pg_try_advisory_xact_lock(1234567890)",
+		"tablename > 'audit_archive_20260601t120000z_old'",
+		"AUDIT_ROTATE_SKIP",
 	} {
 		if !strings.Contains(got, want) {
 			t.Errorf("rotateSQL missing %q in:\n%s", want, got)
 		}
 	}
+	// the guard must run BEFORE the rename, or a concurrent run could already have renamed
+	if strings.Index(got, "pg_try_advisory_xact_lock") > strings.Index(got, "ALTER TABLE rearm.audit RENAME") {
+		t.Errorf("advisory-lock guard must precede the RENAME in:\n%s", got)
+	}
 }
 
 // The rename-aside suffix MUST derive from the (unique) archive name, not the
@@ -99,8 +107,8 @@ func TestRotateSQL(t *testing.T) {
 // 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")
+	a1 := rotateSQL("rearm", "audit", "audit_archive_20260720t100100z_05196fcc", "5s", 1, "")
+	a2 := rotateSQL("rearm", "audit", "audit_archive_20260720t112820z_d2337e60", "5s", 1, "")
 
 	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)
@@ -133,13 +141,90 @@ func TestListArchivesSQL_Anchored(t *testing.T) {
 
 func TestDropArchiveSQL(t *testing.T) {
 	got := dropArchiveSQL("rearm", "audit_archive_x", "5s")
-	for _, want := range []string{"SET LOCAL lock_timeout = '5s';", "DROP TABLE rearm.audit_archive_x;", "COMMIT;"} {
+	// IF EXISTS makes a concurrent double-drop (two overlapping runs) a benign no-op.
+	for _, want := range []string{"SET LOCAL lock_timeout = '5s';", "DROP TABLE IF EXISTS rearm.audit_archive_x;", "COMMIT;"} {
 		if !strings.Contains(got, want) {
 			t.Errorf("dropArchiveSQL missing %q in:\n%s", want, got)
 		}
 	}
 }
 
+func TestNewestArchive(t *testing.T) {
+	archives := []string{
+		"audit_archive_20260601t120000z_a", // older
+		"audit_archive_20260719t120000z_b", // newest
+		"audit_archive_20260610t120000z_c",
+		"audit_archive_notatimestamp_d", // unparseable -> skipped
+	}
+	name, rot, ok := newestArchive(archives, "audit")
+	if !ok || name != "audit_archive_20260719t120000z_b" {
+		t.Fatalf("newestArchive = %q, %v; want the 07-19 archive", name, ok)
+	}
+	if !rot.Equal(time.Date(2026, 7, 19, 12, 0, 0, 0, time.UTC)) {
+		t.Errorf("rot = %v, want 2026-07-19T12:00:00Z", rot)
+	}
+	if _, _, ok := newestArchive(nil, "audit"); ok {
+		t.Error("newestArchive(nil) should report ok=false")
+	}
+	if _, _, ok := newestArchive([]string{"audit_archive_notatimestamp_x"}, "audit"); ok {
+		t.Error("newestArchive with only unparseable names should report ok=false")
+	}
+}
+
+// rotationDecision is the gate that decouples rotation from cron cadence.
+func TestRotationDecision(t *testing.T) {
+	now := time.Date(2026, 7, 20, 12, 0, 0, 0, time.UTC)
+	newest := func(daysAgo float64) time.Time { return now.Add(-time.Duration(daysAgo*24) * time.Hour) }
+	cases := []struct {
+		name       string
+		cfg        *config.AppConfig
+		haveNewest bool
+		newestRot  time.Time
+		want       bool
+	}{
+		{"interval off (0) -> always rotate", &config.AppConfig{RotationInterval: 0}, true, newest(1), true},
+		{"drain-backlog -> always rotate", &config.AppConfig{RotationInterval: 30, DrainBacklog: true}, true, newest(1), true},
+		{"no archive -> rotate (bootstrap)", &config.AppConfig{RotationInterval: 30}, false, time.Time{}, true},
+		{"newest younger than interval -> skip", &config.AppConfig{RotationInterval: 30}, true, newest(29), false},
+		{"newest older than interval -> rotate", &config.AppConfig{RotationInterval: 30}, true, newest(31), true},
+		{"newest just under interval -> skip", &config.AppConfig{RotationInterval: 14}, true, newest(13.9), false},
+		{"newest just over interval -> rotate", &config.AppConfig{RotationInterval: 14}, true, newest(14.1), true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, reason := rotationDecision(tc.cfg, now, tc.newestRot, tc.haveNewest)
+			if got != tc.want {
+				t.Errorf("rotate = %v, want %v (reason %q)", got, tc.want, reason)
+			}
+			if !got && reason == "" {
+				t.Error("a skip must carry a human-readable reason")
+			}
+		})
+	}
+}
+
+func TestAdvisoryLockKey(t *testing.T) {
+	// stable and table-scoped: same input -> same key; different table -> different key.
+	if advisoryLockKey("rearm", "audit") != advisoryLockKey("rearm", "audit") {
+		t.Error("advisoryLockKey not stable for the same schema.table")
+	}
+	if advisoryLockKey("rearm", "audit") == advisoryLockKey("rearm", "other") {
+		t.Error("advisoryLockKey should differ for a different table (else unrelated rotations block each other)")
+	}
+}
+
+func TestIsRotateSkip(t *testing.T) {
+	if !isRotateSkip(fmt.Errorf("psql exec failed: ERROR: AUDIT_ROTATE_SKIP: superseded")) {
+		t.Error("isRotateSkip should recognize the skip token in a wrapped psql error")
+	}
+	if isRotateSkip(errors.New("some other failure")) {
+		t.Error("isRotateSkip must not match an unrelated error")
+	}
+	if isRotateSkip(nil) {
+		t.Error("isRotateSkip(nil) must be false")
+	}
+}
+
 func TestCountInstancesSQL(t *testing.T) {
 	got := countInstancesSQL("rearm", "audit")
 	for _, want := range []string{"count(*)", "rearm.audit", "entity_name = 'instances'"} {
diff --git a/cloud-backups/internal/config/config.go b/cloud-backups/internal/config/config.go
index 5f6cfec..869c254 100644
--- a/cloud-backups/internal/config/config.go
+++ b/cloud-backups/internal/config/config.go
@@ -41,6 +41,7 @@ type AppConfig struct {
 	PGSchema         string `mapstructure:"pg-schema"`
 	AuditTable       string `mapstructure:"audit-table"`
 	RetentionDays    int    `mapstructure:"audit-retention-days"`
+	RotationInterval int    `mapstructure:"rotation-interval-days"`
 	LockTimeout      string `mapstructure:"lock-timeout"`
 	AllowUnencrypted bool   `mapstructure:"allow-unencrypted"`
 	VerifyRestore    bool   `mapstructure:"verify-restore"`
@@ -170,6 +171,11 @@ func (c *AppConfig) ValidatePGBackup() error {
 }
 
 // ValidatePGAuditRotate checks all fields required for the pg audit-rotate command.
+// maxAuditDays bounds retention / rotation-interval day counts so the "now - N days"
+// cutoff can't overflow an int64 time.Duration (nanoseconds overflow at ~106,751 days /
+// 292 years). 100 years is far beyond any real retention and safely clear of the limit.
+const maxAuditDays = 36500
+
 func (c *AppConfig) ValidatePGAuditRotate() error {
 	if _, err := exec.LookPath("psql"); err != nil {
 		return fmt.Errorf("psql not found in PATH: %w", err)
@@ -192,8 +198,24 @@ func (c *AppConfig) ValidatePGAuditRotate() error {
 	if len(c.AuditTable) > 63-34 {
 		return fmt.Errorf("--audit-table / AUDIT_TABLE too long (%d chars); max 29 so the archive name stays within Postgres's 63-byte identifier limit", len(c.AuditTable))
 	}
-	if c.RetentionDays < 0 {
-		return fmt.Errorf("--audit-retention-days / AUDIT_RETENTION_DAYS must be >= 0, got %d", c.RetentionDays)
+	if c.RetentionDays < 0 || c.RetentionDays > maxAuditDays {
+		return fmt.Errorf("--audit-retention-days / AUDIT_RETENTION_DAYS must be between 0 and %d, got %d", maxAuditDays, c.RetentionDays)
+	}
+	// rotation-interval-days decouples ROTATION cadence from the CRON cadence: the cron
+	// reconciles every run, but a new archive is cut only when the newest existing one is
+	// >= this many days old (0 = OFF = rotate every run, today's behavior). This is what
+	// lets a fast cron (per-minute .. daily) keep ~retention/interval coexisting archives
+	// instead of one per run.
+	if c.RotationInterval < 0 {
+		return fmt.Errorf("--rotation-interval-days / ROTATION_INTERVAL_DAYS must be >= 0, got %d", c.RotationInterval)
+	}
+	// interval > retention is degenerate: the lone archive is dropped at retention age
+	// (before the interval elapses), so the "no archive -> rotate" arm fires and the
+	// effective interval collapses back to retention. Reject rather than silently mislead.
+	// interval <= retention (which is itself capped at maxAuditDays above) also bounds the
+	// "now - interval days" cutoff away from int64 time.Duration overflow, so no separate cap.
+	if c.RotationInterval > c.RetentionDays {
+		return fmt.Errorf("--rotation-interval-days / ROTATION_INTERVAL_DAYS (%d) must be <= --audit-retention-days (%d): a larger interval is degenerate (retention drops the archive before the interval elapses)", c.RotationInterval, c.RetentionDays)
 	}
 	if !pgDuration.MatchString(c.LockTimeout) {
 		return fmt.Errorf("--lock-timeout / LOCK_TIMEOUT must be a PostgreSQL duration like 5s or 500ms, got %q", c.LockTimeout)
diff --git a/cloud-backups/internal/config/config_test.go b/cloud-backups/internal/config/config_test.go
index d3983d9..0deaf55 100644
--- a/cloud-backups/internal/config/config_test.go
+++ b/cloud-backups/internal/config/config_test.go
@@ -466,13 +466,22 @@ func TestValidatePGAuditRotate(t *testing.T) {
 		t.Errorf("explicit --allow-unencrypted should pass: %v", err)
 	}
 	bad := map[string]func(*AppConfig){
-		"bad schema":    func(c *AppConfig) { c.PGSchema = "rea rm" },
-		"bad table":     func(c *AppConfig) { c.AuditTable = "audit;drop" },
-		"long table":    func(c *AppConfig) { c.AuditTable = "a_very_long_audit_table_name_wont_fit" },
-		"neg retention": func(c *AppConfig) { c.RetentionDays = -1 },
-		"empty lock":    func(c *AppConfig) { c.LockTimeout = "" },
-		"missing db":    func(c *AppConfig) { c.PGDatabase = "" },
-		"missing buck":  func(c *AppConfig) { c.AWSBucket = "" },
+		"bad schema":           func(c *AppConfig) { c.PGSchema = "rea rm" },
+		"bad table":            func(c *AppConfig) { c.AuditTable = "audit;drop" },
+		"long table":           func(c *AppConfig) { c.AuditTable = "a_very_long_audit_table_name_wont_fit" },
+		"neg retention":        func(c *AppConfig) { c.RetentionDays = -1 },
+		"huge retention":       func(c *AppConfig) { c.RetentionDays = maxAuditDays + 1 },
+		"neg interval":         func(c *AppConfig) { c.RotationInterval = -1 },
+		"interval > retention": func(c *AppConfig) { c.RotationInterval = 31 }, // retention is 30
+		"empty lock":           func(c *AppConfig) { c.LockTimeout = "" },
+		"missing db":           func(c *AppConfig) { c.PGDatabase = "" },
+		"missing buck":         func(c *AppConfig) { c.AWSBucket = "" },
+	}
+	// valid interval settings: off (0) and == retention (single-archive)
+	for _, iv := range []int{0, 15, 30} {
+		if err := (func() *AppConfig { c := base(); c.RotationInterval = iv; return c })().ValidatePGAuditRotate(); err != nil {
+			t.Errorf("rotation-interval-days=%d (<= retention) should pass: %v", iv, err)
+		}
 	}
 	for name, mut := range bad {
 		t.Run(name, func(t *testing.T) {