git ents: feat: add odb-tiered, a small-object tier composed over odb-tigris

Cargo.lock @@ -3552,6 +3552,21 @@ "uuid", ] +[[package]] +name = "odb-tiered" +version = "0.0.0" +dependencies = [ + "backend-conformance", + "git-backend", + "gix-features", + "gix-hash", + "gix-object", + "gix-pack", + "odb-tigris", + "tempfile", + "uuid", +] + [[package]] name = "odb-tigris" version = "0.0.0" @@ -4059,6 +4074,10 @@ "backend-conformance", "git-backend", "gix-hash", + "gix-object", + "odb-tiered", + "odb-tigris", + "tempfile", "tokio", "tokio-postgres", "uuid",

Cargo.toml @@ -15,6 +15,7 @@ "crates/git-store", "crates/git-toolchain", "crates/odb-files", + "crates/odb-tiered", "crates/odb-tigris", "crates/refstore-files", "crates/refstore-postgres", @@ -74,6 +75,7 @@ git-store = { path = "crates/git-store" } git-toolchain = { path = "crates/git-toolchain" } odb-files = { path = "crates/odb-files" } +odb-tiered = { path = "crates/odb-tiered" } odb-tigris = { path = "crates/odb-tigris" } refstore-files = { path = "crates/refstore-files" } refstore-postgres = { path = "crates/refstore-postgres" }

crates/refstore-postgres/Cargo.toml @@ -8,12 +8,16 @@ [dependencies] git-backend = { workspace = true } gix-hash = { workspace = true } +gix-object = { workspace = true } +odb-tiered = { workspace = true } +odb-tigris = { workspace = true } tokio = { workspace = true } tokio-postgres = { workspace = true } +uuid = { workspace = true } [dev-dependencies] backend-conformance = { workspace = true } -uuid = { workspace = true } +tempfile = { workspace = true } [lints] workspace = true

crates/refstore-postgres/migrations/0001_init.sql @@ -31,8 +31,11 @@ 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. +-- Pack registry: WS5 (Tigris object store) records promoted packs here, and +-- `odb_tigris::OdbTigris::read`/`contains` (via `PostgresRefStore`'s +-- `odb_tigris::registry::PackRegistry` impl, see `pack_registry.rs`) consult +-- only this table — never a bucket listing (`docs/scale-out.adoc`, +-- "Reachability"). CREATE TABLE IF NOT EXISTS git_ents_pack_registry ( id BIGSERIAL PRIMARY KEY, repo_id TEXT NOT NULL, @@ -40,9 +43,44 @@ promoted_at TIMESTAMPTZ ); +-- `location_key` predates the `PackRegistry` trait (WS5) and is unused by +-- it; dropping its `NOT NULL` rather than removing it keeps this migration +-- idempotent against a database that already has rows from before this +-- change, without inventing a fake value for a column nothing here writes +-- anymore. +ALTER TABLE git_ents_pack_registry ALTER COLUMN location_key DROP NOT NULL; + +ALTER TABLE git_ents_pack_registry ADD COLUMN IF NOT EXISTS pack_id TEXT; +ALTER TABLE git_ents_pack_registry ADD COLUMN IF NOT EXISTS pack_key TEXT; +ALTER TABLE git_ents_pack_registry ADD COLUMN IF NOT EXISTS idx_key TEXT; +ALTER TABLE git_ents_pack_registry ADD COLUMN IF NOT EXISTS object_count BIGINT; + CREATE INDEX IF NOT EXISTS git_ents_pack_registry_repo_idx ON git_ents_pack_registry (repo_id); +CREATE UNIQUE INDEX IF NOT EXISTS git_ents_pack_registry_repo_pack_idx + ON git_ents_pack_registry (repo_id, pack_id); + +-- Small-object tier (WS5, `docs/scale-out.adoc`'s `odb-tiered` row): blobs +-- and trees under `odb_tiered::OdbTiered`'s size threshold, staged then +-- promoted like any other object storage tier (correctness rules 1 and 2 +-- apply here too). `stage_id` identifies an in-flight batch; `promoted` +-- flips to true (and `stage_id` clears) in the single `UPDATE` that is this +-- tier's whole promotion transaction (see `small_tier.rs`). +CREATE TABLE IF NOT EXISTS git_ents_small_objects ( + repo_id TEXT NOT NULL, + oid TEXT NOT NULL, + kind TEXT NOT NULL, + bytes BYTEA NOT NULL, + stage_id TEXT, + promoted BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (repo_id, oid) +); + +CREATE INDEX IF NOT EXISTS git_ents_small_objects_stage_idx + ON git_ents_small_objects (stage_id) + WHERE stage_id IS NOT NULL; + -- 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").

