feat(push): durable per-address wake latch (migration 0025 + state machine)

One latch row per lease address (community, author, installation) replaces
row-per-event wake fanout in push_wake_outbox. States idle -> pending ->
sending; folds during pending are write-free, folds during sending set only
the owed_* triple (claimed identity immutable), every sending exit is the
same promote-or-idle rule. Accepted deliveries stamp a durable
cooldown_until consulted when the next cycle is armed, bounding accepted
wakes to one per address per cooldown window regardless of gateway speed.

Migration 0025 creates + seeds the latch table from live legacy outbox rows
and leaves push_wake_outbox fully intact for the dual-drain rollout window
(0026 owns retirement). Two global partial indexes back the two-arm
(pending-due / expired-sending recovery) claim.

No caller switches in this commit: relay behavior is unchanged. Covered by
11 PG state-transition tests (incl. fast-gateway cooldown, owed promotion,
retry/recovery request_id stability, lease rotation, brownfield seed) plus
a populated-database migrator test with an in-flight sending row.

Plan: PUSH_LEASE_DB_PERF_PLAN.md Rev 5 (blessed 9/9/9 by Eva + Wren).
Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

crates/buzz-db/src/lib.rs

@@ -37,6 +37,8 @@ pub mod partition;
 pub mod product_feedback;
 /// Community-scoped push lease and durable wake-outbox persistence.
 pub mod push;
+/// Durable per-address NIP-PL wake latch (replaces row-per-event wake fanout).
+pub mod push_latch;
 /// Reaction persistence.
 pub mod reaction;
 /// Relay-level membership persistence (NIP-43).
@@ -1375,6 +1377,59 @@ impl Db {
         .await
     }
 
+    /// Set-wise arm/fold of per-address wake latches for matched events.
+    pub async fn arm_push_latches(
+        &self,
+        community: CommunityId,
+        requests: &[push_latch::LatchArm],
+    ) -> Result<()> {
+        push_latch::arm_latches(&self.pool, community, requests).await
+    }
+
+    /// Globally claim due latch cycles (pending-due + expired-sending recovery).
+    pub async fn claim_due_push_latches(
+        &self,
+        limit: i64,
+        lease_until: DateTime<Utc>,
+    ) -> Result<Vec<push_latch::ClaimedLatch>> {
+        push_latch::claim_due_latches(&self.pool, limit, lease_until).await
+    }
+
+    /// Revalidate a fenced latch claim against the current lease and
+    /// representative event immediately before transport.
+    pub async fn revalidate_push_latch(
+        &self,
+        claim: &push_latch::ClaimedLatch,
+    ) -> Result<push_latch::RevalidateLatchOutcome> {
+        push_latch::revalidate_latch_for_send(&self.pool, claim).await
+    }
+
+    /// Exit a fenced latch claim after an accepted gateway delivery.
+    pub async fn complete_push_latch_delivered(
+        &self,
+        claim: &push_latch::ClaimedLatch,
+        cooldown: chrono::Duration,
+    ) -> Result<bool> {
+        push_latch::complete_latch_delivered(&self.pool, claim, cooldown).await
+    }
+
+    /// Exit a fenced latch claim that delivered nothing (suppressed/terminal).
+    pub async fn release_push_latch_undelivered(
+        &self,
+        claim: &push_latch::ClaimedLatch,
+    ) -> Result<bool> {
+        push_latch::release_latch_undelivered(&self.pool, claim).await
+    }
+
+    /// Return a fenced latch claim to `pending` for a same-cycle retry.
+    pub async fn retry_push_latch(
+        &self,
+        claim: &push_latch::ClaimedLatch,
+        next_attempt_at: DateTime<Utc>,
+    ) -> Result<bool> {
+        push_latch::retry_latch(&self.pool, claim, next_attempt_at).await
+    }
+
     /// Atomically insert an event AND its thread metadata in a single transaction.
     pub async fn insert_event_with_thread_metadata(
         &self,

crates/buzz-db/src/migration.rs

@@ -560,7 +560,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 24);
+        assert_eq!(migrations.len(), 25);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -879,6 +879,23 @@ mod tests {
             .to_lowercase()
             .contains("for update"));
         assert!(ttl_shared.contains("NEW.kind <> 9007"));
