[Jetty] Added the notification log to the PostgreSQL store.

crates/protos/protos/local/temporal/sdk/core/streams/streams.proto

@@ -167,8 +167,17 @@ message PostgresStoreConfig {
   google.protobuf.Duration floor_lease = 6;
   // How long a hold lasts after its reader released it. Unset keeps one day.
   google.protobuf.Duration floor_release_grace = 7;
+  // With `StreamStoreConfig.notification_log`, how many doorbells the log rings, from 1 to 64.
+  // Zero uses 16.
+  uint32 notify_doorbells = 8;
+  // A URL for the connection that listens and the one that applies migrations, when `url` goes
+  // through a pooler in transaction mode. Empty uses `url`.
+  string listen_url = 9;
   // How often a process looks for new records for its waiting reads. Unset uses 100 ms.
   google.protobuf.Duration poll_interval = 10;
+  // Each append also sends NOTIFY, so waiting reads in other processes wake in milliseconds.
+  // Costs append throughput when many producers write at once.
+  bool notify_reads = 11;
   // The store checks the schema version and never changes it. A DBA applies the migrations.
   bool skip_migrations = 12;
   // How often the janitor trims records past the retention. Unset uses one minute.

crates/sdk-core/CHANGELOG.md

@@ -83,6 +83,12 @@ relevant information.
   every `janitor_interval` (one minute). Waiting reads hold no connection and wake within
   `poll_interval` (100 ms). TLS follows the URL's `sslmode` and `sslrootcert`. The C bridge builds
   the store.
+* Experimental: the PostgreSQL store keeps a notification log with `notification_log`, a
+  `notifications` table with 64 partitions that the append writes in its transaction. Drivers take
+  entries with `FOR UPDATE SKIP LOCKED`, and a `NOTIFY` doorbell rings when a partition takes its
+  first entry. `listen_url` names a direct connection for the listener and the migrations when
+  `url` goes through PgBouncer in transaction mode. `notify_reads` sends a `NOTIFY` per append, so
+  waiting reads in other processes wake at once.
 * Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the
   notifications still out. A process calls it before it stops.
 * Experimental: a server without the stream notifier answers the first notifier call with

crates/streams/docs/postgres.md

@@ -41,6 +41,32 @@ store, or for the store's poll. While any read waits, the store runs one stateme
 another process wakes within about `poll_interval`. The server's stream notifier is the main wake
 path, and this wait is the path for a reader without it.
 
+**`notify_reads`.** With it, each append also sends `NOTIFY` on `<schema>_read_<topic id % 64>`
+with the topic id, and the store's listener wakes the waiters of that topic. A reader in another
+process then wakes in a few milliseconds. A reader that waits on a topic with no row yet wakes
+from the poll, since a `NOTIFY` can't name it. It is off by default, since a transaction that sent
+`NOTIFY` takes a lock on the server's notification queue as it commits, so such commits run one
+at a time across the database. With many producers at once, it costs append throughput.
+
+**Notification log.** With `notification_log` on, the append also writes a row to
+`notifications` in its transaction: the topic and the append's last position, in one of 64
+partitions fixed in the schema. Every topic of a chain lands in the partition that the first four
+bytes of SHA-256 over the chain's fields give. A repeat of the newest batch writes none. When the
+partition held no entry before, the append also sends `NOTIFY` on `<schema>_bell_<partition %
+doorbells>` with the partition, so the doorbell costs nothing while a partition stays busy. Drivers
+take entries with `FOR UPDATE SKIP LOCKED`: a consumer's own unacked entries, those another
+consumer left for `claim_idle`, and new ones. An ack deletes them. The janitor keeps the newest
+10,000 entries per partition, so a log with no driver stays bounded. An entry whose topic an owner
+delete removed is acked as broken when a driver takes it.
+
+**Listener.** One connection per store listens for the doorbells and, with `notify_reads`, the read
+channels. It connects with the store, so a store whose listener can't connect fails at once. When
+the connection ends, it connects again with backoff, rings every doorbell for every partition and
+wakes every waiting read, since what was sent while it was away is gone. The driver's sweep finds a
+lost ring too. The listener sets TCP keepalive to notice a dead connection within a minute.
+`LISTEN` keeps state on the server session, so the listener uses `listen_url` when it is set, and
+`url` otherwise. Both must name the primary, since a standby refuses `LISTEN`.
+
 **Retention.** PostgreSQL has no key expiry, so each store runs a janitor every `janitor_interval`
 (one minute). A transaction advisory lock around each step lets one janitor work at a time across
 processes, also through PgBouncer in transaction mode. A pass trims the topics that are due, at most
@@ -69,10 +95,12 @@ so it reports `REFUSED`. A read reports `STORAGE`. A server below PostgreSQL 14
 ## Migrations
 
 The store embeds its migrations and applies the missing ones at connect, each in its own
-transaction, under a session advisory lock on the schema's name. It refuses to start when an
-applied migration differs from the embedded one, or when the database has a migration that needs a
-newer build. A function's name carries its version, such as `append_v1`, so a migration that
-changes a contract adds a new function and keeps the old one for a release.
+transaction, under a session advisory lock on the schema's name. They run on a connection of their
+own to `listen_url` when it is set, since a pooler in transaction mode can't keep a session lock.
+The store refuses to start when an applied migration differs from the embedded one, or when the
+database has a migration that needs a newer build. A function's name carries its version, such
+as `append_v1`, so a migration that changes a contract adds a new function and keeps the old one
+for a release.
 
 With `skip_migrations`, the store only checks that every migration it knows is applied. A DBA then
 applies the files in `src/postgres/migrations/` in order, after `SET search_path = <schema>,
@@ -127,8 +155,10 @@ IAM database authentication is not in this release, so the password comes from t
 **PgBouncer.** Session mode works with no setting. Transaction mode works from PgBouncer 1.21 with
 `max_prepared_statements` set, since the store prepares its statements. The session settings
 apply to the server connection that made each pooled connection, so all clients of that
-PgBouncer database should use the same `READ COMMITTED` default. Migrations take a session lock,
-so run them once through a direct connection, or use `skip_migrations`.
+PgBouncer database should use the same `READ COMMITTED` default. Set `listen_url` to a direct
+connection, which the listener and the migrations use. The conformance suite runs through
+PgBouncer 1.26 in transaction mode with `max_prepared_statements = 200` and a direct
+`listen_url`.
 
 **Servers that speak the protocol.** CockroachDB, YugabyteDB and other servers that speak the
 PostgreSQL protocol are not supported. The store needs advisory locks, PL/pgSQL and PostgreSQL's
@@ -142,9 +172,11 @@ Redis.
 
 ## Tests
 
-`scripts/postgres-test-server.sh` starts PostgreSQL 14 and 17 in Docker on free ports and prints
-the URLs to export. The suite reads `STREAMS_POSTGRES_URL`, a role that may create schemas. Each
-case runs in a schema of its own.
+`scripts/postgres-test-server.sh` starts PostgreSQL 14 and 17 in Docker on free ports, and
+PgBouncer in transaction mode in front of 17, and prints the URLs to export. The suite reads
+`STREAMS_POSTGRES_URL`, a role that may create schemas, and `STREAMS_POSTGRES_PGBOUNCER_URL`. Each
+case runs in a schema of its own. A schema name takes at most 55 bytes, so that the channel names
+stay within 63.
 
 ```sh
 eval "$(crates/streams/scripts/postgres-test-server.sh)"

crates/streams/scripts/postgres-test-server.sh

@@ -1,20 +1,21 @@
 #!/usr/bin/env bash
 # Starts the PostgreSQL servers the stream store tests run against, 14 and 17, in Docker on free
-# ports, with TLS on from the image's self-signed certificate. Prints the variables to export.
-# `stop` stops them. STREAMS_POSTGRES_NAME moves the container names, so two runs can share a
-# host.
+# ports, with TLS on from the image's self-signed certificate, and PgBouncer in transaction mode
+# in front of 17 when its image pulls. Prints the variables to export. `stop` stops them.
+# STREAMS_POSTGRES_NAME moves the container names, so two runs can share a host.
 #
 #   eval "$(crates/streams/scripts/postgres-test-server.sh)"
 #   STREAMS_POSTGRES_URL=postgres://postgres:streams@localhost:<port>/postgres       (17)
 #   STREAMS_POSTGRES_14_URL=postgres://postgres:streams@localhost:<port>/postgres    (14)
 #   STREAMS_POSTGRES_TLS_URL=$STREAMS_POSTGRES_URL
+#   STREAMS_POSTGRES_PGBOUNCER_URL=postgres://postgres:streams@localhost:<port>/postgres
 set -euo pipefail
 
 NAME=${STREAMS_POSTGRES_NAME:-streams-postgres}
 VERSIONS=(14 17)
 
 if [ "${1:-}" = stop ]; then
-  for version in "${VERSIONS[@]}"; do
+  for version in "${VERSIONS[@]}" pgbouncer; do
     docker rm -f "$NAME-$version" >/dev/null 2>&1 || true
   done
   exit 0
@@ -42,6 +43,21 @@ for version in "${VERSIONS[@]}"; do
   eval "URL_$version=postgres://postgres:streams@localhost:$port/postgres"
 done
 
+# Prepared statements in transaction mode need PgBouncer 1.21 or later.
+# PgBouncer resolves names itself, so it reaches 17 by its address on the Docker bridge.
+bouncer="$NAME-pgbouncer"
+address_17=$(docker inspect -f '{{.NetworkSettings.Networks.bridge.IPAddress}}' "$NAME-17")
+if ! docker inspect "$bouncer" >/dev/null 2>&1; then
+  docker run -d --rm --name "$bouncer" -p "127.0.0.1:$(free_port):5432" \
+    -e DATABASE_URL="postgres://postgres:streams@$address_17:5432/postgres" \
+    -e POOL_MODE=transaction -e MAX_PREPARED_STATEMENTS=200 -e AUTH_TYPE=scram-sha-256 \
+    -e MAX_CLIENT_CONN=1000 -e DEFAULT_POOL_SIZE=40 -e LISTEN_PORT=5432 \
+    edoburu/pgbouncer:latest >/dev/null 2>&1 || echo "PgBouncer did not start" >&2
+fi
+if docker inspect "$bouncer" >/dev/null 2>&1; then
+  port=$(docker port "$bouncer" 5432/tcp | head -1 | sed 's/.*://')
+  echo "export STREAMS_POSTGRES_PGBOUNCER_URL=postgres://postgres:streams@localhost:$port/postgres"
+fi
 echo "export STREAMS_POSTGRES_URL=$URL_17"
 echo "export STREAMS_POSTGRES_14_URL=$URL_14"
 echo "export STREAMS_POSTGRES_TLS_URL=$URL_17"

crates/streams/src/lib.rs

@@ -33,7 +33,7 @@ mod subscribe;
 mod tiered;
 
 #[cfg(feature = "postgres")]
-pub use self::postgres::{PostgresStore, PostgresStoreOptions};
+pub use self::postgres::{DEFAULT_POSTGRES_DOORBELLS, PostgresStore, PostgresStoreOptions};
 #[cfg(feature = "redis")]
 pub use self::redis::{DEFAULT_NOTIFY_DOORBELLS, RedisStore, RedisStoreOptions};
 #[cfg(feature = "archive")]

crates/streams/src/postgres/conn.rs

@@ -9,7 +9,10 @@ use rustls::{
     pki_types::{CertificateDer, ServerName, UnixTime, pem::PemObject},
 };
 use std::{sync::Arc, time::Duration};
-use tokio_postgres::{Client, NoTls, config::SslMode};
+use tokio::sync::mpsc;
+use tokio_postgres::{
+    AsyncMessage, Client, Connection, NoTls, Notification, config::SslMode, tls::TlsStream,
+};
 use tokio_postgres_rustls::MakeRustlsConnect;
 
 /// The `sslmode` values libpq takes. `allow` is left out, since it trusts a plain connection
@@ -93,6 +96,40 @@ impl Target {
         }
     }
 