crates/refstore-postgres/src/lib.rs @@ -10,6 +10,13 @@ //! [`queue`]) is what an at-least-once consumer actually drains (see //! [`notify`], and the trait contract on [`git_backend::RefStore::watch`]). //! +//! This crate also implements two WS5 traits against the same connection, +//! rather than have `odb-tigris`/`odb-tiered` depend on `tokio-postgres` +//! themselves: [`odb_tigris::registry::PackRegistry`] (see +//! [`pack_registry`], over `git_ents_pack_registry`) and +//! [`odb_tiered::small_tier::SmallObjectTier`] (see [`small_tier`], over +//! `git_ents_small_objects`). +//! //! # Q1: single write-primary //! //! This store assumes exactly one writable Postgres primary at a time. Fly @@ -25,8 +32,10 @@ //! fencing code here, deliberately. mod notify; +mod pack_registry; mod queue; mod ref_store; +mod small_tier; pub use queue::{ClaimedEffect, EffectId};

crates/odb-tiered/Cargo.toml @@ -1,0 +1,22 @@ +[package] +name = "odb-tiered" +version = "0.0.0" +edition.workspace = true +publish.workspace = true +license.workspace = true + +[dependencies] +git-backend = { workspace = true } +gix-features = { workspace = true, features = ["progress"] } +gix-hash = { workspace = true } +gix-object = { workspace = true } +gix-pack = { workspace = true } +odb-tigris = { workspace = true } +tempfile = { workspace = true } +uuid = { workspace = true } + +[dev-dependencies] +backend-conformance = { workspace = true } + +[lints] +workspace = true