+
+        // Push wake latch: one durable row per lease address replaces
+        // row-per-event wake fanout. The seed must leave the legacy outbox
+        // intact (dual-drain rollout) and cooldown_until must be non-null
+        // with an epoch-zero default so arming never branches on NULL.
+        assert_eq!(migrations[24].version, 25);
+        let latch = migrations[24].sql.as_str();
+        assert!(latch.contains("CREATE TABLE push_wake_latch"));
+        assert!(latch.contains("cooldown_until TIMESTAMPTZ NOT NULL DEFAULT to_timestamp(0)"));
+        assert!(latch.contains("push_wake_latch_due_global"));
+        assert!(latch.contains("push_wake_latch_recovery_global"));
+        assert!(latch.contains("FROM push_wake_outbox"));
+        let latch_lower = strip_sql_comments(latch).to_lowercase();
+        assert!(
+            !latch_lower.contains("drop table") && !latch_lower.contains("delete from"),
+            "0025 must not mutate or drop the legacy outbox; 0026 owns retirement"
+        );
     }
 
     #[test]
@@ -1121,7 +1138,7 @@ mod tests {
         run_migrations(&pool)
             .await
             .expect("retry succeeds after operator repair");
-        assert_eq!(applied_versions(&pool).await.last().copied(), Some(24));
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(25));
     }
 
     #[tokio::test]
