pg tx producer paths wake commit-atomically: publish_with_key_tx/enqueue_tx/outbox_enqueue_tx issue pg_notify on their mechanism-named channel (stream name / queue name / derived backing queue) inside the caller's tx — empty payload, best-effort log-and-swallow, no double-wake on the auto-commit paths; shared tx::wake_tx owner + module-doc section pinning the semantics; new tx_tests harness test pinning pre-commit silence, commit delivery on all three channels, and rollback silence; F-1 engine arm retired in review-wave-4 + implementation.md records (tx_publishes_compose_with_the_handle deterministic: 20 solo runs green; pg suite 116/116 x2 vs harness) (task pg-fix-tx-wake, review 002 Finding 2)

alkstore-postgres/src/store/tx_tests.rs

@@ -5,7 +5,10 @@
 //! dispositions (no-ghosts across job rows, events, notifications, and
 //! offset saves), drop panic-safety + the unknowable-state discard
 //! arms, ops-after-consume → `Closed`, `with_tx` end-to-end, the
-//! 8000-byte notify boundary, and pool accounting under concurrency.
+//! 8000-byte notify boundary, pool accounting under concurrency, and
+//! the producer tx paths' commit-atomic `pg_notify` wakes (delivered
+//! at the caller's commit on the mechanism-named channel; rollback
+//! drops the row with its wake).
 //!
 //! Harness convention (the schema/open tasks'): connection settings
 //! ride the environment (`ALKSTORE_PG_HOST/PORT/USER/PASSWORD/DB`),
@@ -1261,3 +1264,107 @@ async fn many_txs_commit_exactly_once_through_the_pool() {
     drop_schema(&admin, &schema).await;
     Arc::into_inner(store).unwrap().close();
 }
+
+/// Acceptance: the producer tx paths issue their `pg_notify` wakes
+/// commit-atomically (review 002 Finding 2's fix) — a registered
+/// listener receives the wake at/after the caller's commit,
+/// deterministically, for all three channels (the stream name for
+/// `publish_tx`, the queue name for `enqueue_tx`, the derived backing
+/// queue for `outbox_enqueue_tx`); the wake is not statement-atomic
+/// (nothing delivers before the COMMIT); and a rolled-back tx drops
+/// the row *with* its wake (the no-ghosts arm — dead silence after
+/// rollback).
+#[tokio::test(flavor = "multi_thread")]
+async fn tx_producer_wakes_are_commit_atomic_and_rollback_discards_them() {
+    let Some(dsn) = harness_dsn() else {
+        eprintln!("skip: no harness server");
+        return;
+    };
+    let schema = instance_namer("tx-wake")();
+    let store = open_store(&dsn, test_opts(&schema)).await.unwrap();
+    let admin = harness_client().await.unwrap();
+
+    let stream = instance_namer("wake_st")();
+    let queue = instance_namer("wake_q")();
+    let outbox = instance_namer("wake_o")();
+    let derived = crate::resolution::outbox_backing_queue_name(&outbox);
+
+    // Registered + subscribed BEFORE any write rides.
+    for channel in [&stream, &queue, &derived] {
+        store.forwarder().register(channel).await.unwrap();
+    }
+    let mut stream_rx = store.forwarder().subscribe().unwrap();
+    let mut queue_rx = store.forwarder().subscribe().unwrap();
+    let mut outbox_rx = store.forwarder().subscribe().unwrap();
+
+    // One business tx producing on all three channels.
+    let mut tx = store.begin_tx().await.unwrap();
+    tx.publish_tx(&stream, serde_json::json!({"via": "tx"}))
+        .await
+        .unwrap();
+    tx.enqueue_tx(&queue, Default::default(), serde_json::json!({"n": 1}))
+        .await
+        .unwrap();
+    tx.outbox_enqueue_tx(&outbox, Default::default(), serde_json::json!({"n": 2}))
+        .await
+        .unwrap();
+
+    // Statement-issued, not yet delivered: pg_notify inside a tx does
+    // not emit at statement time (commit-atomic, not statement-atomic).
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    for (rx, channel, tag) in [
+        (&mut stream_rx, &stream, "stream"),
+        (&mut queue_rx, &queue, "queue"),
+        (&mut outbox_rx, &derived, "outbox"),
+    ] {
+        assert!(
+            !listener_hears(rx, channel, Duration::from_millis(100)).await,
+            "{tag}: no wake may deliver before the caller's commit"
+        );
+    }
+
+    tx.commit().await.unwrap();
+
+    // The wakes deliver at/after the commit, deterministically.
+    assert!(
+        listener_hears(&mut stream_rx, &stream, Duration::from_secs(5)).await,
+        "a committed publish_tx wakes the stream's channel at commit"
+    );
+    assert!(
+        listener_hears(&mut queue_rx, &queue, Duration::from_secs(5)).await,
+        "a committed enqueue_tx wakes the queue's channel at commit"
+    );
+    assert!(
+        listener_hears(&mut outbox_rx, &derived, Duration::from_secs(5)).await,
+        "a committed outbox_enqueue_tx wakes the derived backing channel at commit"
+    );
+
+    // The no-ghosts arm: the rollback drops the event/job rows *with*
+    // their wakes — the registered channels stay silent.
+    let mut tx = store.begin_tx().await.unwrap();
+    tx.publish_tx(&stream, serde_json::json!({"via": "ghost"}))
+        .await
+        .unwrap();
+    tx.enqueue_tx(&queue, Default::default(), serde_json::json!({"n": 3}))
+        .await
+        .unwrap();
+    tx.outbox_enqueue_tx(&outbox, Default::default(), serde_json::json!({"n": 4}))
+        .await
+        .unwrap();
+    drop(tx);
+    tokio::time::sleep(Duration::from_millis(400)).await;
+    for (rx, channel, tag) in [
+        (&mut stream_rx, &stream, "stream"),
+        (&mut queue_rx, &queue, "queue"),
+        (&mut outbox_rx, &derived, "outbox"),
+    ] {
+        assert!(
+            !listener_hears(rx, channel, Duration::from_millis(300)).await,
+            "{tag}: a rolled-back tx must not wake (commit-atomic, not statement-atomic)"
+        );
+    }
+    assert_eq!(raw_event_count(&admin, &schema, &stream).await, 1);
+
+    drop_schema(&admin, &schema).await;
+    drop(store);
+}