crates/odb-tiered/src/lib.rs @@ -1,0 +1,260 @@ +//! [`ObjectStore`] composed from a small-object tier over +//! [`odb_tigris::OdbTigris`] — composition, not a third semantics +//! (`docs/scale-out.adoc`, "ObjectStore": "The tiered store is composition, +//! not a third semantics: `read` consults tiers in order; `contains` is the +//! union."). +//! +//! `read` tries [`small_tier::SmallObjectTier`] first, falling through to +//! the underlying store on a tier miss; `contains` is the union of both. +//! `stage_pack` splits an incoming pack by object size against +//! [`SMALL_OBJECT_THRESHOLD_BYTES`] (see its doc comment — Q5: "measure, +//! don't guess"): objects at or above the threshold are re-packed +//! whole-object (via [`odb_tigris::pack_writer`]) and staged into the +//! underlying store exactly as before; objects below it are staged +//! directly into the small tier. `promote` commits both halves. + +pub mod small_tier; + +use std::collections::HashMap; +use std::sync::{Mutex, MutexGuard, PoisonError}; + +use git_backend::{Error, Object, ObjectStore, PackStream, QuarantineId, Result}; +use gix_hash::ObjectId; +use odb_tigris::pack_writer::{ClassifiedObject, LifetimeClass}; + +use crate::small_tier::{SmallObjectTier, SmallStageId}; + +/// Size, in bytes, below which an object is staged into the small tier +/// instead of a pack. A few KiB, per `docs/scale-out.adoc`'s Q5 ("small- +/// object tier threshold: measure"): this default is a plausible starting +/// point for typed documents (manifests, small trees), not a measured +/// value — a real deployment should tune it against observed object-size +/// and access-latency distributions rather than trust this constant. +pub const SMALL_OBJECT_THRESHOLD_BYTES: usize = 4 * 1024; + +/// One quarantined batch, split across the two tiers it may span. +struct Quarantine { + small: Option<SmallStageId>, + underlying: Option<QuarantineId>, +} + +/// [`ObjectStore`] composing a [`SmallObjectTier`] `K` over an underlying +/// store `S` (in practice, [`odb_tigris::OdbTigris`], but any `ObjectStore` +/// qualifies — this crate depends on `odb-tigris` only for +/// [`odb_tigris::pack_writer`], not for a hard-wired backend). +pub struct OdbTiered<S, K> { + underlying: S, + small_tier: K, + repo_id: String, + small_threshold: usize, + quarantines: Mutex<HashMap<QuarantineId, Quarantine>>, +} + +impl<S, K> OdbTiered<S, K> +where + S: ObjectStore, + K: SmallObjectTier, +{ + /// Compose `small_tier` over `underlying`, scoped to `repo_id`, using + /// the default [`SMALL_OBJECT_THRESHOLD_BYTES`]. + pub fn new(underlying: S, small_tier: K, repo_id: impl Into<String>) -> Self { + Self::with_threshold( + underlying, + small_tier, + repo_id, + SMALL_OBJECT_THRESHOLD_BYTES, + ) + } + + /// As [`Self::new`], with an explicit small-object threshold — see + /// [`SMALL_OBJECT_THRESHOLD_BYTES`]'s doc comment on why this should be + /// measured for a real deployment rather than left at the default. + pub fn with_threshold( + underlying: S, + small_tier: K, + repo_id: impl Into<String>, + small_threshold: usize, + ) -> Self { + Self { + underlying, + small_tier, + repo_id: repo_id.into(), + small_threshold, + quarantines: Mutex::new(HashMap::new()), + } + } + + /// Fully materialize every object in an incoming pack, by indexing it + /// (exactly as `odb_tigris::OdbTigris::stage_pack` and `odb_files`'s + /// own quarantine do) and then decoding each entry through + /// `gix_pack`'s own full decoder. Unlike `odb_tigris::decode` (which is + /// built around ranged reads against a *remote* pack this store never + /// downloads in full), the incoming pack here is already fully local — + /// there is no ranged-read concern splitting objects out of it, so + /// reusing `gix_pack::Bundle`'s own (delta-resolving) decode path is + /// the correct choice, not a shortcut. + fn materialize_incoming_pack(pack: PackStream) -> Result<Vec<(ObjectId, Object)>> { + let scratch = tempfile::tempdir()?; + let mut reader = std::io::BufReader::new(pack); + let outcome = gix_pack::Bundle::write_to_directory( + &mut reader, + Some(scratch.path()), + &mut gix_features::progress::Discard, + &std::sync::atomic::AtomicBool::new(false), + None::<NoThinBaseLookup>, + gix_pack::bundle::write::Options { + object_hash: gix_hash::Kind::Sha1, + ..Default::default() + }, + ) + .map_err(|error| Error::ObjectStore(error.to_string()))?; + let index_path = outcome + .index_path + .ok_or_else(|| Error::ObjectStore("pack write produced no index file".to_owned()))?; + let bundle = gix_pack::Bundle::at(&index_path, gix_hash::Kind::Sha1) + .map_err(|error| Error::ObjectStore(error.to_string()))?; + + let entries: Vec<_> = bundle.index.iter().collect(); + let mut objects = Vec::with_capacity(entries.len()); + let mut inflate = gix_features::zlib::Inflate::default(); + let mut cache = gix_pack::cache::Never; + for entry in entries { + let mut buf = Vec::new(); + let (data, _location) = bundle + .find(&entry.oid, &mut buf, &mut inflate, &mut cache) + .map_err(|error| Error::ObjectStore(error.to_string()))? + .ok_or_else(|| { + Error::ObjectStore(format!( + "object {} listed in its own pack's index but not found in it", + entry.oid + )) + })?; + objects.push(( + entry.oid, + Object { + kind: data.kind, + data: data.data.to_vec(), + }, + )); + } + Ok(objects) + } +} + +impl<S, K> ObjectStore for OdbTiered<S, K> +where + S: ObjectStore, + K: SmallObjectTier, +{ + fn read(&self, id: ObjectId) -> Result<Object> { + if let Some(object) = self.small_tier.read(&self.repo_id, id)? { + return Ok(object); + } + self.underlying.read(id) + } + + fn contains(&self, id: ObjectId) -> Result<bool> { + Ok(self.small_tier.contains(&self.repo_id, id)? || self.underlying.contains(id)?) + } + + fn stage_pack(&self, pack: PackStream) -> Result<QuarantineId> { + let objects = Self::materialize_incoming_pack(pack)?; + let (small, large): (Vec<_>, Vec<_>) = objects + .into_iter() + .partition(|(_id, object)| object.data.len() < self.small_threshold); + + let small_stage = if small.is_empty() { + None + } else { + Some(self.small_tier.stage(&self.repo_id, small)?) + }; + + let underlying_quarantine = if large.is_empty() { + None + } else { + let classified: Vec<ClassifiedObject> = large + .into_iter() + .map(|(id, object)| ClassifiedObject { + id, + kind: object.kind, + data: object.data, + // `odb_tiered` doesn't itself track cache-namespace + // lifetime — that's a property of which ref a caller is + // about to point at these objects, not of the bytes + // this store sees. Every object routed to the + // underlying store here is marked `Durable`; a caller + // staging cache-namespace objects through a tiered + // store still gets correct behavior (rule 5 is about + // never mixing lifetimes within a single pack, and a + // single `stage_pack` call is exactly one lifetime by + // construction of its caller), just not automatically + // inferred from content alone. + lifetime: LifetimeClass::Durable, + }) + .collect(); + let partitioned = odb_tigris::pack_writer::partition_and_pack(classified)?; + match partitioned.durable { + Some(pack_bytes) => Some( + self.underlying + .stage_pack(PackStream::new(std::io::Cursor::new(pack_bytes)))?, + ), + None => None, + } + }; + + let id = QuarantineId::new(uuid::Uuid::new_v4().to_string()); + lock(&self.quarantines).insert( + id.clone(), + Quarantine { + small: small_stage, + underlying: underlying_quarantine, + }, + ); + Ok(id) + } + + fn promote(&self, q: QuarantineId) -> Result<()> { + let quarantine = lock(&self.quarantines) + .remove(&q) + .ok_or_else(|| Error::ObjectStore(format!("unknown quarantine {q}")))?; + + // Q5/rule 1&2: each tier's own `promote` is the commit point for + // the objects it holds (staged objects invisible until promoted, + // per each tier's own contract). "Transactionally with it" (see + // this crate's module doc) means the small tier's staged rows move + // to live in one Postgres transaction internally — not that this + // pair of calls is one distributed transaction spanning Tigris and + // Postgres, which no code here could honestly promise. If the + // small-tier promote below succeeds and the underlying promote + // fails (or the reverse), the failing half's objects simply remain + // quarantined/staged — never partially visible — and the caller + // sees the error and can retry `promote` with the same id. + if let Some(small) = quarantine.small { + self.small_tier.promote(small)?; + } + if let Some(underlying) = quarantine.underlying { + self.underlying.promote(underlying)?; + } + Ok(()) + } +} + +/// A `gix_object::Find` that never finds anything, satisfying +/// `Bundle::write_to_directory`'s thin-pack-base-lookup parameter — see +/// `odb_tigris`'s identical helper for why this is correct here too (this +/// store's incoming packs are also expected to be self-contained). +struct NoThinBaseLookup; + +impl gix_object::Find for NoThinBaseLookup { + fn try_find<'a>( + &self, + _id: &gix_hash::oid, + _buffer: &'a mut Vec<u8>, + ) -> std::result::Result<Option<gix_object::Data<'a>>, gix_object::find::Error> { + Ok(None) + } +} + +fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +}