+    /// One connection of its own whose notifications arrive on the receiver. The receiver
+    /// closes when the connection ends.
+    pub(super) async fn listen(
+        &self,
+    ) -> Result<(Client, mpsc::UnboundedReceiver<Notification>), tokio_postgres::Error> {
+        let (sender, receiver) = mpsc::unbounded_channel();
+        let config = self.listen_config();
+        let client = match &self.tls {
+            Some(tls) => {
+                let (client, connection) = config.connect(tls.clone()).await?;
+                tokio::spawn(forward(connection, sender));
+                client
+            }
+            None => {
+                let (client, connection) = config.connect(NoTls).await?;
+                tokio::spawn(forward(connection, sender));
+                client
+            }
+        };
+        Ok((client, receiver))
+    }
+
+    /// The listener only receives, so a half-open connection shows only to TCP keepalive. The
+    /// system default notices it after about two hours, and the listener's wakes stop meanwhile.
+    fn listen_config(&self) -> tokio_postgres::Config {
+        let mut config = self.config.clone();
+        config
+            .keepalives(true)
+            .keepalives_idle(Duration::from_secs(30))
+            .keepalives_interval(Duration::from_secs(10))
+            .keepalives_retries(3);
+        config
+    }
+
     /// The pool of `size` connections. Each new connection pins the session settings the store
     /// functions depend on.
     pub(super) fn pool(&self, size: usize, timeout: Duration) -> StreamResult<Pool> {
@@ -124,6 +161,30 @@ impl Target {
     }
 }
 
