git ents: feat: add refstore-postgres, the Postgres RefStore backend

Cargo.lock @@ -465,6 +465,17 @@ "winnow", ] +[[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -666,6 +677,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", +] + [[package]] name = "chrono" version = "0.4.45" @@ -688,6 +710,12 @@ "hashbrown 0.16.1", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "comfy-table" version = "7.2.2" @@ -874,6 +902,15 @@ "memchr", ] +[[package]] +name = "ctutils" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e" +dependencies = [ + "cmov", +] + [[package]] name = "dashmap" version = "6.2.1" @@ -972,6 +1009,7 @@ "block-buffer 0.12.1", "const-oid", "crypto-common 0.2.2", + "ctutils", ] [[package]] @@ -1238,6 +1276,12 @@ "strsim", ] +[[package]] +name = "fallible-iterator" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7" + [[package]] name = "faster-hex" version = "0.10.0" @@ -1354,6 +1398,7 @@ checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -1423,7 +1468,7 @@ dependencies = [ "cfg-if", "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", ] [[package]] @@ -1447,6 +1492,7 @@ "cfg-if", "libc", "r-efi 6.0.0", + "rand_core 0.10.1", ] [[package]] @@ -2605,6 +2651,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest 0.11.3", +] + [[package]] name = "http" version = "1.4.2" @@ -3004,6 +3059,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "404671567c0ac43389a69ae424eefc89390a25d3a5386a54249cdaf13e01b411" +[[package]] +name = "libredox" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c943259e342f1e06ff2da7a83eabdfe7f92ce10262688dbf1895ff0b3e6e4652" +dependencies = [ + "libc", +] + [[package]] name = "linked-hash-map" version = "0.5.6" @@ -3084,6 +3148,16 @@ "syn", ] +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.3", +] + [[package]] name = "memchr" version = "2.8.2" @@ -3129,7 +3203,7 @@ dependencies = [ "libc", "log", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.61.2", ] @@ -3194,6 +3268,24 @@ "autocfg", ] +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags 2.13.0", +] + +[[package]] +name = "objc2-system-configuration" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396" +dependencies = [ + "objc2-core-foundation", +] + [[package]] name = "odb-files" version = "0.0.0" @@ -3302,6 +3394,25 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "phf" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1562dc717473dbaa4c1f85a36410e03c047b2e7df7f45ee938fbef64ae7fadf" +dependencies = [ + "phf_shared", + "serde", +] + +[[package]] +name = "phf_shared" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e57fef6bc5981e38c2ce2d63bfa546861309f875b8a75f092d1d54ae2d64f266" +dependencies = [ + "siphasher", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -3363,6 +3474,35 @@ "winreg", ] +[[package]] +name = "postgres-protocol" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514" +dependencies = [ + "base64", + "byteorder", + "bytes", + "fallible-iterator", + "hmac", + "md-5", + "memchr", + "rand 0.10.2", + "sha2", + "stringprep", +] + +[[package]] +name = "postgres-types" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "851ca9db4932932d69f3ea811b1abe63087a0f740a47692619dd40d4899b68be" +dependencies = [ + "bytes", + "fallible-iterator", + "postgres-protocol", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -3483,7 +3623,18 @@ checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" dependencies = [ "rand_chacha", - "rand_core", + "rand_core 0.9.5", +] + +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", ] [[package]] @@ -3493,7 +3644,7 @@ checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.9.5", ] [[package]] @@ -3505,6 +3656,12 @@ "getrandom 0.3.4", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "rayon" version = "1.12.0" @@ -3545,6 +3702,18 @@ "gix-hash", ] +[[package]] +name = "refstore-postgres" +version = "0.0.0" +dependencies = [ + "backend-conformance", + "git-backend", + "gix-hash", + "tokio", + "tokio-postgres", + "uuid", +] + [[package]] name = "regex" version = "1.12.4" @@ -3915,6 +4084,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" +[[package]] +name = "siphasher" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" + [[package]] name = "slab" version = "0.4.12" @@ -3970,6 +4145,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b2231b7c3057d5e4ad0156fb3dc807d900806020c5ffa3ee6ff2c8c76fb8520" +[[package]] +name = "stringprep" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", + "unicode-properties", +] + [[package]] name = "strip-ansi-escapes" version = "0.2.1" @@ -4212,6 +4398,32 @@ "syn", ] +[[package]] +name = "tokio-postgres" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a528f7d280f6d5b9cd149635c8705b0dd049754bc67d81d31fa25169a93809d3" +dependencies = [ + "async-trait", + "byteorder", + "bytes", + "fallible-iterator", + "futures-channel", + "futures-util", + "log", + "parking_lot", + "percent-encoding", + "phf", + "pin-project-lite", + "postgres-protocol", + "postgres-types", + "rand 0.10.2", + "socket2", + "tokio", + "tokio-util", + "whoami", +] + [[package]] name = "tokio-rustls" version = "0.26.4" @@ -4238,6 +4450,19 @@ "webpki-roots 0.26.11", ] +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -4345,7 +4570,7 @@ "http", "httparse", "log", - "rand", + "rand 0.9.4", "rustls", "rustls-pki-types", "sha1", @@ -4373,6 +4598,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" +[[package]] +name = "unicode-bidi" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" + [[package]] name = "unicode-bom" version = "2.0.3" @@ -4394,6 +4625,12 @@ "tinyvec", ] +[[package]] +name = "unicode-properties" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" + [[package]] name = "unicode-segmentation" version = "1.13.3" @@ -4519,6 +4756,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasi" +version = "0.14.7+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c" +dependencies = [ + "wasip2", +] + [[package]] name = "wasip2" version = "1.0.4+wasi-0.2.12" @@ -4528,6 +4774,15 @@ "wit-bindgen", ] +[[package]] +name = "wasite" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fe902b4a6b8028a753d5424909b764ccf79b7a209eac9bf97e59cda9f71a42" +dependencies = [ + "wasi 0.14.7+wasi-0.2.4", +] + [[package]] name = "wasm-bindgen" version = "0.2.126" @@ -4579,6 +4834,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30261027ea4225cf7c8287aa70bd103c1e78508606c77caf24bb468326ce7b5f" +[[package]] +name = "web-sys" +version = "0.3.103" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "webpki-roots" version = "0.26.11" @@ -4597,6 +4862,19 @@ "rustls-pki-types", ] +[[package]] +name = "whoami" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "998767ef88740d1f5b0682a9c53c24431453923962269c2db68ee43788c5a40d" +dependencies = [ + "libc", + "libredox", + "objc2-system-configuration", + "wasite", + "web-sys", +] + [[package]] name = "winapi" version = "0.3.9"