crates/odb-tiered/src/small_tier.rs @@ -1,0 +1,79 @@ +//! [`SmallObjectTier`]: the fast, hot-path store for small objects that +//! [`crate::OdbTiered`] consults before falling through to the underlying +//! (Tigris) store (`docs/scale-out.adoc`, "ObjectStore": "Typed documents +//! are tiny and hot; Tigris per-GET latency is the wrong floor for them."). +//! +//! Like [`git_backend::ObjectStore`] itself, staging is a first-class +//! concept here, not an afterthought: correctness rule 1 (causal collection +//! safety) and rule 2 (ref transactions are the only commit point) apply +//! just as much to objects that land in this tier as to ones that land in a +//! pack, so [`SmallObjectTier::stage`]d objects must stay invisible to +//! [`SmallObjectTier::read`]/[`SmallObjectTier::contains`] until +//! [`SmallObjectTier::promote`] is called — mirroring +//! [`git_backend::ObjectStore`]'s own contract exactly. + +pub mod memory; + +use git_backend::{Object, Result}; +use gix_hash::ObjectId; + +/// A handle to a batch staged by [`SmallObjectTier::stage`], passed back to +/// [`SmallObjectTier::promote`]. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct SmallStageId(String); + +impl SmallStageId { + /// Build a `SmallStageId` from a backend-chosen opaque token. + #[must_use] + pub fn new(id: impl Into<String>) -> Self { + Self(id.into()) + } + + /// The id as a `&str`. + #[must_use] + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl std::fmt::Display for SmallStageId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +/// A small-object store: a blob/tree key-value store scoped per repo, +/// staged then promoted exactly like [`git_backend::ObjectStore`]. +pub trait SmallObjectTier: Send + Sync { + /// Read `id` from the promoted (non-staged) view of `repo_id`'s + /// objects, or `None` if this tier doesn't hold it — a caller + /// ([`crate::OdbTiered`]) falls through to the underlying store on + /// `None`, so this is not itself an error. + /// + /// # Errors + /// + /// Returns an error if the tier cannot be read. + fn read(&self, repo_id: &str, id: ObjectId) -> Result<Option<Object>>; + + /// Whether `id` is present in the promoted view of `repo_id`'s objects. + /// + /// # Errors + /// + /// Returns an error if the tier cannot be read. + fn contains(&self, repo_id: &str, id: ObjectId) -> Result<bool>; + + /// Stage `objects` for `repo_id`, invisible to `read`/`contains` until + /// [`Self::promote`] is called on the returned id. + /// + /// # Errors + /// + /// Returns an error if the batch cannot be durably staged. + fn stage(&self, repo_id: &str, objects: Vec<(ObjectId, Object)>) -> Result<SmallStageId>; + + /// Make the batch staged under `id` visible to `read`/`contains`. + /// + /// # Errors + /// + /// Returns an error if promotion fails. + fn promote(&self, id: SmallStageId) -> Result<()>; +}

