feat(pgbouncer): peer so_reuseport workers to forward cross-worker query cancels

README.md

@@ -17,7 +17,7 @@ Forks boot in seconds against a CoW snapshot of `/var/lib/postgresql`. Postgres
 ## What it does
 
 - **Forks with the substrate.** Every byte under `/var/lib/postgresql` snapshots atomically; the new box boots on a CoW copy. No `pg_dump`, no replication topology, no wait.
-- **PgBouncer on the front, sized to load.** Transaction pooling and TLS termination on `:5432`; Postgres listens on the loopback only. The pooler runs one worker when idle and adds `so_reuseport` workers across cores as connection-handshake load rises, then reaps them when it falls. The worker count survives restarts.
+- **PgBouncer on the front, sized to load.** Transaction pooling and TLS termination on `:5432`; Postgres listens on the loopback only. The pooler runs one worker when idle and adds `so_reuseport` workers across cores as connection-handshake load rises, then reaps them when it falls. The worker count survives restarts. Workers peer, so a query-cancel the kernel routes to the wrong worker is forwarded to the one holding the session.
 - **Logical decoding from day one.** `wal_level = logical`, `max_wal_senders = 10`, `max_replication_slots = 10`. No primary restart to enable CDC later.
 - **Standard extensions, pinned.** pgvector, pgvectorscale, PostGIS, pg_cron, pg_partman, pg_jsonschema, hypopg, pg_repack, pg_search, pg_stat_statements, pg_trgm, auto_explain.
 - **Beyond extensions on the same volume.** `beyond-auth` and `beyond-queue` ship in the image and live under their own schemas in your database. Forking your DB forks their state automatically.

beyond-pg-core/src/mmds.rs

@@ -57,7 +57,9 @@ pub enum MmdsError {
     InvalidPassword,
     #[error("BEYOND_PG_PRIMARY_CONNINFO is required when BEYOND_PG_TIER=replica")]
     MissingPrimaryConninfo,
-    #[error("BEYOND_PG_RECOVERY_TARGET_TIME requires BEYOND_PG_WAL_SINK (the WAL source to replay from)")]
+    #[error(
+        "BEYOND_PG_RECOVERY_TARGET_TIME requires BEYOND_PG_WAL_SINK (the WAL source to replay from)"
+    )]
     RecoveryTargetWithoutWalSource,
     #[error("MMDS metadata not available: {0}")]
     Unavailable(String),
@@ -252,8 +254,8 @@ mod tests {
             .expect("should parse");
         assert_eq!(cfg.replication_password.as_deref(), Some("replpw"));
         // Empty string is treated as absent (same as the other optional keys).
-        let cfg2 = parse(base_meta(&[("BEYOND_PG_REPLICATION_PASSWORD", "")]))
-            .expect("should parse");
+        let cfg2 =
+            parse(base_meta(&[("BEYOND_PG_REPLICATION_PASSWORD", "")])).expect("should parse");
         assert!(cfg2.replication_password.is_none());
     }
 

beyond-pg-init/src/volumes.rs

@@ -41,7 +41,10 @@ pub fn mount_from_mmds() {
     if attachments.is_empty() {
         return;
     }
-    eprintln!("[init] mounting {} data volume(s) from MMDS", attachments.len());
+    eprintln!(
+        "[init] mounting {} data volume(s) from MMDS",
+        attachments.len()
+    );
     mount_data_volumes(&attachments);
 }
 
@@ -77,7 +80,10 @@ fn parse_attachments(val: &Value) -> Vec<AttachmentMeta> {
                 path: v["path"].as_str()?.to_string(),
                 fstype: v["fstype"].as_str().unwrap_or("ext4").to_string(),
                 readonly: v["readonly"].as_bool().unwrap_or(false),
-                storage_class: v["storage_class"].as_str().unwrap_or("standard").to_string(),
+                storage_class: v["storage_class"]
+                    .as_str()
+                    .unwrap_or("standard")
+                    .to_string(),
             })
         })
         .collect()