alkstore-postgres/src/tx.rs

@@ -54,6 +54,21 @@
 //! the same byte string ADR-020 §4 pins as the stored row bytes —
 //! which is valid UTF-8 by construction.
 //!
+//! # The tx wakes
+//!
+//! The producer paths (`publish_with_key_tx` — which `publish_tx`
+//! rides, `enqueue_tx`, `outbox_enqueue_tx`) issue `pg_notify` on the
+//! mechanism's wake channel (the stream name / the queue name / the
+//! outbox's derived backing-queue name — the mechanism-name-is-the-
+//! channel realization) **inside the caller's transaction** ([`wake_tx`]).
+//! NOTIFY's native transactional delivery makes the wake commit-atomic:
+//! it delivers at the caller's COMMIT, and a rollback discards the row
+//! *with* its wake (the tx/no-ghosts rows' pin). Wakes are best-effort
+//! (ADR-006; the engine-wide posture — a wake must never fail a write):
+//! a failed statement is logged and swallowed, and since a failed
+//! statement aborts the tx server-side the truth still surfaces typed
+//! at COMMIT. `run_once` is a pull op, consumer-driven — no wake there.
+//!
 //! # Engine-side arithmetic
 //!
 //! The stamp/derivation arithmetic lives in [`crate::resolution`]
@@ -260,6 +275,26 @@ fn enqueue_row_sql(schema: &str) -> String {
     )
 }
 
+/// The in-tx wake statement (empty payload — `Wake { channel }` only,
+/// ADR-008 §3): issued inside the caller's transaction, so NOTIFY's
+/// native transactional delivery makes it commit-atomic — delivers at
+/// the caller's COMMIT, discarded with the row by a rollback. Best-
+/// effort at the op (the engine-wide posture — a wake must never fail
+/// a write): a failed statement is logged and swallowed; because a
+/// failed statement aborts the tx server-side, the truth still
+/// surfaces typed at COMMIT.
+async fn wake_tx(client: &tokio_postgres::Client, channel: &str) {
+    if let Err(wake_err) = client
+        .query_one("SELECT pg_notify($1, '')", &[&channel])
+        .await
+    {
+        eprintln!(
+            "alkstore-postgres: tx wake failed for {channel:?} \
+             (best-effort: the durable row lands at the caller's commit): {wake_err}"
+        );
+    }
+}
+
 /// Encode a trait-crossing payload `Value` into the exact stored bytes
 /// (the serde_json serialization — ADR-020 §4). The notify carriage
 /// reuses it: the JSON serialization is valid UTF-8 text by
@@ -282,8 +317,19 @@ impl TxHandle for PgTxHandle {
         Box::pin(async move {
             let client = self.client()?;
             let bytes = encode_payload_bytes(&payload)?;
-            self.enqueue_row(client, &queue, bytes, &opts, plain_queue_default_stamps())
-                .await
+            let id = self
+                .enqueue_row(client, &queue, bytes, &opts, plain_queue_default_stamps())
+                .await?;
+            // Posture-parity wake (the contract obligation is the
+            // stream path's): pg_notify on the queue's wake channel
+            // inside the caller's tx — commit-atomic like every
+            // in-tx notify, matching SQLite's watcher firing on the
+            // committed write. Best-effort (the queues row's pinned
+            // re-poll safety net covers a gap; a failed statement
+            // aborts the tx server-side, so the truth surfaces typed
+            // at COMMIT).
+            wake_tx(client, &queue).await;
+            Ok(id)
         })
     }
 
@@ -324,6 +370,14 @@ impl TxHandle for PgTxHandle {
                 )
                 .await
                 .map_err(pg_error)?;
+            // Then wake, inside the caller's tx (the contract
+            // obligation — engine-postgres.md pins `pg_notify` as the
+            // streams wake trigger, and core-contract.md pins "events
+            // never require polling to become visible" with rollback
+            // dropping the event row *with* its wake): the stream's
+            // wake channel, the mechanism-name-is-the-channel
+            // realization — the auto-commit publish's twin.
+            wake_tx(client, &stream).await;
             Ok(row.get(0))
         })
     }
@@ -556,8 +610,14 @@ impl TxHandle for PgTxHandle {
             let client = self.client()?;
             let bytes = encode_payload_bytes(&payload)?;
             let queue = outbox_backing_queue_name(&outbox);
-            self.enqueue_row(client, &queue, bytes, &opts, outbox_default_stamps())
-                .await
+            let id = self
+                .enqueue_row(client, &queue, bytes, &opts, outbox_default_stamps())
+                .await?;
+            // The outbox's posture-parity twin of `enqueue_tx`'s wake:
+            // the derived backing-queue name is the channel (the
+            // scheduler's fire-wake targets the same name).
+            wake_tx(client, &queue).await;
+            Ok(id)
         })
     }