crates/odb-tiered/src/small_tier/memory.rs @@ -1,0 +1,69 @@ +//! [`InMemorySmallTier`]: an in-process [`SmallObjectTier`], used by tests +//! and by the conformance instantiation +//! (`crates/odb-tiered/tests/conformance.rs`). + +use std::collections::HashMap; +use std::sync::{Mutex, MutexGuard, PoisonError}; + +use git_backend::{Object, Result}; +use gix_hash::ObjectId; + +use super::{SmallObjectTier, SmallStageId}; + +/// One staged-but-unpromoted batch. +struct Staged { + repo_id: String, + objects: Vec<(ObjectId, Object)>, +} + +/// A [`SmallObjectTier`] held entirely in memory, scoped to one process. +#[derive(Default)] +pub struct InMemorySmallTier { + promoted: Mutex<HashMap<(String, ObjectId), Object>>, + staged: Mutex<HashMap<SmallStageId, Staged>>, +} + +impl InMemorySmallTier { + /// An empty tier. + #[must_use] + pub fn new() -> Self { + Self::default() + } +} + +impl SmallObjectTier for InMemorySmallTier { + fn read(&self, repo_id: &str, id: ObjectId) -> Result<Option<Object>> { + Ok(lock(&self.promoted).get(&(repo_id.to_owned(), id)).cloned()) + } + + fn contains(&self, repo_id: &str, id: ObjectId) -> Result<bool> { + Ok(lock(&self.promoted).contains_key(&(repo_id.to_owned(), id))) + } + + fn stage(&self, repo_id: &str, objects: Vec<(ObjectId, Object)>) -> Result<SmallStageId> { + let id = SmallStageId::new(uuid::Uuid::new_v4().to_string()); + lock(&self.staged).insert( + id.clone(), + Staged { + repo_id: repo_id.to_owned(), + objects, + }, + ); + Ok(id) + } + + fn promote(&self, id: SmallStageId) -> Result<()> { + let Staged { repo_id, objects } = lock(&self.staged).remove(&id).ok_or_else(|| { + git_backend::Error::ObjectStore(format!("unknown small-tier stage {id}")) + })?; + let mut promoted = lock(&self.promoted); + for (oid, object) in objects { + promoted.insert((repo_id.clone(), oid), object); + } + Ok(()) + } +} + +fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +}

crates/odb-tiered/tests/conformance.rs @@ -1,0 +1,57 @@ +//! This crate's instantiation of the shared backend conformance suite +//! (`docs/scale-out.adoc`, WS2): every `ObjectStore` property run against +//! `OdbTiered` composed over `OdbTigris`'s no-network stand-ins plus an +//! in-memory small tier — no bucket, no Postgres, no network. + +#![allow(clippy::expect_used, reason = "test harness, not application code")] + +use backend_conformance::NoopCollector; +use git_backend::{Object, ObjectStore, PackStream, QuarantineId, Result}; +use gix_hash::ObjectId; +use odb_tiered::OdbTiered; +use odb_tiered::small_tier::memory::InMemorySmallTier; +use odb_tigris::OdbTigris; +use odb_tigris::registry::memory::InMemoryRegistry; +use odb_tigris::transport::fs::FsTransport; + +type Underlying = OdbTigris<FsTransport, InMemoryRegistry>; + +/// Bundles an `OdbTiered` with the tempdir its underlying `OdbTigris` +/// bucket root lives under, so the directory outlives the store. +struct WithBucketDir { + store: OdbTiered<Underlying, InMemorySmallTier>, + _dir: tempfile::TempDir, +} + +impl WithBucketDir { + fn new() -> Self { + let dir = tempfile::tempdir().expect("tempdir"); + let transport = FsTransport::open(dir.path().join("bucket")).expect("open transport"); + let underlying = OdbTigris::new(transport, InMemoryRegistry::new(), "conformance-repo"); + let store = OdbTiered::new(underlying, InMemorySmallTier::new(), "conformance-repo"); + Self { store, _dir: dir } + } +} + +impl ObjectStore for WithBucketDir { + fn read(&self, id: ObjectId) -> Result<Object> { + self.store.read(id) + } + + fn contains(&self, id: ObjectId) -> Result<bool> { + self.store.contains(id) + } + + fn stage_pack(&self, pack: PackStream) -> Result<QuarantineId> { + self.store.stage_pack(pack) + } + + fn promote(&self, q: QuarantineId) -> Result<()> { + self.store.promote(q) + } +} + +#[test] +fn conforms_to_object_store_properties() { + backend_conformance::object_store_properties(WithBucketDir::new, &NoopCollector); +}