@@ -100,7 +106,10 @@ fn mount_data_volumes(attachments: &[AttachmentMeta]) {
 
 fn mount_one(a: &AttachmentMeta) -> Result<(), String> {
     if !wait_for_device(&a.device) {
-        return Err(format!("device {} not present after {DEVICE_WAIT:?}", a.device));
+        return Err(format!(
+            "device {} not present after {DEVICE_WAIT:?}",
+            a.device
+        ));
     }
 
     ensure_dir(&a.path)?;
@@ -182,7 +191,10 @@ fn is_mounted(path: &str, device: &str) -> Result<bool, String> {
 /// Return the device currently mounted at `path`, or `None` if nothing is.
 fn mounted_device(path: &str) -> Result<Option<String>, String> {
     let info = read_mountinfo()?;
-    Ok(info.into_iter().find(|(mp, _)| mp == path).map(|(_, dev)| dev))
+    Ok(info
+        .into_iter()
+        .find(|(mp, _)| mp == path)
+        .map(|(_, dev)| dev))
 }
 
 fn read_mountinfo() -> Result<Vec<(String, String)>, String> {
@@ -312,9 +324,13 @@ mod tests {
 
     #[test]
     fn parse_mountinfo_extracts_mountpoint_and_source() {
-        let line = "36 35 8:1 / /var/lib/postgresql rw,noatime shared:1 - ext4 /dev/vdb rw,data=ordered";
+        let line =
+            "36 35 8:1 / /var/lib/postgresql rw,noatime shared:1 - ext4 /dev/vdb rw,data=ordered";
         let entries = parse_mountinfo(line);
-        assert_eq!(entries, vec![("/var/lib/postgresql".to_string(), "/dev/vdb".to_string())]);
+        assert_eq!(
+            entries,
+            vec![("/var/lib/postgresql".to_string(), "/dev/vdb".to_string())]
+        );
     }
 
     #[test]

mise.toml

@@ -27,6 +27,10 @@ run = "cargo test --test boot -- --include-ignored --nocapture"
 description = "Run supervisor lifecycle tests (requires Docker + musl target + beyond-pg-test image)"
 run = "cargo test --test supervisor -- --include-ignored --nocapture"
 
+[tasks."test:peering"]
+description = "Run PgBouncer so_reuseport peering (cross-worker cancel) e2e test (requires Docker + beyond-pg-test image)"
+run = "cargo test --test pgbouncer_peering -- --include-ignored --nocapture --test-threads=1"
+
 [tasks."test:handoff"]
 description = "Run beyond-pg-init → supervisor handoff e2e tests (requires Docker + musl target + beyond-pg-test image)"
 run = "cargo test --release --test handoff_e2e -- --include-ignored --nocapture --test-threads=1"

packer/files/pgbouncer/pgbouncer.ini

@@ -33,8 +33,9 @@ ignore_startup_parameters = extra_float_digits,search_path
 logfile =
 pidfile =
 
-# No per-process unix socket or pidfile: beyond-pg runs 1..N so_reuseport workers that
-# share the TCP :5432 port, and pgbouncer defaults unix_socket_dir to /tmp — leaving it
-# set would make every worker collide on /tmp/.s.PGSQL.5432. Clients use TCP :5432; the
-# direct Postgres path is :5433.
-unix_socket_dir =
+# so_reuseport, unix_socket_dir, peer_id and the peer-routing map are appended per
+# worker at boot by beyond-pg (config.rs::pgbouncer_ini). beyond-pg runs 1..N
+# so_reuseport workers sharing TCP :5432; each gets a DISTINCT unix_socket_dir so
+# their peer sockets don't collide, which is what lets a query-cancel that the
+# kernel routes to the "wrong" worker be forwarded to the one holding the session.
+# Clients use TCP :5432; the direct Postgres path is :5433.

src/boot.rs

@@ -98,6 +98,26 @@ fn ensure_socket_dir() {
     }
 }
 
+/// Create a per-worker PgBouncer peer socket directory owned by `postgres`.
+/// Idempotent: create_dir_all is a no-op when it exists, and re-chowning is safe.
+/// The socket inside (`.s.PGSQL.5432`) is created by pgbouncer after it drops to
+/// the postgres user, so the directory must be writable by that user.
+fn ensure_peer_socket_dir(peer_id: usize) {
+    let dir = config::pgb_peer_socket_dir(peer_id);
+    if let Err(e) = std::fs::create_dir_all(&dir) {
+        warn!("ensure_peer_socket_dir: create {dir}: {e}");
+        return;
+    }
+    match std::process::Command::new("chown")
+        .args(["postgres:postgres", &dir])
+        .status()
+    {
+        Ok(s) if s.success() => info!("pgbouncer peer socket dir {dir} ready (owner postgres)"),
+        Ok(s) => warn!("chown {dir} exited {s}; continuing"),
+        Err(e) => warn!("chown {dir} failed to spawn: {e}; continuing"),
+    }
+}
+
 async fn do_boot_primary(cfg: &MmdsConfig) -> Result<(), BootError> {
     info!("boot: step 1/8 maybe_initdb");
     maybe_initdb(cfg).await?;
@@ -416,8 +436,7 @@ mod tests {
         let cfg = test_cfg(Some("http://10.0.0.5:9000"), Some("2026-05-14 03:00:00"));
         write_pitr_config_into(&cfg, pgdata.path()).unwrap();
 
-        let content =
-            std::fs::read_to_string(pgdata.path().join("conf.d/05-pitr.conf")).unwrap();
+        let content = std::fs::read_to_string(pgdata.path().join("conf.d/05-pitr.conf")).unwrap();
         assert!(
             content.contains("restore_command = 'curl -f -s http://10.0.0.5:9000/%f -o %p'"),
             "restore_command wrong: {content}"
@@ -798,15 +817,26 @@ fn write_config_files(cfg: &MmdsConfig, tls: &crate::tls::TlsConfig) -> Result<(
     };
     write_atomic(Path::new(PG_HBA_PATH), &hba)?;
 
-    // pgbouncer.ini — overwritten every boot, with client_tls_* keys pointing
-    // at the same cert Postgres uses.
+    // pgbouncer.ini — one config per possible so_reuseport worker, overwritten every
+    // boot, with client_tls_* keys pointing at the same cert Postgres uses. Workers
+    // are spawned lazily by the scaler, but all their configs (and peer socket dirs)
+    // are laid down up front so the shared [peers] map is stable and a scaled-up
+    // worker never waits on config. peer_id 1 keeps the canonical pgbouncer.ini.
     if let Some(parent) = Path::new(PGBOUNCER_INI_PATH).parent() {
         std::fs::create_dir_all(parent)?;
     }
-    write_atomic(
-        Path::new(PGBOUNCER_INI_PATH),
-        &config::pgbouncer_ini(tls, cfg.ram_bytes, cfg.vcpus),
-    )?;
+    let max_workers = config::pgbouncer_max_workers(cfg.vcpus);
+    for peer_id in 1..=max_workers {
+        write_atomic(
+            Path::new(&config::pgbouncer_ini_path(peer_id)),
+            &config::pgbouncer_ini(tls, cfg.ram_bytes, cfg.vcpus, peer_id),
+        )?;
+        // Peer socket dir must exist and be writable by the postgres user pgbouncer
+        // drops to (only created when peering is active, i.e. max_workers > 1).
+        if max_workers > 1 {
+            ensure_peer_socket_dir(peer_id);
+        }
+    }
 
     Ok(())
 }