Backport #509 and #512 to 0.6: replica identity and prune truncating only used children

CHANGELOG.md

@@ -4,8 +4,17 @@ Notable changes between releases. Detailed migration notes for storage transitio
 
 ## [Unreleased]
 
+### Upgrading
+
+- The schema stays at v40. `awa migrate` applies one new repeatable patch, `replica_identity`. Unlike the earlier 0.6 patches it takes table locks: `ACCESS EXCLUSIVE` on each table it changes (`awa.admin_dirty_queue_marks`, `awa.admin_dirty_kind_marks`, `awa.job_unique_claims` and each `queue_terminal_count_deltas` table and partition), plus `SHARE` on `idx_awa_jobs_unique`, held until the patch commits, which is milliseconds once granted. Each lock wait is bounded by the session's `lock_timeout`, or 300 ms when none is set; a timeout fails the patch with `55P03` and changes nothing, so run `awa migrate` again. A re-run after success takes no lock. External migration runners apply `R__replica_identity.sql` after `V40` and after the other `R__` scripts, as the owner of the awa and queue-storage tables. 0.6.17 binaries run unchanged on the patched schema, so binaries and the patch can roll in either order. See the [upgrade guide](docs/upgrade-0.6.17-to-0.6.18.md).
+
+### Changed
+
+- Ring prune truncates only the children of the sealed slot that hold storage (0.6 port of #512). Queue, lease and claim prune take the same locks on every child of the slot, in the same order, as in 0.6.17, then `TRUNCATE` only the children with heap, TOAST, free-space-map or visibility-map pages, or index pages beyond an empty btree, GiST or BRIN index. A child that never held a row keeps its relfilenode, so a prune no longer swaps relfilenodes and writes WAL and catalog rows for empty partitions, such as the receipt children of a schema without receipts. A child emptied by `DELETE` and `VACUUM` that kept expanded indexes or TOAST is still truncated. The empty-slot fast path is unchanged. No configuration or schema change.
+
 ### Fixed
 
+- **awa keeps working when its schema is in a logical-replication publication (0.6 port of #509).** With a `CREATE PUBLICATION ... FOR ALL TABLES` (Debezium's default) or `FOR TABLES IN SCHEMA awa`, PostgreSQL rejected every `UPDATE` and `DELETE` on the 19 awa tables without a primary key, so the admin dirty-mark drain failed on every maintenance tick with `cannot delete from table "admin_dirty_queue_marks" because it does not have a replica identity and publishes deletes`, and unique-claim release failed the same way. The `replica_identity` patch gives `awa.job_unique_claims` its uniqueness index as replica identity and sets `REPLICA IDENTITY FULL` on the two mark tables and on `queue_terminal_count_deltas` and each of its partitions, which adds no WAL on their insert-and-truncate path. PostgreSQL does not copy the setting to partitions, so the queue-storage installer sets it on every partition of a schema prepared later. Keeping the awa schema out of CDC publications remains the recommendation; see [logical replication and CDC](docs/deploying-on-managed-postgres.md#logical-replication-and-cdc).
 - `JobContext::flush_progress` (Python `ctx.flush_progress()`) on a queue-storage attempt with receipts, the default, no longer returns success without writing the progress. A heartbeat leaves out an attempt whose lock another transaction holds, and since 0.6.16 an explicit flush shared that path, so a flush that overlapped a heartbeat or a completion of the same attempt was dropped and the worker treated the progress as persisted. An explicit flush now waits for the attempt's lock; heartbeats still skip.
 
 ## [0.6.17] — 2026-11-12

README.md

@@ -392,6 +392,7 @@ All coordination through Postgres. The Rust runtime owns dispatch, leases, heart
 | [0.6.14 → 0.6.15 upgrade](docs/upgrade-0.6.14-to-0.6.15.md) | Rolling patch upgrade: bounded worker finalization, the claim batch index and the batch compatibility insert |
 | [0.6.15 → 0.6.16 upgrade](docs/upgrade-0.6.15-to-0.6.16.md) | Rolling patch upgrade: bounded rescue drains of abandoned claims, batch compatibility inserts that give way to per-row producers, new-lane provisioning order, the operator prune halt and its grants, and claim batch index fixes |
 | [0.6.16 → 0.6.17 upgrade](docs/upgrade-0.6.16-to-0.6.17.md) | Rolling patch upgrade: asynchronous claim commits, the maintenance eligibility table, the claim tombstone patch, the changed `/api/stats` and `state_counts` shape, and prune lock waits |
+| [0.6.17 → 0.6.18 upgrade](docs/upgrade-0.6.17-to-0.6.18.md) | Rolling patch upgrade: the replica identity patch and the table locks it takes, and prune truncating only partitions that hold rows |
 | [Configuration reference](docs/configuration.md) | `QueueConfig`, `ClientBuilder`, Python `start()`, env vars |
 | [HTTP workers and callbacks](docs/http-callbacks.md) | Serverless dispatch, callback receiver endpoints, signature verification |
 | [Lifecycle hooks](docs/lifecycle-hooks.md) | React to job start/finish/retry/callback events for metrics, alerting, cleanup |

awa-model/migrations/patches/replica_identity.sql

@@ -0,0 +1,91 @@
+-- Schema patch replica_identity: give every awa table a replica identity
+-- (awa 0.7 migration v047, minus the schema_version row).
+--
+-- PostgreSQL rejects UPDATE and DELETE on a table that is in a logical
+-- replication publication and has no primary key, replica-identity index or
+-- FULL identity. Tables without one on 0.6:
+--   * awa.admin_dirty_queue_marks, awa.admin_dirty_kind_marks: FULL. ADR-045
+--     forbids any index or constraint on them.
+--   * awa.job_unique_claims: USING INDEX idx_awa_jobs_unique.
+--   * <schema>.queue_terminal_count_deltas and each of its partitions, in
+--     every queue-storage schema: FULL. Partitions do not inherit the
+--     parent's setting.
+-- Tracking: https://github.com/hardbyte/awa/pull/509
+--
+-- Applied after the v023 definitions, whose installer sets the delta tables'
+-- identity in schemas prepared later.
+--
+-- Live load: each ALTER takes ACCESS EXCLUSIVE on its table (and SHARE on
+-- idx_awa_jobs_unique), held until the patch's transaction commits. Each lock
+-- wait is bounded by the session's lock_timeout, or 300ms when none is set;
+-- a timeout aborts the patch and leaves nothing changed. A table that already
+-- has its identity is not altered, so a re-run locks nothing.
+--
+-- 0.6 delivery: the 0.6 series cannot add a migration version, so this ships as
+-- an idempotent schema patch. `awa migrate` applies it when the schema is
+-- exactly at v040 and the probe in migrations.rs reports it missing;
+-- `awa migrate --sql` prints it as a repeatable R__ script for external
+-- runners. Replica identity only changes what the WAL records for UPDATE and
+-- DELETE, so released 0.6 binaries run the same against a patched or an
+-- unpatched schema. Run it as the owner of the awa and queue-storage tables.
+
+DO $$
+DECLARE
+    v_lock_timeout TEXT := current_setting('lock_timeout');
+    v_target RECORD;
+BEGIN
+    IF v_lock_timeout = '0' THEN
+        PERFORM set_config('lock_timeout', '300ms', true);
+    END IF;
+
+    FOR v_target IN
+        SELECT c.oid::regclass AS relation, 'FULL' AS identity, 0 AS ord,
+               n.nspname, c.relkind, c.relname
+        FROM pg_class AS c
+        JOIN pg_namespace AS n ON n.oid = c.relnamespace
+        WHERE n.nspname = 'awa'
+          AND c.relname IN ('admin_dirty_queue_marks', 'admin_dirty_kind_marks')
+          AND c.relkind = 'r'
+          AND c.relreplident <> 'f'
+        UNION ALL
+        SELECT c.oid::regclass, 'USING INDEX idx_awa_jobs_unique', 1,
+               n.nspname, c.relkind, c.relname
+        FROM pg_class AS c
+        JOIN pg_namespace AS n ON n.oid = c.relnamespace
+        WHERE n.nspname = 'awa'
+          AND c.relname = 'job_unique_claims'
+          AND to_regclass('awa.idx_awa_jobs_unique') IS NOT NULL
+          AND NOT EXISTS (
+              SELECT 1 FROM pg_index AS i
+              WHERE i.indrelid = c.oid AND i.indisreplident
+          )
+        UNION ALL
+        SELECT c.oid::regclass, 'FULL', 2, n.nspname, c.relkind, c.relname
+        FROM pg_class AS c
+        JOIN pg_namespace AS n ON n.oid = c.relnamespace
+        WHERE has_schema_privilege(current_user, n.oid, 'USAGE')
+          AND EXISTS (
+              SELECT 1 FROM pg_proc AS awa_p
+              WHERE awa_p.pronamespace = n.oid
+                AND awa_p.proname = 'claim_ready_runtime'
+                AND oidvectortypes(awa_p.proargtypes)
+                    = 'text, bigint, double precision, double precision'
+          )
+          AND c.relkind IN ('r', 'p')
+          AND c.relreplident <> 'f'
+          AND c.relname ~ '^queue_terminal_count_deltas(_[0-9]+)?$'
+        ORDER BY ord, nspname, relkind, relname
+    LOOP
+        BEGIN
+            EXECUTE format('ALTER TABLE %s REPLICA IDENTITY %s', v_target.relation, v_target.identity);
+        EXCEPTION WHEN lock_not_available THEN
+            RAISE EXCEPTION 'replica_identity: could not lock % within lock_timeout %',
+                v_target.relation, current_setting('lock_timeout')
+                USING ERRCODE = 'lock_not_available',
+                      HINT = 'Nothing was changed. Run awa migrate again, or raise lock_timeout for the migrating session.';
+        END;
+    END LOOP;
+
+    PERFORM set_config('lock_timeout', v_lock_timeout, true);
+END;
+$$;

awa-model/migrations/patches/wait_free_dirty_marks.sql

@@ -29,6 +29,37 @@ CREATE TABLE IF NOT EXISTS awa.admin_dirty_kind_marks (
     touched_at TIMESTAMPTZ NOT NULL DEFAULT now()
 );
 
+-- A mark table created by this transaction gets REPLICA IDENTITY FULL here,
+-- as the replica_identity patch would give it, whichever order a migration
+-- runner applies the two in. Only a table this session already holds in
+-- ACCESS EXCLUSIVE mode is altered, so the change never waits for a lock.
+DO $$
+DECLARE
+    v_table REGCLASS;
+BEGIN
+    FOR v_table IN
+        SELECT c.oid::regclass
+        FROM pg_class AS c
+        WHERE c.oid IN (
+                to_regclass('awa.admin_dirty_queue_marks'),
+                to_regclass('awa.admin_dirty_kind_marks')
+            )
+          AND c.relreplident <> 'f'
+          AND EXISTS (
+              SELECT 1 FROM pg_locks AS l
+              WHERE l.locktype = 'relation'
+                AND l.database = (SELECT oid FROM pg_database WHERE datname = current_database())
+                AND l.relation = c.oid
+                AND l.pid = pg_backend_pid()
+                AND l.mode = 'AccessExclusiveLock'
+                AND l.granted
+          )
+    LOOP
+        EXECUTE format('ALTER TABLE %s REPLICA IDENTITY %s', v_table, 'FULL');
+    END LOOP;
+END;
+$$;
+
 -- Carry pending dirtiness across the switch so nothing waits for the next
 -- full refresh. Marks that released trigger bodies commit into the old
 -- tables after this copy are reconciled by refresh_admin_metadata().

awa-model/migrations/v023_install_queue_storage_substrate.sql

@@ -869,6 +869,7 @@ DECLARE
     v_initial_gen    BIGINT;
     v_claimed_cte    TEXT;
     v_claim_runtime  REGPROCEDURE;
+    v_identity_rel   REGCLASS;
 BEGIN
     IF p_schema IS NULL OR p_schema !~ '^[a-z_][a-z0-9_]*$' THEN
         RAISE EXCEPTION 'install_queue_storage_substrate: schema name must match ^[a-z_][a-z0-9_]*$; got %',
@@ -4343,6 +4344,25 @@ BEGIN
         'Queue-storage claim allocator. Claims committed ready lanes, skips spent/tombstoned lanes, and leaves claim-cursor advancement safe under transaction rollback.'
     );
 
+    --------------------------------------------------------------------
+    -- REPLICA IDENTITY FULL on the terminal-count delta parent and each
+    -- of its partitions, which have no primary key and do not inherit
+    -- the parent's setting. Guarded: a re-run takes no lock.
+    --------------------------------------------------------------------
+
+    FOR v_identity_rel IN
+        SELECT c.oid::regclass
+        FROM pg_class AS c
+        JOIN pg_namespace AS n ON n.oid = c.relnamespace
+        WHERE n.nspname = p_schema
+          AND c.relkind IN ('r', 'p')
+          AND c.relreplident <> 'f'
+          AND c.relname ~ '^queue_terminal_count_deltas(_[0-9]+)?$'
+        ORDER BY c.relkind, c.relname
+    LOOP
+        EXECUTE format('ALTER TABLE %s REPLICA IDENTITY FULL', v_identity_rel);
+    END LOOP;
+
     --------------------------------------------------------------------
     -- Seed ring-slot generation rows. Slot 0 starts at generation 0;
     -- the rest start at -1 (sentinel for "never rotated through").

awa-model/src/lib.rs

@@ -16,6 +16,7 @@ mod read_transaction;
 pub mod reschedule;
 pub mod runtime_control;
 pub mod storage;
+mod storage_footprint_sql;
 pub mod unique;
 
 // Re-exports for ergonomics

awa-model/src/migrations.rs

@@ -423,6 +423,47 @@ macro_rules! correlated_claim_tombstones_pending_sql {
     };
 }
 