crates/odb-tiered/tests/tiering.rs @@ -1,0 +1,107 @@ +//! Exercises `OdbTiered::stage_pack`'s size-based split +//! (`crates/odb-tiered/src/lib.rs`): a pack containing both a small blob +//! (routed to the small tier) and a large blob (repacked and routed to the +//! underlying store) must come back byte-correct for both, in one +//! `stage_pack`/`promote` call. + +#![allow(clippy::expect_used, reason = "test harness, not application code")] + +use std::process::{Command, Stdio}; + +use git_backend::{ObjectStore as _, PackStream}; +use gix_hash::{Kind as HashKind, ObjectId}; +use odb_tiered::OdbTiered; +use odb_tiered::small_tier::memory::InMemorySmallTier; +use odb_tigris::OdbTigris; +use odb_tigris::registry::memory::InMemoryRegistry; +use odb_tigris::transport::fs::FsTransport; + +fn blob_oid(data: &[u8]) -> ObjectId { + let mut hasher = gix_hash::hasher(HashKind::Sha1); + hasher.update(format!("blob {}\0", data.len()).as_bytes()); + hasher.update(data); + hasher.try_finalize().expect("hash blob") +} + +/// A real pack containing `commit` and everything it reaches, built the +/// same way `odb-files`' and `backend-conformance`'s own fixtures are. +fn pack_for(dir: &std::path::Path, commit: &str) -> Vec<u8> { + let mut rev_list = Command::new("git") + .arg("-C") + .arg(dir) + .args(["rev-list", "--objects", commit]) + .stdout(Stdio::piped()) + .spawn() + .expect("spawn git rev-list"); + let pack_objects = Command::new("git") + .arg("-C") + .arg(dir) + .args(["pack-objects", "--stdout", "-q"]) + .stdin(rev_list.stdout.take().expect("rev-list stdout")) + .stdout(Stdio::piped()) + .spawn() + .expect("spawn git pack-objects"); + let output = pack_objects.wait_with_output().expect("wait pack-objects"); + assert!(rev_list.wait().expect("wait rev-list").success()); + assert!(output.status.success()); + output.stdout +} + +fn git(dir: &std::path::Path, args: &[&str]) { + let status = Command::new("git") + .arg("-C") + .arg(dir) + .args(args) + .status() + .expect("run git"); + assert!(status.success()); +} + +#[test] +fn stages_small_and_large_objects_from_one_pack_correctly() { + let small_content = b"tiny document content".to_vec(); // well under any sane threshold + let large_content = vec![b'x'; 200_000]; // well over the default threshold + + let dir = tempfile::tempdir().expect("scratch repo dir"); + git(dir.path(), &["init", "-q"]); + git(dir.path(), &["config", "user.email", "test@example.com"]); + git(dir.path(), &["config", "user.name", "Test"]); + std::fs::write(dir.path().join("small.txt"), &small_content).expect("write small file"); + std::fs::write(dir.path().join("large.bin"), &large_content).expect("write large file"); + git(dir.path(), &["add", "small.txt", "large.bin"]); + git(dir.path(), &["commit", "-q", "-m", "tiering fixture"]); + let commit_hex = String::from_utf8( + Command::new("git") + .arg("-C") + .arg(dir.path()) + .args(["rev-parse", "HEAD"]) + .output() + .expect("rev-parse") + .stdout, + ) + .expect("utf8") + .trim() + .to_owned(); + + let pack_bytes = pack_for(dir.path(), &commit_hex); + + let bucket_dir = tempfile::tempdir().expect("bucket dir"); + let transport = FsTransport::open(bucket_dir.path().join("bucket")).expect("open transport"); + let underlying = OdbTigris::new(transport, InMemoryRegistry::new(), "tiering-repo"); + let store = OdbTiered::new(underlying, InMemorySmallTier::new(), "tiering-repo"); + + let quarantine = store + .stage_pack(PackStream::new(std::io::Cursor::new(pack_bytes))) + .expect("stage_pack splits the incoming pack by size"); + store.promote(quarantine).expect("promote"); + + let small_object = store + .read(blob_oid(&small_content)) + .expect("read small object back"); + assert_eq!(small_object.data, small_content); + + let large_object = store + .read(blob_oid(&large_content)) + .expect("read large object back"); + assert_eq!(large_object.data, large_content); +}