@@ -1184,6 +1201,82 @@ mod tests {
         assert_eq!(after, vec![(1, Some(true)), (30_350, None)]);
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn brownfield_0025_seeds_latches_from_in_flight_outbox_rows() {
+        let pool = connect_test_pool().await;
+        reset_public_schema(&pool).await;
+        MIGRATOR
+            .run_to(24, &pool)
+            .await
+            .expect("apply migrations through 24");
+
+        let community_id = uuid::Uuid::new_v4();
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community_id)
+            .bind(format!("pre-0025-{}.example", community_id.simple()))
+            .execute(&pool)
+            .await
+            .expect("insert community");
+        sqlx::query(
+            "INSERT INTO push_leases (community_id, author, installation_id, source_event_id, \
+                 source_created_at, generation, active, app_profile, endpoint_hash, \
+                 endpoint_grant, max_class, subscriptions, expires_at) \
+             VALUES ($1, $2, 'install', $3, 1, 1, true, 'ios-production', $4, 'grant', \
+                 'default', '[]'::jsonb, $5)",
+        )
+        .bind(community_id)
+        .bind([1_u8; 32])
+        .bind([2_u8; 32])
+        .bind([3_u8; 32])
+        .bind(i64::MAX / 2)
+        .execute(&pool)
+        .await
+        .expect("insert live lease");
+        // An in-flight `sending` wake claimed by an old pod at migration time.
+        sqlx::query(
+            "INSERT INTO push_wake_outbox (community_id, author, installation_id, \
+                 lease_generation, endpoint_hash, event_id, class, expires_at, state, \
+                 lease_until, claim_id) \
+             VALUES ($1, $2, 'install', 1, $3, $4, 'default', $5, 'sending', \
+                 now() + interval '1 minute', gen_random_uuid())",
+        )
+        .bind(community_id)
+        .bind([1_u8; 32])
+        .bind([3_u8; 32])
+        .bind([4_u8; 32])
+        .bind(i64::MAX / 2)
+        .execute(&pool)
+        .await
+        .expect("insert in-flight legacy wake");
+
+        run_migrations(&pool)
+            .await
+            .expect("brownfield migration onto populated database");
+
+        let latch: (String, Vec<u8>) = sqlx::query_as(
+            "SELECT state, event_id FROM push_wake_latch \
+             WHERE community_id = $1 AND author = $2 AND installation_id = 'install'",
+        )
+        .bind(community_id)
+        .bind([1_u8; 32])
+        .fetch_one(&pool)
+        .await
+        .expect("seeded latch for the in-flight address");
+        assert_eq!(latch, ("pending".into(), [4_u8; 32].to_vec()));
+
+        // 0025 must leave the legacy row fully intact — old pods still own it
+        // until the rolling deploy finishes (0026 owns retirement).
+        let legacy: (String, Option<uuid::Uuid>) =
+            sqlx::query_as("SELECT state, claim_id FROM push_wake_outbox WHERE community_id = $1")
+                .bind(community_id)
+                .fetch_one(&pool)
+                .await
+                .expect("legacy row survives 0025");
+        assert_eq!(legacy.0, "sending");
+        assert!(legacy.1.is_some(), "old pod's claim fence untouched");
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn run_migrations_applies_consolidated_initial_schema_on_fresh_database() {

migrations/0025_push_wake_latch.sql

@@ -0,0 +1,86 @@
+-- NIP-PL durable wake latch: one row per lease address replaces row-per-event
+-- wake fanout in push_wake_outbox. Content-free pushes make per-event rows
+-- redundant: a wake means "sync now", so an endpoint owed a wake needs at most
+-- one durable pending cycle plus a record of work that arrived mid-send.
+--
+-- State machine (crates/buzz-db/src/push_latch.rs is the single writer):
+--   idle -> pending    first matched event; leading wake due at
+--                      GREATEST(now(), cooldown_until) so a genuinely idle
+--                      address wakes immediately while a just-woken one defers
+--                      to the cooldown boundary.
+--   pending -> pending matched events fold write-free unless the lease
+--                      generation changed or the current cycle expired.
+--   pending -> sending worker claim (fenced by claim_id/lease_until).
+--   sending            claimed identity is IMMUTABLE; a matched event may only
+--                      set the owed_* triple (first-owed wins; generation
+--                      change refreshes it).
+--   sending -> pending accepted delivery with owed work: promote owed_* to the
+--                      current cycle, mint a new request_id, due at the new
+--                      cooldown boundary. Suppressed/terminal exits promote
+--                      immediately (no wake reached the device, no cooldown).
+--   sending -> idle    exit with no owed work. cooldown_until persists on the
+--                      idle row: it is what bounds accepted wakes to one per
+--                      address per cooldown window regardless of gateway speed.
+--
+-- cooldown_until is NOT NULL with epoch-zero default so arming logic never
+-- branches on NULL. request_id is the stable gateway/APNs replay-fence id for
+-- one wake cycle: constant across retries, minted per cycle.
+CREATE TABLE push_wake_latch (
+    community_id UUID NOT NULL REFERENCES communities(id),
+    author BYTEA NOT NULL CHECK (length(author) = 32),
+    installation_id TEXT NOT NULL,
+    state TEXT NOT NULL CHECK (state IN ('idle', 'pending', 'sending')),
+    generation BIGINT NOT NULL CHECK (generation > 0),
+    event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
+    expires_at BIGINT NOT NULL,
+    request_id UUID NOT NULL,
+    owed_event_id BYTEA CHECK (owed_event_id IS NULL OR length(owed_event_id) = 32),
+    owed_generation BIGINT,
+    owed_expires_at BIGINT,
+    cooldown_until TIMESTAMPTZ NOT NULL DEFAULT to_timestamp(0),
+    next_attempt_at TIMESTAMPTZ,
+    lease_until TIMESTAMPTZ,
+    claim_id UUID,
+    attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (community_id, author, installation_id),
+    FOREIGN KEY (community_id, author, installation_id)
+        REFERENCES push_leases (community_id, author, installation_id),
+    -- The owed triple is all-present or all-absent.
+    CHECK ((owed_event_id IS NULL) = (owed_generation IS NULL)
+       AND (owed_event_id IS NULL) = (owed_expires_at IS NULL)),
+    -- A pending cycle is always due at some point; a claim always carries its
+    -- fence. Idle rows carry neither.
+    CHECK (state <> 'pending' OR next_attempt_at IS NOT NULL),
+    CHECK ((state = 'sending') = (claim_id IS NOT NULL)),
+    CHECK ((state = 'sending') = (lease_until IS NOT NULL))
+);
+
+-- Global (not community-prefixed) partial indexes: the delivery worker claims
+-- across all communities in one statement. Two arms, two indexes — the claim
+-- query is shaped as UNION ALL so each arm is independently indexable.
+CREATE INDEX push_wake_latch_due_global
+    ON push_wake_latch (next_attempt_at) WHERE state = 'pending';
+CREATE INDEX push_wake_latch_recovery_global
+    ON push_wake_latch (lease_until) WHERE state = 'sending';
+
+-- Brownfield seed: any address that currently has live legacy outbox work is
+-- owed a wake. One pending latch per such address, representative = its newest
+-- live legacy row. The legacy table and its workers are left fully intact:
+-- old pods keep enqueueing/draining push_wake_outbox during the rolling
+-- deploy, and new workers dual-drain both sources until migration 0026
+-- retires the legacy path. Duplicate wakes across the two systems during the
+-- window are content-free and harmless; lost wakes are not possible.
+INSERT INTO push_wake_latch (
+    community_id, author, installation_id, state, generation, event_id,
+    expires_at, request_id, next_attempt_at
+)
+SELECT DISTINCT ON (o.community_id, o.author, o.installation_id)
+    o.community_id, o.author, o.installation_id, 'pending', o.lease_generation,
+    o.event_id, o.expires_at, gen_random_uuid(), now()
+FROM push_wake_outbox o
+WHERE o.state IN ('pending', 'sending')
+  AND o.expires_at > EXTRACT(EPOCH FROM now())::bigint
+ORDER BY o.community_id, o.author, o.installation_id, o.created_at DESC
+ON CONFLICT DO NOTHING;