+/// Rows naming what still lacks the `replica_identity` patch: the installer,
+/// the awa tables without a primary key, and each `queue_terminal_count_deltas`
+/// table or partition of an installed queue-storage substrate (found as for
+/// `correlated_claim_tombstones`) whose replica identity is not FULL. A missing
+/// mark table counts, so the patch runs after `wait_free_dirty_marks` creates
+/// it.
+macro_rules! replica_identity_pending_sql {
+    () => {
+        "SELECT 'awa.install_queue_storage_substrate'::text AS target \
+         WHERE EXISTS ( \
+             SELECT 1 FROM pg_proc AS p \
+             WHERE p.pronamespace = 'awa'::regnamespace \
+               AND p.proname = 'install_queue_storage_substrate' \
+               AND p.prosrc NOT LIKE '%REPLICA IDENTITY FULL%') \
+         UNION ALL \
+         SELECT 'replica identity of awa.' || marks.relname \
+         FROM (VALUES ('admin_dirty_kind_marks'), ('admin_dirty_queue_marks')) AS marks(relname) \
+         WHERE NOT EXISTS (SELECT 1 FROM pg_class AS c \
+                           WHERE c.oid = to_regclass('awa.' || marks.relname) \
+                             AND c.relreplident = 'f') \
+         UNION ALL \
+         SELECT 'replica identity of awa.job_unique_claims' \
+         WHERE to_regclass('awa.job_unique_claims') IS NOT NULL \
+           AND NOT EXISTS (SELECT 1 FROM pg_index AS i \
+                           WHERE i.indrelid = to_regclass('awa.job_unique_claims') \
+                             AND i.indisreplident) \
+         UNION ALL \
+         SELECT format('replica identity of %I.%I', n.nspname, c.relname) \
+         FROM pg_class AS c \
+         JOIN pg_namespace AS n ON n.oid = c.relnamespace \
+         WHERE c.relkind IN ('r', 'p') \
+           AND c.relreplident <> 'f' \
+           AND c.relname ~ '^queue_terminal_count_deltas(_[0-9]+)?$' \
+           AND EXISTS (SELECT 1 FROM pg_proc AS p \
+                       WHERE p.pronamespace = n.oid \
+                         AND p.proname = 'claim_ready_runtime' \
+                         AND oidvectortypes(p.proargtypes) \
+                             = 'text, bigint, double precision, double precision')"
+    };
+}
+
 /// SQL returning one `text` row per object or schema that still needs the
 /// named patch, for patches that touch more than one place.
 fn patch_targets_sql(name: &str) -> Option<&'static str> {