Cargo.toml @@ -16,6 +16,7 @@ "crates/git-toolchain", "crates/odb-files", "crates/refstore-files", + "crates/refstore-postgres", ] [workspace.package] @@ -73,6 +74,7 @@ git-toolchain = { path = "crates/git-toolchain" } odb-files = { path = "crates/odb-files" } refstore-files = { path = "crates/refstore-files" } +refstore-postgres = { path = "crates/refstore-postgres" } gix = "0.84" gix-actor = "0.41" gix-date = "0.15" @@ -99,6 +101,9 @@ "io-util", "sync", ] } +tokio-postgres = { version = "0.7", default-features = false, features = [ + "runtime", +] } uuid = { version = "1", features = ["v4"] } # These lint configurations were originally pulled from [Evan Schwartz][1].

crates/refstore-postgres/Cargo.toml @@ -1,0 +1,19 @@ +[package] +name = "refstore-postgres" +version = "0.0.0" +edition.workspace = true +publish.workspace = true +license.workspace = true + +[dependencies] +git-backend = { workspace = true } +gix-hash = { workspace = true } +tokio = { workspace = true } +tokio-postgres = { workspace = true } + +[dev-dependencies] +backend-conformance = { workspace = true } +uuid = { workspace = true } + +[lints] +workspace = true