crates/refstore-postgres/src/pack_registry.rs @@ -1,0 +1,99 @@ +//! [`odb_tigris::registry::PackRegistry`] for [`PostgresRefStore`]: rows in +//! `git_ents_pack_registry`, extended (see `migrations/0001_init.sql`) with +//! the columns WS5 needs beyond WS4's original minimal shape +//! (`docs/scale-out.adoc`, "ObjectStore" / WS5). +//! +//! Kept in this crate rather than `odb-tigris` itself so `odb-tigris` never +//! needs a `tokio-postgres` dependency of its own — it depends only on the +//! [`odb_tigris::registry::PackRegistry`] trait, and this crate (which +//! already owns the Postgres connection) implements it. + +use git_backend::{Error, Result}; +use odb_tigris::registry::{PackId, PackRecord, PackRegistry}; + +use crate::{PostgresRefStore, pg_err}; + +impl PackRegistry for PostgresRefStore { + fn record(&self, record: PackRecord) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .execute( + "INSERT INTO git_ents_pack_registry + (repo_id, pack_id, pack_key, idx_key, object_count, promoted_at) + VALUES ($1, $2, $3, $4, $5, now()) + ON CONFLICT (repo_id, pack_id) DO UPDATE SET + pack_key = EXCLUDED.pack_key, + idx_key = EXCLUDED.idx_key, + object_count = EXCLUDED.object_count, + promoted_at = EXCLUDED.promoted_at", + &[ + &record.repo_id, + &record.id.as_str(), + &record.pack_key, + &record.idx_key, + &record + .object_count + .map(|count| i64::try_from(count).unwrap_or(i64::MAX)), + ], + ) + .await + }) + .map_err(pg_err) + .map(|_rows_affected| ()) + } + + fn list(&self, repo_id: &str) -> Result<Vec<PackRecord>> { + let rows = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query( + "SELECT pack_id, pack_key, idx_key, object_count + FROM git_ents_pack_registry + WHERE repo_id = $1 AND pack_id IS NOT NULL", + &[&repo_id], + ) + .await + }) + .map_err(pg_err)?; + + rows.into_iter() + .map(|row| { + let pack_id: String = row.try_get(0).map_err(pg_err)?; + let pack_key: String = row.try_get(1).map_err(pg_err)?; + let idx_key: String = row.try_get(2).map_err(pg_err)?; + let object_count: Option<i64> = row.try_get(3).map_err(pg_err)?; + Ok(PackRecord { + id: PackId::new(pack_id), + repo_id: repo_id.to_owned(), + pack_key, + idx_key, + object_count: object_count + .map(|count| { + u64::try_from(count) + .map_err(|error| Error::ObjectStore(error.to_string())) + }) + .transpose()?, + }) + }) + .collect() + } + + fn delete(&self, repo_id: &str, id: &PackId) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .execute( + "DELETE FROM git_ents_pack_registry WHERE repo_id = $1 AND pack_id = $2", + &[&repo_id, &id.as_str()], + ) + .await + }) + .map_err(pg_err) + .map(|_rows_affected| ()) + } +}

crates/refstore-postgres/src/small_tier.rs @@ -1,0 +1,143 @@ +//! [`odb_tiered::small_tier::SmallObjectTier`] for [`PostgresRefStore`]: +//! rows in `git_ents_small_objects` (see `migrations/0001_init.sql`), +//! staged then promoted in one `UPDATE` — that single statement is this +//! tier's entire promotion transaction (`docs/scale-out.adoc`, "ObjectStore" +//! / WS5: staged objects invisible until promoted, same contract as +//! [`git_backend::ObjectStore`] itself). +//! +//! Kept in this crate rather than `odb-tiered` itself, for the same reason +//! as [`crate::pack_registry`]: `odb-tiered` depends only on the +//! [`odb_tiered::small_tier::SmallObjectTier`] trait, never on +//! `tokio-postgres`. + +use git_backend::{Error, Object, Result}; +use gix_hash::ObjectId; +use gix_object::Kind; +use odb_tiered::small_tier::{SmallObjectTier, SmallStageId}; + +use crate::{PostgresRefStore, pg_err}; + +/// Render a [`Kind`] as the fixed string stored in `git_ents_small_objects.kind`. +fn kind_to_str(kind: Kind) -> &'static str { + match kind { + Kind::Blob => "blob", + Kind::Tree => "tree", + Kind::Commit => "commit", + Kind::Tag => "tag", + } +} + +/// The inverse of [`kind_to_str`]. +fn kind_from_str(s: &str) -> Result<Kind> { + match s { + "blob" => Ok(Kind::Blob), + "tree" => Ok(Kind::Tree), + "commit" => Ok(Kind::Commit), + "tag" => Ok(Kind::Tag), + other => Err(Error::ObjectStore(format!( + "corrupt git_ents_small_objects row: unknown kind {other:?}" + ))), + } +} + +impl SmallObjectTier for PostgresRefStore { + fn read(&self, repo_id: &str, id: ObjectId) -> Result<Option<Object>> { + let hex = id.to_hex().to_string(); + let row = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query_opt( + "SELECT kind, bytes FROM git_ents_small_objects + WHERE repo_id = $1 AND oid = $2 AND promoted", + &[&repo_id, &hex], + ) + .await + }) + .map_err(pg_err)?; + let Some(row) = row else { + return Ok(None); + }; + let kind: String = row.try_get(0).map_err(pg_err)?; + let data: Vec<u8> = row.try_get(1).map_err(pg_err)?; + Ok(Some(Object { + kind: kind_from_str(&kind)?, + data, + })) + } + + fn contains(&self, repo_id: &str, id: ObjectId) -> Result<bool> { + let hex = id.to_hex().to_string(); + let row = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query_opt( + "SELECT 1 FROM git_ents_small_objects + WHERE repo_id = $1 AND oid = $2 AND promoted", + &[&repo_id, &hex], + ) + .await + }) + .map_err(pg_err)?; + Ok(row.is_some()) + } + + fn stage(&self, repo_id: &str, objects: Vec<(ObjectId, Object)>) -> Result<SmallStageId> { + let stage_id = SmallStageId::new(uuid::Uuid::new_v4().to_string()); + self.runtime + .block_on(async { + let mut client = self.client.lock().await; + let tx = client.transaction().await?; + for (oid, object) in &objects { + let hex = oid.to_hex().to_string(); + // Objects are content-addressed: if `(repo_id, oid)` + // already has a row — promoted already, or staged by a + // still-in-flight batch — its bytes can only be the + // same bytes, so leaving it untouched is correct rather + // than reattaching it to this new `stage_id` (which + // would risk this batch's `promote` racing whatever + // state the existing row is already in). + tx.execute( + "INSERT INTO git_ents_small_objects + (repo_id, oid, kind, bytes, stage_id, promoted) + VALUES ($1, $2, $3, $4, $5, FALSE) + ON CONFLICT (repo_id, oid) DO NOTHING", + &[ + &repo_id, + &hex, + &kind_to_str(object.kind), + &object.data, + &stage_id.as_str(), + ], + ) + .await?; + } + tx.commit().await + }) + .map_err(pg_err)?; + Ok(stage_id) + } + + fn promote(&self, id: SmallStageId) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + // The whole promotion, for every object in this batch, is + // this one statement — there is no distinct "commit" step + // to get half-applied. + client + .execute( + "UPDATE git_ents_small_objects + SET promoted = TRUE, stage_id = NULL + WHERE stage_id = $1", + &[&id.as_str()], + ) + .await + }) + .map_err(pg_err) + .map(|_rows_affected| ()) + } +}