@@ -442,6 +483,7 @@ fn patch_targets_sql(name: &str) -> Option<&'static str> {
             correlated_claim_tombstones_pending_sql!(),
             " ORDER BY 1"
         )),
+        "replica_identity" => Some(concat!(replica_identity_pending_sql!(), " ORDER BY 1")),
         _ => None,
     }
 }
@@ -565,6 +607,19 @@ pub const SCHEMA_PATCHES: &[SchemaPatch] = &[
             ")"
         ),
     },
+    SchemaPatch {
+        name: "replica_identity",
+        description: "Replica identity for every awa table (awa 0.7 migration v047)",
+        // The v023 definitions first, so the installer sets the identity in
+        // schemas prepared later; the patch then sets it on existing tables.
+        // Keep last: it holds table locks until its transaction commits.
+        sql: concat!(
+            include_str!("../migrations/v023_install_queue_storage_substrate.sql"),
+            "\n",
+            include_str!("../migrations/patches/replica_identity.sql"),
+        ),
+        applied_probe: concat!("SELECT NOT EXISTS (", replica_identity_pending_sql!(), ")"),
+    },
 ];
 
 /// Schema patches a database still needs to reach this binary's schema shape.
@@ -1175,6 +1230,18 @@ mod tests {
                 _ => patch.sql.to_string(),
             };
             let upper = top_level.to_uppercase();
+            // The only ALTER TABLE a patch may run is the guarded replica
+            // identity change.
+            let guarded_identity = "ALTER TABLE %S REPLICA IDENTITY %S";
+            if upper.contains(guarded_identity) {
+                assert!(
+                    upper.contains("RELREPLIDENT <> 'F'")
+                        && (!upper.contains("USING INDEX") || upper.contains("INDISREPLIDENT")),
+                    "patch {} must skip tables that already have their identity",
+                    patch.name
+                );
+            }
+            let upper = upper.replace(guarded_identity, "");
             assert!(
                 !upper.contains("INSERT INTO AWA.SCHEMA_VERSION"),
                 "patch {} must not record a schema version",