crates/refstore-postgres/migrations/0001_init.sql @@ -1,0 +1,75 @@ +-- Schema for `refstore-postgres` (`docs/scale-out.adoc`, "RefStore" / WS4). +-- Applied idempotently: every statement is guarded so running this file +-- against an already-migrated database is a no-op. + +-- One row per ref. The primary key is also the prefix-iteration index: a +-- `text_pattern_ops` index makes `LIKE 'prefix%'` scans (used by +-- `iter_prefix`) index-backed regardless of the database's default locale, +-- since that opclass compares raw bytes rather than collated text. +CREATE TABLE IF NOT EXISTS git_ents_refs ( + repo_id TEXT NOT NULL, + name TEXT NOT NULL, + oid TEXT NOT NULL, + PRIMARY KEY (repo_id, name) +); + +CREATE INDEX IF NOT EXISTS git_ents_refs_prefix_idx + ON git_ents_refs (repo_id, name text_pattern_ops); + +-- Append-only reflog: one row per applied `RefEdit`, written in the same SQL +-- transaction as the ref mutation it records. +CREATE TABLE IF NOT EXISTS git_ents_reflog ( + id BIGSERIAL PRIMARY KEY, + repo_id TEXT NOT NULL, + name TEXT NOT NULL, + old_oid TEXT, + new_oid TEXT, + message TEXT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS git_ents_reflog_lookup_idx + ON git_ents_reflog (repo_id, name, id DESC); + +-- Pack registry: minimal columns only. WS5 (Tigris object store) consumes +-- this to record promotion of staged packs; nothing here reads it yet. +CREATE TABLE IF NOT EXISTS git_ents_pack_registry ( + id BIGSERIAL PRIMARY KEY, + repo_id TEXT NOT NULL, + location_key TEXT NOT NULL, + promoted_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS git_ents_pack_registry_repo_idx + ON git_ents_pack_registry (repo_id); + +-- Effect queue: the at-least-once source of truth `watch`'s NOTIFY hint +-- points consumers back at (`docs/scale-out.adoc`, "RefStore": "the effect +-- queue table ... is the source of truth"). +CREATE TABLE IF NOT EXISTS git_ents_effect_queue ( + id BIGSERIAL PRIMARY KEY, + repo_id TEXT NOT NULL, + payload TEXT NOT NULL, + state TEXT NOT NULL DEFAULT 'enqueued' + CHECK (state IN ('enqueued', 'claimed', 'done')), + claimed_by TEXT, + enqueued_at TIMESTAMPTZ NOT NULL DEFAULT now(), + claimed_at TIMESTAMPTZ, + done_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS git_ents_effect_queue_state_idx + ON git_ents_effect_queue (repo_id, state, id); + +-- Op records index: one row per accepted push, pointing at the op record +-- object (the push-cert-plus-outcome artifact `git-protocol` builds via +-- `attestation::build_op_record`) by OID. +CREATE TABLE IF NOT EXISTS git_ents_op_records ( + id BIGSERIAL PRIMARY KEY, + repo_id TEXT NOT NULL, + op_oid TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS git_ents_op_records_repo_idx + ON git_ents_op_records (repo_id, created_at DESC);

crates/refstore-postgres/src/lib.rs @@ -1,0 +1,136 @@ +//! [`git_backend::RefStore`] over Postgres — the cloud default backend +//! (`docs/scale-out.adoc`, "RefStore" / WS4). +//! +//! A row per ref, keyed `(repo_id, name)`; `transaction` is one SQL +//! transaction of conditional `UPDATE`/`INSERT`/`DELETE`s, every edit's +//! [`git_backend::Expected`] precondition checked by the statement's own +//! `WHERE` clause and reported via `RETURNING` (see [`ref_store`]). +//! `watch` is `LISTEN`/`NOTIFY`, fired on commit from `transaction`, and +//! remains a wakeup hint only — the `git_ents_effect_queue` table (see +//! [`queue`]) is what an at-least-once consumer actually drains (see +//! [`notify`], and the trait contract on [`git_backend::RefStore::watch`]). +//! +//! # Q1: single write-primary +//! +//! This store assumes exactly one writable Postgres primary at a time. Fly +//! managed-Postgres failover semantics must guarantee a fenced single +//! primary or synchronous replication before this backend is deployed +//! against it (`docs/scale-out.adoc`, Q1 / "Non-goals": "no distributed ref +//! consensus"). A split-brain primary — two writers both believing they +//! hold it — is the one unrecoverable failure mode for `RefStore`: two +//! primaries would each serialize CAS locally and correctly, but the two +//! serializations could disagree, which no amount of correct SQL here can +//! detect or repair. Enforcing single-primary is deployment configuration +//! (`fly-replay`, replica topology), not this crate's job — there is no +//! fencing code here, deliberately. + +mod notify; +mod queue; +mod ref_store; + +pub use queue::{ClaimedEffect, EffectId}; + +use git_backend::{Error, Result}; + +/// Migration SQL applied idempotently by [`PostgresRefStore::migrate`]: +/// refs, reflog, pack registry, effect queue, op records (`docs/ +/// scale-out.adoc`, WS4's schema list). +const MIGRATION_SQL: &str = include_str!("../migrations/0001_init.sql"); + +/// The reflog message every transaction's rows carry, mirroring +/// `refstore-files`' fixed `LOG_MESSAGE` — a write through this backend is +/// self-contained, not dependent on caller-supplied metadata. +const LOG_MESSAGE: &str = "git-backend: transaction"; + +/// [`git_backend::RefStore`] over a Postgres database: one row per ref, +/// scoped to a single `repo_id` (`docs/scale-out.adoc`'s "namespace per +/// repo" rule — no cross-repo query this store issues ever omits the +/// `repo_id` filter). +/// +/// Holds one [`tokio_postgres::Client`] behind a [`tokio::sync::Mutex`], +/// per the dependency policy: no connection pool. `transaction` needs +/// exclusive use of the connection for the duration of its SQL transaction +/// (two overlapping `BEGIN`s on one session would corrupt each other), and +/// serializing every other method through the same lock keeps the whole +/// store's concurrency story in one place rather than reasoning about which +/// methods are safe to interleave. +/// +/// Owns a dedicated [`tokio::runtime::Runtime`] so [`git_backend::RefStore`] +/// (a sync trait) can drive [`tokio_postgres`]'s async client; every trait +/// method is a `block_on` call. +pub struct PostgresRefStore { + runtime: tokio::runtime::Runtime, + client: tokio::sync::Mutex<tokio_postgres::Client>, + repo_id: String, + notify: tokio::sync::broadcast::Sender<String>, +} + +/// Map a [`tokio_postgres::Error`] onto this crate's shared [`Error`] type. +fn pg_err(error: tokio_postgres::Error) -> Error { + Error::RefStore(error.to_string()) +} + +impl PostgresRefStore { + /// Connect to `conninfo` (a libpq connection string, e.g. `"host=... + /// user=... dbname=..."`), scope every operation to `repo_id`, apply the + /// migration (see [`Self::migrate`]), and start listening for this + /// store's `NOTIFY` channel. + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the connection, migration, or initial + /// `LISTEN` fails, or if the dedicated Tokio runtime cannot be created. + pub fn connect(conninfo: &str, repo_id: impl Into<String>) -> Result<Self> { + let runtime = + tokio::runtime::Runtime::new().map_err(|error| Error::RefStore(error.to_string()))?; + let (client, connection) = runtime + .block_on(tokio_postgres::connect(conninfo, tokio_postgres::NoTls)) + .map_err(pg_err)?; + + let (notify_tx, _receiver) = tokio::sync::broadcast::channel(notify::CHANNEL_CAPACITY); + runtime.spawn(notify::pump(connection, notify_tx.clone())); + + let store = Self { + runtime, + client: tokio::sync::Mutex::new(client), + repo_id: repo_id.into(), + notify: notify_tx, + }; + store.migrate()?; + store.listen()?; + Ok(store) + } + + /// Apply the embedded migration SQL. Every statement is guarded (`CREATE + /// TABLE IF NOT EXISTS`, `CREATE INDEX IF NOT EXISTS`), so calling this + /// again against an already-migrated database is a no-op. [`Self::connect`] + /// already calls this; exposed for callers that want migration as an + /// explicit, separately-timed step (e.g. a deploy hook run once ahead of + /// bringing up store instances). + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if any migration statement fails. + pub fn migrate(&self) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client.batch_execute(MIGRATION_SQL).await + }) + .map_err(pg_err) + } + + /// Issue this store's `LISTEN`, so `NOTIFY`s fired by any store (this + /// one or another process's) against the same channel reach this + /// connection's [`notify::pump`]. + fn listen(&self) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .batch_execute(&format!("LISTEN {}", notify::CHANNEL)) + .await + }) + .map_err(pg_err) + } +}

crates/refstore-postgres/src/notify.rs @@ -1,0 +1,91 @@ +//! `LISTEN`/`NOTIFY` plumbing behind [`crate::PostgresRefStore::watch`]. +//! +//! Postgres delivers notifications as out-of-band messages on the very +//! connection that issued `LISTEN`, surfaced through +//! [`tokio_postgres::Connection::poll_message`] rather than through +//! [`tokio_postgres::Client`]'s normal request/response methods. [`pump`] +//! drives that connection for the lifetime of the store, broadcasting every +//! payload to whichever [`crate::PostgresRefStore::watch`] calls are +//! currently subscribed; nothing here is trusted as a delivery guarantee — +//! see the trait-level contract on [`git_backend::RefStore::watch`]. + +use tokio::sync::broadcast; +use tokio_postgres::AsyncMessage; + +/// The fixed channel every store `LISTEN`s on and `NOTIFY`s through. One +/// physical database can host many repositories' stores; each notification +/// payload is prefixed with its repo id so listeners on this shared channel +/// can filter out other repos' traffic (see [`decode`]). +pub(crate) const CHANNEL: &str = "git_ents_refstore"; + +/// How many undelivered payloads a slow [`git_backend::RefStore::watch`] +/// subscriber tolerates before it starts missing them. `watch` is a hint +/// only, so a lagging subscriber is told to assume something changed +/// (see [`crate::ref_store::bridge_watch`]) rather than silently losing +/// state. +pub(crate) const CHANNEL_CAPACITY: usize = 256; + +/// Drive `connection`'s asynchronous messages for as long as the store +/// lives, forwarding every `NOTIFY` payload to `sender`. Errors and a closed +/// connection both end the pump silently: with nobody left to hand the +/// error to, the only correct move is to stop, which simply turns every +/// subsequent `watch` subscriber's hint stream quiet — never incorrect, +/// per the trait's best-effort contract. +pub(crate) async fn pump<S, T>( + mut connection: tokio_postgres::Connection<S, T>, + sender: broadcast::Sender<String>, +) where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, + T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + loop { + let message = std::future::poll_fn(|cx| connection.poll_message(cx)).await; + match message { + Some(Ok(AsyncMessage::Notification(notification))) => { + let _ignored_no_subscribers = sender.send(notification.payload().to_owned()); + } + Some(Ok(_)) => {} + Some(Err(_)) | None => break, + } + } +} + +/// Build the payload one [`crate::ref_store`] transaction `NOTIFY`s with: +/// the repo id, then one changed ref name per line. Ref names cannot +/// contain control characters, so `\n` is a safe separator with no +/// escaping needed. +pub(crate) fn encode(repo_id: &str, changed_names: &[String]) -> String { + let mut payload = String::from(repo_id); + for name in changed_names { + payload.push('\n'); + payload.push_str(name); + } + payload +} + +/// Whether `payload` (as built by [`encode`]) reports a change for +/// `repo_id` under `prefix`. +pub(crate) fn matches(payload: &str, repo_id: &str, prefix: &str) -> bool { + let mut lines = payload.split('\n'); + if lines.next() != Some(repo_id) { + return false; + } + lines.any(|name| name.starts_with(prefix)) +} + +#[cfg(test)] +mod tests { + use super::{encode, matches}; + + #[test] + fn matches_filters_by_repo_and_prefix() { + let payload = encode( + "repo-a", + &["refs/heads/main".to_owned(), "refs/meta/x".to_owned()], + ); + assert!(matches(&payload, "repo-a", "refs/heads/")); + assert!(matches(&payload, "repo-a", "refs/meta/")); + assert!(!matches(&payload, "repo-a", "refs/cache/")); + assert!(!matches(&payload, "repo-b", "refs/heads/")); + } +}

crates/refstore-postgres/src/queue.rs @@ -1,0 +1,167 @@ +//! Minimal Rust surface over `git_ents_effect_queue` and +//! `git_ents_op_records` (`docs/scale-out.adoc`, WS4's schema list). Neither +//! table is part of the [`git_backend::RefStore`] contract; these methods +//! exist so the schema is exercised end-to-end rather than asserted only by +//! its `CREATE TABLE` statements, and so a future dispatcher (WS7) and +//! `git-protocol` (op records) have something to call. + +use git_backend::{Error, Result}; +use gix_hash::ObjectId; + +use crate::{PostgresRefStore, pg_err}; + +/// The primary key of a row in `git_ents_effect_queue`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EffectId(i64); + +/// One row claimed off the effect queue by [`PostgresRefStore::claim_effects`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClaimedEffect { + /// The claimed row's id, needed to later call + /// [`PostgresRefStore::complete_effect`]. + pub id: EffectId, + /// The effect payload enqueued by [`PostgresRefStore::enqueue_effect`]. + pub payload: String, +} + +impl PostgresRefStore { + /// Append one effect payload to this store's repo-scoped queue in state + /// `enqueued`, returning its id. + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the insert fails. + pub fn enqueue_effect(&self, payload: &str) -> Result<EffectId> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .query_one( + "INSERT INTO git_ents_effect_queue (repo_id, payload) + VALUES ($1, $2) RETURNING id", + &[&self.repo_id, &payload], + ) + .await + }) + .map_err(pg_err) + .and_then(|row| row.try_get::<_, i64>(0).map(EffectId).map_err(pg_err)) + } + + /// Atomically claim up to `limit` of this store's oldest `enqueued` + /// rows for `claimed_by`, marking them `claimed` so a concurrent + /// dispatcher never double-claims them (`FOR UPDATE SKIP LOCKED`). This + /// is the at-least-once queue the doc mandates: rows survive here + /// independent of any `watch` subscriber's connection. + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the claim query fails. + pub fn claim_effects(&self, claimed_by: &str, limit: i64) -> Result<Vec<ClaimedEffect>> { + let rows = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query( + "UPDATE git_ents_effect_queue + SET state = 'claimed', claimed_by = $1, claimed_at = now() + WHERE id IN ( + SELECT id FROM git_ents_effect_queue + WHERE repo_id = $2 AND state = 'enqueued' + ORDER BY id + LIMIT $3 + FOR UPDATE SKIP LOCKED + ) + RETURNING id, payload", + &[&claimed_by, &self.repo_id, &limit], + ) + .await + }) + .map_err(pg_err)?; + + rows.into_iter() + .map(|row| { + let id: i64 = row.try_get(0).map_err(pg_err)?; + let payload: String = row.try_get(1).map_err(pg_err)?; + Ok(ClaimedEffect { + id: EffectId(id), + payload, + }) + }) + .collect() + } + + /// Mark a previously [`Self::claim_effects`]-claimed row `done`. + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the update fails. + pub fn complete_effect(&self, id: EffectId) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .execute( + "UPDATE git_ents_effect_queue SET state = 'done', done_at = now() + WHERE id = $1 AND repo_id = $2", + &[&id.0, &self.repo_id], + ) + .await + }) + .map_err(pg_err) + .map(|_rows_affected| ()) + } + + /// Record one accepted push's op record OID (`docs/scale-out.adoc`, + /// "Attested push": "Push ID = op record OID, uniformly"). + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the insert fails. + pub fn record_op(&self, op_oid: ObjectId) -> Result<()> { + let hex = op_oid.to_hex().to_string(); + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .execute( + "INSERT INTO git_ents_op_records (repo_id, op_oid) VALUES ($1, $2)", + &[&self.repo_id, &hex], + ) + .await + }) + .map_err(pg_err) + .map(|_rows_affected| ()) + } + + /// This store's recorded op record OIDs, most recent first. + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the query fails, or if a stored OID is + /// not valid hex (a corrupted row, never written by + /// [`Self::record_op`]). + pub fn op_records(&self) -> Result<Vec<ObjectId>> { + let rows = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query( + "SELECT op_oid FROM git_ents_op_records + WHERE repo_id = $1 ORDER BY created_at DESC, id DESC", + &[&self.repo_id], + ) + .await + }) + .map_err(pg_err)?; + + rows.into_iter() + .map(|row| { + let hex: String = row.try_get(0).map_err(pg_err)?; + ObjectId::from_hex(hex.as_bytes()) + .map_err(|error| Error::RefStore(error.to_string())) + }) + .collect() + } +}

crates/refstore-postgres/src/ref_store.rs @@ -1,0 +1,338 @@ +//! [`git_backend::RefStore`] for [`crate::PostgresRefStore`]: transaction = +//! one SQL transaction of conditional `UPDATE`/`INSERT`/`DELETE`s, per +//! `docs/scale-out.adoc`'s `refstore-postgres` row. + +use git_backend::{ + Error, Expected, RefEdit, RefEventStream, RefIter, RefLogEntry, RefLogIter, RefName, RefStore, + Result, TxOutcome, +}; +use gix_hash::ObjectId; +use tokio_postgres::Transaction; + +use crate::{LOG_MESSAGE, PostgresRefStore, notify, pg_err}; + +/// The outcome of applying one [`RefEdit`] inside a transaction: whether its +/// [`Expected`] precondition held and, if so, what actually changed (for the +/// reflog row appended alongside it). +enum EditResult { + /// The precondition failed; the whole transaction must roll back. + Mismatch, + /// The precondition held but the edit was a no-op (e.g. `MustNotExist` + /// against a ref that already didn't exist) — nothing to log. + NoChange, + /// The precondition held and the ref's value changed from `old` to + /// `new`. + Changed { + /// The ref's hex oid before this edit, or `None` if it did not + /// exist. + old: Option<String>, + /// The ref's hex oid after this edit, or `None` if it was deleted. + new: Option<String>, + }, +} + +/// Parse a hex string read back from a `TEXT` oid column. +fn parse_oid(hex: &str) -> Result<ObjectId> { + ObjectId::from_hex(hex.as_bytes()).map_err(|error| Error::RefStore(error.to_string())) +} + +/// Escape `%`, `_`, and `\` in `prefix` for use in a `LIKE ... ESCAPE '\'` +/// pattern, then append the wildcard `%`. +fn like_pattern(prefix: &str) -> String { + let mut pattern = String::with_capacity(prefix.len().saturating_add(1)); + for ch in prefix.chars() { + if matches!(ch, '\\' | '%' | '_') { + pattern.push('\\'); + } + pattern.push(ch); + } + pattern.push('%'); + pattern +} + +/// Apply one [`RefEdit`] inside `tx`, returning what happened. Every branch +/// is a single conditional statement whose row count (via `RETURNING`) +/// reports whether the precondition held — no separate lock-then-check step +/// is needed; Postgres's own row-level locking on the `UPDATE`/`DELETE`/ +/// unique-conflicting `INSERT` makes each branch atomic on its own. +async fn apply_edit( + tx: &Transaction<'_>, + repo_id: &str, + edit: &RefEdit, +) -> std::result::Result<EditResult, tokio_postgres::Error> { + let name = edit.name.as_str(); + let new_hex = edit.new.map(|oid| oid.to_hex().to_string()); + + match (&edit.expected, &new_hex) { + (Expected::Any, Some(new)) => { + let old = tx + .query_opt( + "SELECT oid FROM git_ents_refs WHERE repo_id = $1 AND name = $2", + &[&repo_id, &name], + ) + .await? + .map(|row| row.try_get::<_, String>(0)) + .transpose()?; + tx.execute( + "INSERT INTO git_ents_refs (repo_id, name, oid) VALUES ($1, $2, $3) + ON CONFLICT (repo_id, name) DO UPDATE SET oid = EXCLUDED.oid", + &[&repo_id, &name, new], + ) + .await?; + Ok(EditResult::Changed { + old, + new: Some(new.clone()), + }) + } + (Expected::Any, None) => { + let old = tx + .query_opt( + "SELECT oid FROM git_ents_refs WHERE repo_id = $1 AND name = $2", + &[&repo_id, &name], + ) + .await? + .map(|row| row.try_get::<_, String>(0)) + .transpose()?; + if old.is_none() { + return Ok(EditResult::NoChange); + } + tx.execute( + "DELETE FROM git_ents_refs WHERE repo_id = $1 AND name = $2", + &[&repo_id, &name], + ) + .await?; + Ok(EditResult::Changed { old, new: None }) + } + (Expected::MustNotExist, Some(new)) => { + let row = tx + .query_opt( + "INSERT INTO git_ents_refs (repo_id, name, oid) VALUES ($1, $2, $3) + ON CONFLICT (repo_id, name) DO NOTHING RETURNING oid", + &[&repo_id, &name, new], + ) + .await?; + Ok(match row { + Some(_) => EditResult::Changed { + old: None, + new: Some(new.clone()), + }, + None => EditResult::Mismatch, + }) + } + (Expected::MustNotExist, None) => { + let row = tx + .query_opt( + "SELECT 1 FROM git_ents_refs WHERE repo_id = $1 AND name = $2", + &[&repo_id, &name], + ) + .await?; + Ok(if row.is_none() { + EditResult::NoChange + } else { + EditResult::Mismatch + }) + } + (Expected::MustExistAndMatch(expected), Some(new)) => { + let expected_hex = expected.to_hex().to_string(); + let row = tx + .query_opt( + "UPDATE git_ents_refs SET oid = $3 WHERE repo_id = $1 AND name = $2 AND oid = $4 + RETURNING oid", + &[&repo_id, &name, new, &expected_hex], + ) + .await?; + Ok(match row { + Some(_) => EditResult::Changed { + old: Some(expected_hex), + new: Some(new.clone()), + }, + None => EditResult::Mismatch, + }) + } + (Expected::MustExistAndMatch(expected), None) => { + let expected_hex = expected.to_hex().to_string(); + let row = tx + .query_opt( + "DELETE FROM git_ents_refs WHERE repo_id = $1 AND name = $2 AND oid = $3 + RETURNING oid", + &[&repo_id, &name, &expected_hex], + ) + .await?; + Ok(match row { + Some(_) => EditResult::Changed { + old: Some(expected_hex), + new: None, + }, + None => EditResult::Mismatch, + }) + } + } +} + +impl RefStore for PostgresRefStore { + fn get(&self, name: &RefName) -> Result<Option<ObjectId>> { + self.runtime.block_on(async { + let client = self.client.lock().await; + let row = client + .query_opt( + "SELECT oid FROM git_ents_refs WHERE repo_id = $1 AND name = $2", + &[&self.repo_id, &name.as_str()], + ) + .await + .map_err(pg_err)?; + match row { + Some(row) => { + let hex: String = row.try_get(0).map_err(pg_err)?; + parse_oid(&hex).map(Some) + } + None => Ok(None), + } + }) + } + + fn iter_prefix(&self, prefix: &RefName) -> Result<RefIter> { + let pattern = like_pattern(prefix.as_str()); + let rows = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query( + "SELECT name, oid FROM git_ents_refs + WHERE repo_id = $1 AND name LIKE $2 ESCAPE '\\' + ORDER BY name", + &[&self.repo_id, &pattern], + ) + .await + }) + .map_err(pg_err)?; + + let mut out = Vec::with_capacity(rows.len()); + for row in rows { + out.push((|| { + let name: String = row.try_get(0).map_err(pg_err)?; + let hex: String = row.try_get(1).map_err(pg_err)?; + Ok((RefName::new(name), parse_oid(&hex)?)) + })()); + } + Ok(RefIter::new(out.into_iter())) + } + + fn transaction(&self, edits: &[RefEdit]) -> Result<TxOutcome> { + self.runtime.block_on(async { + let mut client = self.client.lock().await; + let tx = client.transaction().await.map_err(pg_err)?; + + let mut changed_names = Vec::new(); + for edit in edits { + match apply_edit(&tx, &self.repo_id, edit).await.map_err(pg_err)? { + EditResult::Mismatch => { + tx.rollback().await.map_err(pg_err)?; + return Ok(TxOutcome::Rejected { + name: edit.name.clone(), + }); + } + EditResult::NoChange => {} + EditResult::Changed { old, new } => { + tx.execute( + "INSERT INTO git_ents_reflog + (repo_id, name, old_oid, new_oid, message) + VALUES ($1, $2, $3, $4, $5)", + &[&self.repo_id, &edit.name.as_str(), &old, &new, &LOG_MESSAGE], + ) + .await + .map_err(pg_err)?; + changed_names.push(edit.name.as_str().to_owned()); + } + } + } + + if !changed_names.is_empty() { + let payload = notify::encode(&self.repo_id, &changed_names); + tx.execute("SELECT pg_notify($1, $2)", &[&notify::CHANNEL, &payload]) + .await + .map_err(pg_err)?; + } + tx.commit().await.map_err(pg_err)?; + Ok(TxOutcome::Applied) + }) + } + + fn watch(&self, prefix: &RefName) -> Result<RefEventStream> { + Ok(bridge_watch( + &self.runtime, + self.notify.subscribe(), + self.repo_id.clone(), + prefix.as_str().to_owned(), + )) + } + + fn log(&self, name: &RefName) -> Result<RefLogIter> { + let rows = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query( + "SELECT old_oid, new_oid, message, + extract(epoch from recorded_at)::bigint + FROM git_ents_reflog + WHERE repo_id = $1 AND name = $2 + ORDER BY id DESC", + &[&self.repo_id, &name.as_str()], + ) + .await + }) + .map_err(pg_err)?; + + let mut entries = Vec::with_capacity(rows.len()); + for row in rows { + entries.push((|| { + let old: Option<String> = row.try_get(0).map_err(pg_err)?; + let new: Option<String> = row.try_get(1).map_err(pg_err)?; + let message: String = row.try_get(2).map_err(pg_err)?; + let seconds: i64 = row.try_get(3).map_err(pg_err)?; + Ok(RefLogEntry { + old: old.as_deref().map(parse_oid).transpose()?, + new: new.as_deref().map(parse_oid).transpose()?, + message, + seconds: u64::try_from(seconds).unwrap_or(0), + }) + })()); + } + Ok(RefLogIter::new(entries.into_iter())) + } +} + +/// Bridge a [`tokio::sync::broadcast::Receiver`] of raw `NOTIFY` payloads +/// into the blocking [`RefEventStream`] the trait exposes: a background task +/// filters every payload by `repo_id`/`prefix` and forwards a match as a +/// [`git_backend::RefEvent`]. A lagged receiver (the subscriber fell behind +/// and missed payloads) still emits a hint — per `watch`'s contract, a +/// consumer never trusts hint precision, only that draining its own state +/// is due. +fn bridge_watch( + runtime: &tokio::runtime::Runtime, + mut receiver: tokio::sync::broadcast::Receiver<String>, + repo_id: String, + prefix: String, +) -> RefEventStream { + let (sender, events) = std::sync::mpsc::channel(); + runtime.spawn(async move { + loop { + match receiver.recv().await { + Ok(payload) => { + if !notify::matches(&payload, &repo_id, &prefix) { + continue; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + if sender.send(git_backend::RefEvent).is_err() { + break; + } + } + }); + RefEventStream::new(events) +}

crates/refstore-postgres/tests/conformance.rs @@ -1,0 +1,202 @@ +//! Conformance and targeted tests for `refstore-postgres` +//! (`docs/scale-out.adoc`, WS4 / WS2), gated on a reachable Postgres: +//! +//! 1. `GIT_ENTS_TEST_POSTGRES_URL`, if set — an already-running Postgres. +//! 2. A throwaway `docker run` Postgres container, if docker is available. +//! 3. Otherwise: print a message and return — a visible skip (matching +//! `git-effect`'s own `docker_backend_runs_a_trivial_effect` gate), not a +//! silently-green test. + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + reason = "test harness and assertions, not application code" +)] + +use std::process::{Command, Stdio}; +use std::time::Duration; + +use backend_conformance::WithScratchRepo; +use git_backend::{Expected, RefEdit, RefName, RefStore as _}; +use refstore_postgres::PostgresRefStore; + +/// A reachable test Postgres: either an externally supplied instance or a +/// throwaway docker container this harness starts and stops. +enum TestPostgres { + External(String), + Docker { container_id: String, url: String }, +} + +impl TestPostgres { + fn url(&self) -> &str { + match self { + Self::External(url) | Self::Docker { url, .. } => url, + } + } +} + +impl Drop for TestPostgres { + fn drop(&mut self) { + if let Self::Docker { container_id, .. } = self { + let _ignored = Command::new("docker") + .args(["rm", "-f", container_id]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } + } +} + +/// Obtain a test Postgres per the priority order in the module doc, or +/// `None` if neither an external URL nor docker is available. +fn test_postgres() -> Option<TestPostgres> { + if let Ok(url) = std::env::var("GIT_ENTS_TEST_POSTGRES_URL") { + return Some(TestPostgres::External(url)); + } + if !docker_available() { + return None; + } + start_docker_postgres() +} + +fn docker_available() -> bool { + Command::new("docker") + .arg("version") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|status| status.success()) + .unwrap_or(false) +} + +fn start_docker_postgres() -> Option<TestPostgres> { + let output = Command::new("docker") + .args([ + "run", + "-d", + "--rm", + "-e", + "POSTGRES_PASSWORD=postgres", + "-p", + "127.0.0.1::5432", + "postgres:16-alpine", + ]) + .output() + .ok()?; + if !output.status.success() { + eprintln!( + "refstore-postgres tests: docker run failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + return None; + } + let container_id = String::from_utf8_lossy(&output.stdout).trim().to_owned(); + + for _ in 0..120 { + let ready = Command::new("docker") + .args(["exec", &container_id, "pg_isready", "-U", "postgres"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|status| status.success()) + .unwrap_or(false); + if ready { + break; + } + std::thread::sleep(Duration::from_millis(250)); + } + + let port_output = Command::new("docker") + .args(["port", &container_id, "5432"]) + .output() + .ok()?; + let mapping = String::from_utf8_lossy(&port_output.stdout); + let port = mapping + .lines() + .next()? + .rsplit(':') + .next()? + .trim() + .to_owned(); + + Some(TestPostgres::Docker { + url: format!("host=127.0.0.1 port={port} user=postgres password=postgres dbname=postgres"), + container_id, + }) +} + +/// Get a [`TestPostgres`], or print a skip message and return from the +/// calling test — a visible skip rather than a silently-green pass. +macro_rules! require_postgres { + ($name:literal) => { + match test_postgres() { + Some(pg) => pg, + None => { + eprintln!(concat!( + "skipping ", + $name, + ": set GIT_ENTS_TEST_POSTGRES_URL, or make docker available" + )); + return; + } + } + }; +} + +#[test] +fn conforms_to_ref_store_properties() { + let pg = require_postgres!("conforms_to_ref_store_properties"); + let url = pg.url().to_owned(); + backend_conformance::ref_store_properties(move || { + let repo_id = format!("conformance-{}", uuid::Uuid::new_v4()); + let url = url.clone(); + WithScratchRepo::new(move |_path| PostgresRefStore::connect(&url, repo_id)) + }); +} + +#[test] +fn notify_hint_fires_on_a_transaction_commit() { + let pg = require_postgres!("notify_hint_fires_on_a_transaction_commit"); + let repo_id = format!("notify-{}", uuid::Uuid::new_v4()); + let store = PostgresRefStore::connect(pg.url(), repo_id).expect("connect"); + + let watcher = store.watch(&RefName::new("refs/")).expect("watch"); + let oid = backend_conformance::distinct_oids(1) + .into_iter() + .next() + .expect("one oid"); + store + .transaction(&[RefEdit { + name: RefName::new("refs/heads/watched"), + expected: Expected::MustNotExist, + new: Some(oid), + }]) + .expect("transaction"); + + assert!( + watcher.recv_timeout(Duration::from_secs(10)).is_some(), + "expected a NOTIFY-driven wakeup hint after a committed transaction" + ); +} + +#[test] +fn queue_table_survives_a_dropped_connection() { + let pg = require_postgres!("queue_table_survives_a_dropped_connection"); + let repo_id = format!("queue-{}", uuid::Uuid::new_v4()); + + let id = { + let store = PostgresRefStore::connect(pg.url(), repo_id.clone()).expect("connect"); + store.enqueue_effect("payload-a").expect("enqueue") + }; + // `store` (and its one connection) is dropped here; the row must + // survive in Postgres regardless, per the queue table's at-least-once + // contract (`docs/scale-out.adoc`, "RefStore"). + + let store = PostgresRefStore::connect(pg.url(), repo_id).expect("reconnect"); + let claimed = store.claim_effects("worker-1", 10).expect("claim"); + assert_eq!(claimed.len(), 1); + let claimed_one = claimed.first().expect("one claimed row"); + assert_eq!(claimed_one.id, id); + assert_eq!(claimed_one.payload, "payload-a"); + store.complete_effect(id).expect("complete"); +}

feat: add refstore-postgres, the Postgres RefStore backend

Reviews

Start a review