+/// Drives a connection and hands its notifications on, until it ends.
+async fn forward<S, T>(
+    mut connection: Connection<S, T>,
+    sender: mpsc::UnboundedSender<Notification>,
+) where
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+    T: TlsStream + Unpin,
+{
+    while let Some(message) = std::future::poll_fn(|cx| connection.poll_message(cx)).await {
+        match message {
+            Ok(AsyncMessage::Notification(notification)) => {
+                if sender.send(notification).is_err() {
+                    return;
+                }
+            }
+            Ok(_) => {}
+            Err(error) => {
+                tracing::debug!("The PostgreSQL store's listener connection ended: {error}");
+                return;
+            }
+        }
+    }
+}
+
 /// Every call is one statement that depends on `READ COMMITTED`: a statement that waited for a
 /// row lock reads the newest version of the row. The session pins it, so a database default of
 /// `SERIALIZABLE` can't change it.
@@ -400,6 +461,20 @@ mod tests {
         assert_eq!(mode, Mode::VerifyCa);
     }
 
+    #[test]
+    fn the_listener_notices_a_dead_connection_within_a_minute() {
+        let target = Target {
+            config: parse_url("postgres://u:p@localhost/db").unwrap().0,
+            tls: None,
+            mode: Mode::Disable,
+        };
+        let config = target.listen_config();
+        assert!(config.get_keepalives());
+        let detect = config.get_keepalives_idle()
+            + config.get_keepalives_interval().unwrap() * config.get_keepalives_retries().unwrap();
+        assert!(detect <= Duration::from_secs(60), "{detect:?}");
+    }
+
     #[test]
     fn another_scheme_or_sslmode_is_refused() {
         for url in [

crates/streams/src/postgres/janitor.rs

@@ -2,7 +2,7 @@
 //! transaction advisory lock around each step lets one janitor work at a time across processes.
 
 use super::{
-    DELETE_BATCH, Inner,
+    DELETE_BATCH, Inner, LOG_CAP,
     errors::{Call, mapped, timed_out, unreachable},
     migrations::LOCK_SPACE,
 };
@@ -29,8 +29,9 @@ pub(super) async fn run(inner: Weak<Inner>, interval: Duration) {
 }
 
 impl Inner {
-    /// One pass: trims the topics that are due, finishes purges, and deletes idle chains. A pass
-    /// stops when another process's janitor holds the lock, since that one does the work.
+    /// One pass: trims the topics that are due, finishes purges, caps the notification log, and
+    /// deletes idle chains. A pass stops when another process's janitor holds the lock, since
+    /// that one does the work.
     pub(super) async fn janitor_pass(&self) -> StreamResult<u64> {
         let mut client = self.pool.get().await.map_err(unreachable)?;
         let mut deleted = 0;
@@ -61,6 +62,10 @@ impl Inner {
                 break;
             }
         }
+        if self.log {
+            self.locked(&mut client, &self.sql.cap_log, &[&LOG_CAP])
+                .await?;
+        }
         self.locked(
             &mut client,
             &self.sql.idle_chains,