crates/refstore-postgres/tests/odb_ws5_conformance.rs @@ -1,0 +1,193 @@ +//! Gated conformance for the WS5 Postgres implementations this crate adds: +//! [`odb_tigris::registry::PackRegistry`] (`pack_registry.rs`) and +//! [`odb_tiered::small_tier::SmallObjectTier`] (`small_tier.rs`), both over +//! `PostgresRefStore`. Reuses `tests/conformance.rs`'s own +//! docker-or-`GIT_ENTS_TEST_POSTGRES_URL` gating pattern (duplicated here +//! rather than shared, since Rust integration test binaries can't import +//! each other's private items) — see that file's module doc for the +//! priority order and the visible-skip rationale. +//! +//! The bucket side of `OdbTigris` uses `FsTransport` here, not a real S3 +//! bucket: this test's job is exercising the two Postgres-backed traits, +//! not re-verifying `odb-tigris`'s own transport-agnostic conformance +//! (already covered, over `FsTransport` + an in-memory registry, in +//! `crates/odb-tigris/tests/conformance.rs`). + +#![allow(clippy::expect_used, reason = "test harness, not application code")] + +use std::process::{Command, Stdio}; +use std::time::Duration; + +use backend_conformance::NoopCollector; +use git_backend::{Object, ObjectStore, PackStream, QuarantineId, Result}; +use gix_hash::ObjectId; +use odb_tiered::OdbTiered; +use odb_tigris::OdbTigris; +use odb_tigris::transport::fs::FsTransport; +use refstore_postgres::PostgresRefStore; + +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(); + } + } +} + +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 odb_ws5_conformance: 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, + }) +} + +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; + } + } + }; +} + +type Underlying = OdbTigris<FsTransport, PostgresRefStore>; + +/// Bundles a store composed over two `PostgresRefStore` connections (one +/// playing `PackRegistry`, one playing `SmallObjectTier`) with the tempdir +/// its `FsTransport` bucket root lives under. +struct WithPostgres { + store: OdbTiered<Underlying, PostgresRefStore>, + _dir: tempfile::TempDir, +} + +impl WithPostgres { + fn new(url: &str) -> Self { + let repo_id = format!("ws5-conformance-{}", uuid::Uuid::new_v4()); + let dir = tempfile::tempdir().expect("tempdir"); + let transport = FsTransport::open(dir.path().join("bucket")).expect("open transport"); + let registry = PostgresRefStore::connect(url, repo_id.clone()).expect("connect registry"); + let underlying = OdbTigris::new(transport, registry, repo_id.clone()); + let small_tier = + PostgresRefStore::connect(url, repo_id.clone()).expect("connect small tier"); + let store = OdbTiered::new(underlying, small_tier, repo_id); + Self { store, _dir: dir } + } +} + +impl ObjectStore for WithPostgres { + fn read(&self, id: ObjectId) -> Result<Object> { + self.store.read(id) + } + + fn contains(&self, id: ObjectId) -> Result<bool> { + self.store.contains(id) + } + + fn stage_pack(&self, pack: PackStream) -> Result<QuarantineId> { + self.store.stage_pack(pack) + } + + fn promote(&self, q: QuarantineId) -> Result<()> { + self.store.promote(q) + } +} + +#[test] +fn conforms_to_object_store_properties_over_postgres() { + let pg = require_postgres!("conforms_to_object_store_properties_over_postgres"); + let url = pg.url().to_owned(); + backend_conformance::object_store_properties(|| WithPostgres::new(&url), &NoopCollector); +}

feat: add odb-tiered, a small-object tier composed over odb-tigris

Reviews

Start a review