Add durable manifest-rooted result inventories

Cargo.lock

@@ -2204,6 +2204,23 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "remora-result"
+version = "0.0.0"
+dependencies = [
+ "bytes",
+ "futures-util",
+ "remora-domain",
+ "remora-durable",
+ "remora-stream",
+ "remora-target",
+ "serde",
+ "serde_json",
+ "sqlx",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "remora-stream"
 version = "0.0.0"

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "crates/remora-domain",
     "crates/remora-durable",
     "crates/remora-identity",
+    "crates/remora-result",
     "crates/remora-stream",
     "crates/remora-target",
     "crates/remora-worker",

crates/remora-result/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "remora-result"
+edition.workspace = true
+publish = false
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+bytes = "1.11.1"
+futures-util = "0.3.32"
+remora-domain.workspace = true
+remora-durable.workspace = true
+remora-stream = { path = "../remora-stream" }
+remora-target = { path = "../remora-target" }
+serde.workspace = true
+serde_json.workspace = true
+sqlx.workspace = true
+thiserror.workspace = true
+
+[dev-dependencies]
+remora-target = { path = "../remora-target", features = ["test-utils"] }
+tokio.workspace = true

crates/remora-result/migrations/0001_result_inventory.sql

@@ -0,0 +1,40 @@
+CREATE TABLE result_inventories (
+    attempt_id TEXT PRIMARY KEY,
+    state TEXT NOT NULL DEFAULT 'open' CHECK (state IN ('open', 'closed', 'sealed')),
+    manifest_schema_version INTEGER,
+    manifest_id TEXT,
+    manifest_store TEXT,
+    manifest_bucket TEXT,
+    manifest_key TEXT,
+    manifest_version_id TEXT,
+    manifest_sha256 TEXT,
+    manifest_size BIGINT,
+    member_count BIGINT NOT NULL DEFAULT 0 CHECK (member_count >= 0),
+    total_bytes BIGINT NOT NULL DEFAULT 0 CHECK (total_bytes >= 0),
+    CHECK (
+        (state <> 'sealed' AND manifest_schema_version IS NULL AND manifest_id IS NULL
+            AND manifest_store IS NULL AND manifest_bucket IS NULL AND manifest_key IS NULL
+            AND manifest_version_id IS NULL AND manifest_sha256 IS NULL AND manifest_size IS NULL)
+        OR
+        (state = 'sealed' AND manifest_schema_version IS NOT NULL AND manifest_id IS NOT NULL
+            AND manifest_store IS NOT NULL AND manifest_bucket IS NOT NULL AND manifest_key IS NOT NULL
+            AND manifest_sha256 IS NOT NULL AND manifest_size IS NOT NULL)
+    )
+);
+
+CREATE TABLE result_inventory_members (
+    attempt_id TEXT NOT NULL REFERENCES result_inventories(attempt_id) ON DELETE CASCADE,
+    member_id TEXT NOT NULL,
+    path TEXT NOT NULL,
+    size BIGINT NOT NULL CHECK (size >= 0),
+    object_store TEXT NOT NULL,
+    object_bucket TEXT NOT NULL,
+    object_key TEXT NOT NULL,
+    object_version_id TEXT,
+    object_sha256 TEXT NOT NULL,
+    PRIMARY KEY (attempt_id, member_id),
+    UNIQUE (attempt_id, path)
+);
+
+CREATE INDEX result_inventory_members_order_idx
+    ON result_inventory_members (attempt_id, path, member_id);

crates/remora-result/src/error.rs

@@ -0,0 +1,62 @@
+use remora_durable::DurableError;
+use remora_target::TargetError;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ResultError {
+    #[error("database operation failed")]
+    Database(#[source] sqlx::Error),
+    #[error("durable substrate operation failed")]
+    Durable(#[source] DurableError),
+    #[error("target operation failed")]
+    Target(#[source] TargetError),
+    #[error("manifest streaming write failed")]
+    ManifestWrite(#[source] remora_stream::StreamError),
+    #[error("invalid result member path")]
+    InvalidMemberPath,
+    #[error("invalid exact object identity")]
+    InvalidObjectIdentity,
+    #[error("result inventory already exists")]
+    InventoryExists,
+    #[error("result inventory is missing")]
+    InventoryNotFound,
+    #[error("result inventory is not open")]
+    InventoryNotOpen,
+    #[error("result inventory is empty")]
+    EmptyInventory,
+    #[error("duplicate result member identity")]
+    DuplicateMember,
+    #[error("result member path collides after canonicalization")]
+    PathCollision,
+    #[error("result aggregate integrity validation failed")]
+    Integrity,
+    #[error("result descriptor belongs to a different attempt")]
+    StaleAttempt,
+    #[error("unsupported result manifest schema version {0}")]
+    UnsupportedManifestVersion(u32),
+    #[error("result is not sealed")]
+    NotSealed,
+    #[error("result member is missing")]
+    MemberNotFound,
+    #[error("invalid page size")]
+    InvalidPageSize,
+    #[error("result size exceeds supported durable range")]
+    SizeOverflow,
+}
+
+impl From<sqlx::Error> for ResultError {
+    fn from(value: sqlx::Error) -> Self {
+        Self::Database(value)
+    }
+}
+
+impl From<DurableError> for ResultError {
+    fn from(value: DurableError) -> Self {
+        Self::Durable(value)
+    }
+}
+
+impl From<TargetError> for ResultError {
+    fn from(value: TargetError) -> Self {
+        Self::Target(value)
+    }
+}

crates/remora-result/src/inventory.rs

@@ -0,0 +1,483 @@
+use futures_util::TryStreamExt;
+use remora_domain::{MemberPath, OpaqueId};
+use remora_stream::TargetBody;
+use remora_target::{
+    ContentSha256, CreateObjectRequest, ObjectIdentity, StoredObject, TargetStore,
+    validate_attempt_key,
+};
+use serde::Serialize;
+use sqlx::PgPool;
+
+use crate::ResultError;
+use crate::types::{
+    CommittedResult, MANIFEST_SCHEMA_VERSION, PageSize, PublicationInput, ResultMember,
+    ResultMemberPage,
+};
+
+#[derive(Clone)]
+pub struct ResultInventory {
+    pool: PgPool,
+}
+
+impl ResultInventory {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+
+    pub async fn create(&self, attempt_id: &OpaqueId) -> Result<(), ResultError> {
+        let result = sqlx::query("INSERT INTO result_inventories (attempt_id) VALUES ($1)")
+            .bind(attempt_id.as_str())
+            .execute(&self.pool)
+            .await;
+        match result {
+            Ok(_) => Ok(()),
+            Err(error) if is_unique_violation(&error, "result_inventories_pkey") => {
+                Err(ResultError::InventoryExists)
+            }
+            Err(error) => Err(error.into()),
+        }
+    }
+
+    pub async fn add_member(
+        &self,
+        attempt_id: &OpaqueId,
+        member: &ResultMember,
+    ) -> Result<(), ResultError> {
+        validate_member_identity(attempt_id, member)?;
+        let size = i64::try_from(member.size()).map_err(|_| ResultError::SizeOverflow)?;
+        let mut tx = self.pool.begin().await?;
+        let state: Option<String> = sqlx::query_scalar(
+            "SELECT state FROM result_inventories WHERE attempt_id = $1 FOR UPDATE",
+        )
+        .bind(attempt_id.as_str())
+        .fetch_optional(tx.as_mut())
+        .await?;
+        match state.as_deref() {
+            None => return Err(ResultError::InventoryNotFound),
+            Some("open") => {}
+            Some(_) => return Err(ResultError::InventoryNotOpen),
+        }
+
+        let insert = sqlx::query(
+            "INSERT INTO result_inventory_members (
+                attempt_id, member_id, path, size, object_store, object_bucket,
+                object_key, object_version_id, object_sha256
+             ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
+        )
+        .bind(attempt_id.as_str())
+        .bind(member.id().as_str())
+        .bind(member.path())
+        .bind(size)
+        .bind(member.object().store())
+        .bind(member.object().bucket())
+        .bind(member.object().key())
+        .bind(member.object().version_id())
+        .bind(member.content_sha256().as_str())
+        .execute(tx.as_mut())
+        .await;
+        if let Err(error) = insert {
+            return match error
+                .as_database_error()
+                .and_then(|error| error.constraint())
+            {
+                Some("result_inventory_members_pkey") => Err(ResultError::DuplicateMember),
+                Some("result_inventory_members_attempt_id_path_key") => {
+                    Err(ResultError::PathCollision)
+                }
+                _ => Err(error.into()),
+            };
+        }
+        sqlx::query(
+            "UPDATE result_inventories
+             SET member_count = member_count + 1, total_bytes = total_bytes + $2
+             WHERE attempt_id = $1",
+        )
+        .bind(attempt_id.as_str())
+        .bind(size)
+        .execute(tx.as_mut())
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn close(&self, attempt_id: &OpaqueId) -> Result<(), ResultError> {
+        let mut tx = self.pool.begin().await?;
+        let state: Option<(String, i64)> = sqlx::query_as(
+            "SELECT state, member_count FROM result_inventories WHERE attempt_id = $1 FOR UPDATE",
+        )
+        .bind(attempt_id.as_str())
+        .fetch_optional(tx.as_mut())
+        .await?;
+        let Some((state, count)) = state else {
+            return Err(ResultError::InventoryNotFound);
+        };
+        if count == 0 {
+            return Err(ResultError::EmptyInventory);
+        }
+        if state == "open" {
+            sqlx::query("UPDATE result_inventories SET state = 'closed' WHERE attempt_id = $1")
+                .bind(attempt_id.as_str())
+                .execute(tx.as_mut())
+                .await?;
+        }
+        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn create_manifest(
+        &self,
+        attempt_id: &OpaqueId,
+        manifest_id: &OpaqueId,
+        target: &TargetStore,
+    ) -> Result<PublicationInput, ResultError> {
+        self.close(attempt_id).await?;
+        if let Some(existing) = self.publication_input(attempt_id).await? {
+            return if existing.manifest_id == *manifest_id {
+                Ok(existing)
+            } else {
+                Err(ResultError::NotSealed)
+            };
+        }
+        self.validate_members(attempt_id, target).await?;
+        let (member_count, total_bytes) = self.inventory_totals(attempt_id).await?;
+        let request = CreateObjectRequest::for_attempt_manifest(attempt_id, manifest_id);
+        let mut body = target.open_create_only(&request).await?;
+        body.write_all(&manifest_header(attempt_id))
+            .await
+            .map_err(ResultError::ManifestWrite)?;
+
+        let mut rows = sqlx::query_as::<_, MemberRow>(MEMBER_SELECT_ORDERED)
+            .bind(attempt_id.as_str())
+            .fetch(&self.pool);
+        let mut first = true;
+        while let Some(row) = rows.try_next().await? {
+            if !first {
+                body.write_all(b",")
+                    .await
+                    .map_err(ResultError::ManifestWrite)?;
+            }
+            first = false;
+            let member = row.into_member()?;
+            let encoded = serde_json::to_vec(&member).map_err(|_| ResultError::Integrity)?;
+            body.write_all(&encoded)
+                .await
+                .map_err(ResultError::ManifestWrite)?;
+        }
+        body.write_all(&manifest_footer(member_count, total_bytes))
+            .await
+            .map_err(ResultError::ManifestWrite)?;
+        let written = body.finish_object().await?;
+        let manifest = StoredObject {
+            identity: written.identity,
+            content_sha256: written.content_sha256,
+            size: written.bytes_written,
+        };
+        let input = PublicationInput {
+            attempt_id: attempt_id.clone(),
+            manifest_id: manifest_id.clone(),
+            schema_version: MANIFEST_SCHEMA_VERSION,
+            manifest,
+            member_count,
+            total_bytes,
+        };
+        self.persist_sealed(&input).await?;
+        Ok(input)
+    }
+
+    pub async fn validate_publication_input(
+        &self,
+        attempt_id: &OpaqueId,
+        input: &PublicationInput,
+    ) -> Result<(), ResultError> {
+        if input.attempt_id != *attempt_id {
+            return Err(ResultError::StaleAttempt);
+        }
+        require_supported_version(input.schema_version)?;
+        let expected_key = remora_target::attempt_manifest_key(attempt_id, &input.manifest_id);
+        if input.manifest.identity.key != expected_key {
+            return Err(ResultError::InvalidObjectIdentity);
+        }
+        match self.publication_input(attempt_id).await? {
+            Some(stored) if stored == *input => Ok(()),
+            Some(_) => Err(ResultError::Integrity),
+            None => Err(ResultError::NotSealed),
+        }
+    }
+
+    pub async fn resolve_committed(
+        &self,
+        committed: &CommittedResult,
+    ) -> Result<PublicationInput, ResultError> {
+        require_supported_version(committed.schema_version)?;
+        self.validate_publication_input(&committed.attempt_id, committed)
+            .await?;
+        Ok(committed.clone())
+    }
+
+    pub async fn page_members(
+        &self,
+        committed: &CommittedResult,
+        after_path: Option<&str>,
+        page_size: PageSize,
+    ) -> Result<ResultMemberPage, ResultError> {
+        self.resolve_committed(committed).await?;
+        let after = after_path
+            .map(MemberPath::parse)
+            .transpose()
+            .map_err(|_| ResultError::InvalidMemberPath)?
+            .map(|path| path.canonical())
+            .unwrap_or_default();
+        let fetch = i64::from(page_size.get()) + 1;
+        let rows = sqlx::query_as::<_, MemberRow>(MEMBER_PAGE_SELECT)
+            .bind(committed.attempt_id.as_str())
+            .bind(after)
+            .bind(fetch)
+            .fetch_all(&self.pool)
+            .await?;
+        let has_more = rows.len() > page_size.get() as usize;
+        let mut items = rows
+            .into_iter()
+            .take(page_size.get() as usize)
+            .map(MemberRow::into_member)
+            .collect::<Result<Vec<_>, _>>()?;
+        let next_after = has_more
+            .then(|| items.last().map(|item| item.path().to_owned()))
+            .flatten();
+        items.shrink_to_fit();
+        Ok(ResultMemberPage { items, next_after })
+    }
+
+    pub async fn lookup_member(
+        &self,
+        committed: &CommittedResult,
+        member_id: &OpaqueId,
+    ) -> Result<ResultMember, ResultError> {
+        self.resolve_committed(committed).await?;
+        sqlx::query_as::<_, MemberRow>(MEMBER_LOOKUP_SELECT)
+            .bind(committed.attempt_id.as_str())
+            .bind(member_id.as_str())
+            .fetch_optional(&self.pool)
+            .await?
+            .ok_or(ResultError::MemberNotFound)?
+            .into_member()
+    }
+
+    async fn validate_members(
+        &self,
+        attempt_id: &OpaqueId,
+        target: &TargetStore,
+    ) -> Result<(), ResultError> {
+        let mut rows = sqlx::query_as::<_, MemberRow>(MEMBER_SELECT)
+            .bind(attempt_id.as_str())
+            .fetch(&self.pool);
+        while let Some(row) = rows.try_next().await? {
+            let member = row.into_member()?;
+            validate_member_identity(attempt_id, &member)?;
+            let observed = target.verify_exact(&member.exact_identity()).await?;
+            if observed.identity != member.exact_identity()
+                || observed.size != member.size()
+                || observed.content_sha256.as_ref() != Some(member.content_sha256())
+            {
+                return Err(ResultError::Integrity);
+            }
+        }
+        Ok(())
+    }
+
+    async fn inventory_totals(&self, attempt_id: &OpaqueId) -> Result<(u64, u64), ResultError> {
+        let (count, bytes): (i64, i64) = sqlx::query_as(
+            "SELECT member_count, total_bytes FROM result_inventories WHERE attempt_id = $1",
+        )
+        .bind(attempt_id.as_str())
+        .fetch_one(&self.pool)
+        .await?;
+        Ok((count as u64, bytes as u64))
+    }
+
+    async fn persist_sealed(&self, input: &PublicationInput) -> Result<(), ResultError> {
+        let result = sqlx::query(
+            "UPDATE result_inventories SET
+                state = 'sealed', manifest_schema_version = $2, manifest_id = $3,
+                manifest_store = $4, manifest_bucket = $5, manifest_key = $6,
+                manifest_version_id = $7, manifest_sha256 = $8, manifest_size = $9
+             WHERE attempt_id = $1 AND state = 'closed'",
+        )
+        .bind(input.attempt_id.as_str())
+        .bind(i32::try_from(input.schema_version).map_err(|_| ResultError::SizeOverflow)?)
+        .bind(input.manifest_id.as_str())
+        .bind(&input.manifest.identity.target_store_id)
+        .bind(&input.manifest.identity.bucket)
+        .bind(&input.manifest.identity.key)
+        .bind(&input.manifest.identity.version_id)
+        .bind(input.manifest.content_sha256.as_str())
+        .bind(i64::try_from(input.manifest.size).map_err(|_| ResultError::SizeOverflow)?)
+        .execute(&self.pool)
+        .await?;
+        if result.rows_affected() != 1 {
+            return Err(ResultError::NotSealed);
+        }
+        Ok(())
+    }
+
+    async fn publication_input(
+        &self,
+        attempt_id: &OpaqueId,
+    ) -> Result<Option<PublicationInput>, ResultError> {
+        let row = sqlx::query_as::<_, InventoryRow>(
+            "SELECT state, manifest_schema_version, manifest_id, manifest_store,
+                    manifest_bucket, manifest_key, manifest_version_id, manifest_sha256,
+                    manifest_size, member_count, total_bytes
+             FROM result_inventories WHERE attempt_id = $1",
+        )
+        .bind(attempt_id.as_str())
+        .fetch_optional(&self.pool)
+        .await?
+        .ok_or(ResultError::InventoryNotFound)?;
+        row.into_publication_input(attempt_id)
+    }
+}
+
+const MEMBER_SELECT: &str = "SELECT member_id, path, size, object_store, object_bucket,
+    object_key, object_version_id, object_sha256
+    FROM result_inventory_members WHERE attempt_id = $1";
+const MEMBER_SELECT_ORDERED: &str = "SELECT member_id, path, size, object_store, object_bucket,
+    object_key, object_version_id, object_sha256
+    FROM result_inventory_members WHERE attempt_id = $1 ORDER BY path, member_id";
+const MEMBER_PAGE_SELECT: &str = "SELECT member_id, path, size, object_store, object_bucket,
+    object_key, object_version_id, object_sha256
+    FROM result_inventory_members
+    WHERE attempt_id = $1 AND path > $2 ORDER BY path, member_id LIMIT $3";
+const MEMBER_LOOKUP_SELECT: &str = "SELECT member_id, path, size, object_store, object_bucket,
+    object_key, object_version_id, object_sha256
+    FROM result_inventory_members WHERE attempt_id = $1 AND member_id = $2";
+
+#[derive(sqlx::FromRow)]
+struct MemberRow {
+    member_id: String,
+    path: String,
+    size: i64,
+    object_store: String,
+    object_bucket: String,
+    object_key: String,
+    object_version_id: Option<String>,
+    object_sha256: String,
+}
+
+impl MemberRow {
+    fn into_member(self) -> Result<ResultMember, ResultError> {
+        let id = OpaqueId::parse(self.member_id).map_err(|_| ResultError::Integrity)?;
+        let size = u64::try_from(self.size).map_err(|_| ResultError::Integrity)?;
+        let digest =
+            ContentSha256::parse(self.object_sha256).map_err(|_| ResultError::Integrity)?;
+        ResultMember::new(
+            id,
+            &self.path,
+            size,
+            ObjectIdentity {
+                target_store_id: self.object_store,
+                bucket: self.object_bucket,
+                key: self.object_key,
+                version_id: self.object_version_id,
+            },
+            digest,
+        )
+    }
+}
+
+#[derive(sqlx::FromRow)]
+struct InventoryRow {
+    state: String,
+    manifest_schema_version: Option<i32>,
+    manifest_id: Option<String>,
+    manifest_store: Option<String>,
+    manifest_bucket: Option<String>,
+    manifest_key: Option<String>,
+    manifest_version_id: Option<String>,
+    manifest_sha256: Option<String>,
+    manifest_size: Option<i64>,
+    member_count: i64,
+    total_bytes: i64,
+}
+
+impl InventoryRow {
+    fn into_publication_input(
+        self,
+        attempt_id: &OpaqueId,
+    ) -> Result<Option<PublicationInput>, ResultError> {
+        if self.state != "sealed" {
+            return Ok(None);
+        }
+        let schema_version =
+            u32::try_from(self.manifest_schema_version.ok_or(ResultError::Integrity)?)
+                .map_err(|_| ResultError::Integrity)?;
+        let manifest_id = OpaqueId::parse(self.manifest_id.ok_or(ResultError::Integrity)?)
+            .map_err(|_| ResultError::Integrity)?;
+        let digest = ContentSha256::parse(self.manifest_sha256.ok_or(ResultError::Integrity)?)
+            .map_err(|_| ResultError::Integrity)?;
+        Ok(Some(PublicationInput {
+            attempt_id: attempt_id.clone(),
+            manifest_id,
+            schema_version,
+            manifest: StoredObject {
+                identity: ObjectIdentity {
+                    target_store_id: self.manifest_store.ok_or(ResultError::Integrity)?,
+                    bucket: self.manifest_bucket.ok_or(ResultError::Integrity)?,
+                    key: self.manifest_key.ok_or(ResultError::Integrity)?,
+                    version_id: self.manifest_version_id,
+                },
+                content_sha256: digest,
+                size: u64::try_from(self.manifest_size.ok_or(ResultError::Integrity)?)
+                    .map_err(|_| ResultError::Integrity)?,
+            },
+            member_count: u64::try_from(self.member_count).map_err(|_| ResultError::Integrity)?,
+            total_bytes: u64::try_from(self.total_bytes).map_err(|_| ResultError::Integrity)?,
+        }))
+    }
+}
+
+fn validate_member_identity(
+    attempt_id: &OpaqueId,
+    member: &ResultMember,
+) -> Result<(), ResultError> {
+    validate_attempt_key(member.object().key()).map_err(|_| ResultError::InvalidObjectIdentity)?;
+    let prefix = format!("_remora/a/{}/o/", attempt_id.as_str());
+    if !member.object().key().starts_with(&prefix) {
+        return Err(ResultError::StaleAttempt);
+    }
+    Ok(())
+}
+
+fn require_supported_version(version: u32) -> Result<(), ResultError> {
+    if version != MANIFEST_SCHEMA_VERSION {
+        return Err(ResultError::UnsupportedManifestVersion(version));
+    }
+    Ok(())
+}
+
+fn is_unique_violation(error: &sqlx::Error, constraint: &str) -> bool {
+    error
+        .as_database_error()
+        .is_some_and(|error| error.constraint() == Some(constraint))
+}
+
+#[derive(Serialize)]
+struct ManifestHeader<'a> {
+    schema_version: u32,
+    attempt_id: &'a OpaqueId,
+}
+
+fn manifest_header(attempt_id: &OpaqueId) -> Vec<u8> {
+    let header = ManifestHeader {
+        schema_version: MANIFEST_SCHEMA_VERSION,
+        attempt_id,
+    };
+    let mut bytes = serde_json::to_vec(&header).expect("manifest header is serializable");
+    bytes.pop();
+    bytes.extend_from_slice(b",\"members\":[");
+    bytes
+}
+
+fn manifest_footer(member_count: u64, total_bytes: u64) -> Vec<u8> {
+    format!("],\"member_count\":{member_count},\"total_bytes\":{total_bytes}}}\n").into_bytes()
+}

crates/remora-result/src/lib.rs

@@ -0,0 +1,14 @@
+//! Durable, manifest-rooted result inventory and exact access primitives.
+
+mod error;
+mod inventory;
+mod migrate;
+mod types;
+
+pub use error::ResultError;
+pub use inventory::ResultInventory;
+pub use migrate::{migrate, migrator};
+pub use types::{
+    CommittedResult, MANIFEST_SCHEMA_VERSION, ManifestObjectIdentity, PageSize, PublicationInput,
+    ResultMember, ResultMemberPage,
+};

crates/remora-result/src/migrate.rs

@@ -0,0 +1,17 @@
+use remora_durable::DurableError;
+use sqlx::PgPool;
+
+const MIGRATION_TABLE: &str = "_remora_result_migrations";
+
+pub async fn migrate(pool: &PgPool) -> Result<(), DurableError> {
+    let mut migrator = sqlx::migrate!("./migrations");
+    migrator.dangerous_set_table_name(MIGRATION_TABLE);
+    migrator.run(pool).await?;
+    Ok(())
+}
+
+pub fn migrator() -> sqlx::migrate::Migrator {
+    let mut migrator = sqlx::migrate!("./migrations");
+    migrator.dangerous_set_table_name(MIGRATION_TABLE);
+    migrator
+}

crates/remora-result/src/types.rs

@@ -0,0 +1,165 @@
+use std::num::NonZeroU32;
+
+use remora_domain::{MemberPath, OpaqueId};
+use remora_target::{ContentSha256, ObjectIdentity, StoredObject};
+use serde::Serialize;
+
+use crate::ResultError;
+
+pub const MANIFEST_SCHEMA_VERSION: u32 = 1;
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+pub struct ResultMember {
+    id: OpaqueId,
+    path: String,
+    size: u64,
+    object: ManifestObjectIdentity,
+}
+
+impl ResultMember {
+    pub fn new(
+        id: OpaqueId,
+        path: &str,
+        size: u64,
+        object: ObjectIdentity,
+        content_sha256: ContentSha256,
+    ) -> Result<Self, ResultError> {
+        let path = MemberPath::parse(path)
+            .map_err(|_| ResultError::InvalidMemberPath)?
+            .canonical();
+        Ok(Self {
+            id,
+            path,
+            size,
+            object: ManifestObjectIdentity::new(object, content_sha256),
+        })
+    }
+
+    #[must_use]
+    pub fn exact_identity(&self) -> ObjectIdentity {
+        self.object.exact_identity()
+    }
+
+    #[must_use]
+    pub fn content_sha256(&self) -> &ContentSha256 {
+        &self.object.sha256
+    }
+
+    #[must_use]
+    pub fn id(&self) -> &OpaqueId {
+        &self.id
+    }
+
+    #[must_use]
+    pub fn path(&self) -> &str {
+        &self.path
+    }
+
+    #[must_use]
+    pub fn size(&self) -> u64 {
+        self.size
+    }
+
+    #[must_use]
+    pub fn object(&self) -> &ManifestObjectIdentity {
+        &self.object
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+pub struct ManifestObjectIdentity {
+    store: String,
+    bucket: String,
+    key: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    version_id: Option<String>,
+    #[serde(serialize_with = "serialize_sha256")]
+    sha256: ContentSha256,
+}
+
+fn serialize_sha256<S>(digest: &ContentSha256, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    serializer.serialize_str(digest.as_str())
+}
+
+impl ManifestObjectIdentity {
+    fn new(identity: ObjectIdentity, sha256: ContentSha256) -> Self {
+        Self {
+            store: identity.target_store_id,
+            bucket: identity.bucket,
+            key: identity.key,
+            version_id: identity.version_id,
+            sha256,
+        }
+    }
+
+    #[must_use]
+    pub fn exact_identity(&self) -> ObjectIdentity {
+        ObjectIdentity {
+            target_store_id: self.store.clone(),
+            bucket: self.bucket.clone(),
+            key: self.key.clone(),
+            version_id: self.version_id.clone(),
+        }
+    }
+
+    #[must_use]
+    pub fn store(&self) -> &str {
+        &self.store
+    }
+
+    #[must_use]
+    pub fn bucket(&self) -> &str {
+        &self.bucket
+    }
+
+    #[must_use]
+    pub fn key(&self) -> &str {
+        &self.key
+    }
+
+    #[must_use]
+    pub fn version_id(&self) -> Option<&str> {
+        self.version_id.as_deref()
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PublicationInput {
+    pub attempt_id: OpaqueId,
+    pub manifest_id: OpaqueId,
+    pub schema_version: u32,
+    pub manifest: StoredObject,
+    pub member_count: u64,
+    pub total_bytes: u64,
+}
+
+pub type CommittedResult = PublicationInput;
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ResultMemberPage {
+    pub items: Vec<ResultMember>,
+    pub next_after: Option<String>,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct PageSize(NonZeroU32);
+
+impl PageSize {
+    pub const MAX: u32 = 1_000;
+
+    pub fn new(value: u32) -> Result<Self, ResultError> {
+        let value = NonZeroU32::new(value).ok_or(ResultError::InvalidPageSize)?;
+        if value.get() > Self::MAX {
+            return Err(ResultError::InvalidPageSize);
+        }
+        Ok(Self(value))
+    }
+
+    #[must_use]
+    pub fn get(self) -> u32 {
+        self.0.get()
+    }
+}

crates/remora-result/tests/substrate.rs

@@ -0,0 +1,318 @@
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use remora_domain::OpaqueId;
+use remora_durable::{PoolConfig, connect, migrate as durable_migrate};
+use remora_result::{
+    MANIFEST_SCHEMA_VERSION, PageSize, ResultError, ResultInventory, ResultMember,
+    migrate as result_migrate,
+};
+use remora_target::{
+    ContentSha256, CreateObjectRequest, ObjectIdentity, TargetStore, TargetStoreConfig, hash_bytes,
+};
+use serde_json::Value;
+use sqlx::PgPool;
+
+static NEXT_ID: AtomicU64 = AtomicU64::new(0);
+
+fn unique_id(kind: &str) -> OpaqueId {
+    let nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("system time")
+        .as_nanos();
+    let sequence = NEXT_ID.fetch_add(1, Ordering::Relaxed);
+    OpaqueId::parse(format!("{kind}-{nanos:x}-{sequence:x}")).expect("opaque id")
+}
+
+async fn test_pool() -> Option<PgPool> {
+    let Some(url) = std::env::var("DATABASE_URL")
+        .ok()
+        .filter(|value| !value.is_empty())
+    else {
+        if std::env::var("REMORA_RESULT_REQUIRE").is_ok() {
+            panic!("DATABASE_URL is required when REMORA_RESULT_REQUIRE is set");
+        }
+        return None;
+    };
+    let pool = connect(
+        &url,
+        PoolConfig {
+            max_connections: 8,
+            acquire_timeout: Duration::from_secs(10),
+            idle_timeout: Some(Duration::from_secs(30)),
+        },
+    )
+    .await
+    .expect("connect postgres");
+    durable_migrate(&pool).await.expect("durable migration");
+    result_migrate(&pool).await.expect("result migration");
+    Some(pool)
+}
+
+fn target_store() -> Option<TargetStore> {
+    if std::env::var("REMORA_TARGET_ENDPOINT").is_err() {
+        if std::env::var("REMORA_RESULT_REQUIRE").is_ok() {
+            panic!("target fixture is required when REMORA_RESULT_REQUIRE is set");
+        }
+        return None;
+    }
+    TargetStore::connect(TargetStoreConfig::from_env().expect("target config"))
+        .map(Some)
+        .expect("connect target")
+}
+
+async fn stored_member(
+    target: &TargetStore,
+    attempt: &OpaqueId,
+    path: &str,
+    body: &[u8],
+) -> ResultMember {
+    let member_id = unique_id("member");
+    let stored = target
+        .put_create_only(
+            &CreateObjectRequest::for_attempt_object(attempt, &member_id, None),
+            body.to_vec(),
+        )
+        .await
+        .expect("store member");
+    ResultMember::new(
+        member_id,
+        path,
+        stored.size,
+        stored.identity,
+        stored.content_sha256,
+    )
+    .expect("member")
+}
+
+#[tokio::test]
+async fn single_member_manifest_resolves_pages_and_exact_access() {
+    let (Some(pool), Some(target)) = (test_pool().await, target_store()) else {
+        return;
+    };
+    let inventory = ResultInventory::new(pool);
+    let attempt = unique_id("attempt");
+    inventory.create(&attempt).await.expect("create inventory");
+    let member = stored_member(&target, &attempt, "models/é.bin", b"member bytes").await;
+    inventory
+        .add_member(&attempt, &member)
+        .await
+        .expect("add member");
+
+    let committed = inventory
+        .create_manifest(&attempt, &unique_id("manifest"), &target)
+        .await
+        .expect("create manifest");
+    assert_eq!(committed.schema_version, MANIFEST_SCHEMA_VERSION);
+    assert_eq!(committed.member_count, 1);
+    assert_eq!(committed.total_bytes, 12);
+    inventory
+        .validate_publication_input(&attempt, &committed)
+        .await
+        .expect("valid publication input");
+
+    let manifest_bytes = target
+        .read_exact(&committed.manifest.identity)
+        .await
+        .expect("read exact manifest");
+    assert_eq!(
+        hash_bytes(&manifest_bytes),
+        committed.manifest.content_sha256
+    );
+    let manifest: Value = serde_json::from_slice(&manifest_bytes).expect("manifest JSON");
+    assert_eq!(manifest["schema_version"], 1);
+    assert_eq!(manifest["member_count"], 1);
+    assert_eq!(manifest["members"][0]["path"], "models/é.bin");
+
+    let page = inventory
+        .page_members(&committed, None, PageSize::new(10).unwrap())
+        .await
+        .expect("page members");
+    assert_eq!(page.items.len(), 1);
+    assert_eq!(page.items[0], member);
+    assert!(page.next_after.is_none());
+    let found = inventory
+        .lookup_member(&committed, member.id())
+        .await
+        .expect("lookup exact member");
+    assert_eq!(found, member);
+}
+
+#[tokio::test]
+async fn rejects_normalization_collision_and_closed_inventory_mutation() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    let inventory = ResultInventory::new(pool);
+    let attempt = unique_id("attempt");
+    inventory.create(&attempt).await.expect("create inventory");
+    let first = ResultMember::new(
+        unique_id("member"),
+        "café/file",
+        1,
+        member_identity(&attempt, &unique_id("object")),
+        hash_bytes(b"a"),
+    )
+    .unwrap();
+    let collision = ResultMember::new(
+        unique_id("member"),
+        "cafe\u{301}/file",
+        1,
+        member_identity(&attempt, &unique_id("object")),
+        hash_bytes(b"b"),
+    )
+    .unwrap();
+    inventory.add_member(&attempt, &first).await.unwrap();
+    assert!(matches!(
+        inventory.add_member(&attempt, &collision).await,
+        Err(ResultError::PathCollision)
+    ));
+    inventory.close(&attempt).await.unwrap();
+    assert!(matches!(
+        inventory.add_member(&attempt, &collision).await,
+        Err(ResultError::InventoryNotOpen)
+    ));
+}
+
+#[tokio::test]
+async fn integrity_stale_attempt_and_descriptor_version_fail_closed() {
+    let (Some(pool), Some(target)) = (test_pool().await, target_store()) else {
+        return;
+    };
+    let inventory = ResultInventory::new(pool);
+    let attempt = unique_id("attempt");
+    inventory.create(&attempt).await.unwrap();
+    let stored = stored_member(&target, &attempt, "payload.bin", b"payload").await;
+    let invalid = ResultMember::new(
+        stored.id().clone(),
+        stored.path(),
+        stored.size() + 1,
+        stored.exact_identity(),
+        stored.content_sha256().clone(),
+    )
+    .unwrap();
+    inventory.add_member(&attempt, &invalid).await.unwrap();
+    assert!(matches!(
+        inventory
+            .create_manifest(&attempt, &unique_id("manifest"), &target)
+            .await,
+        Err(ResultError::Integrity)
+    ));
+
+    let other_attempt = unique_id("attempt");
+    let stale = ResultMember::new(
+        unique_id("member"),
+        "stale.bin",
+        0,
+        member_identity(&other_attempt, &unique_id("object")),
+        hash_bytes(b""),
+    )
+    .unwrap();
+    assert!(matches!(
+        inventory.add_member(&attempt, &stale).await,
+        Err(ResultError::StaleAttempt)
+    ));
+
+    let mut unsupported = placeholder_publication(&attempt);
+    unsupported.schema_version = MANIFEST_SCHEMA_VERSION + 1;
+    assert!(matches!(
+        inventory.resolve_committed(&unsupported).await,
+        Err(ResultError::UnsupportedManifestVersion(_))
+    ));
+}
+
+#[tokio::test]
+async fn pages_large_ordered_inventory_without_materializing_all_rows() {
+    let Some(pool) = test_pool().await else {
+        return;
+    };
+    let inventory = ResultInventory::new(pool.clone());
+    let attempt = unique_id("attempt");
+    inventory.create(&attempt).await.unwrap();
+    let digest = hash_bytes(b"");
+    for index in (0..2_500_u32).rev() {
+        let member_id = OpaqueId::parse(format!("member-{index:08}")).unwrap();
+        let member = ResultMember::new(
+            member_id,
+            &format!("inventory/{index:08}"),
+            0,
+            member_identity(&attempt, &unique_id("object")),
+            digest.clone(),
+        )
+        .unwrap();
+        inventory.add_member(&attempt, &member).await.unwrap();
+    }
+    inventory.close(&attempt).await.unwrap();
+    let mut committed = placeholder_publication(&attempt);
+    committed.member_count = 2_500;
+    sqlx::query(
+        "UPDATE result_inventories SET
+            state = 'sealed', manifest_schema_version = $2, manifest_id = $3,
+            manifest_store = $4, manifest_bucket = $5, manifest_key = $6,
+            manifest_version_id = $7, manifest_sha256 = $8, manifest_size = $9
+         WHERE attempt_id = $1",
+    )
+    .bind(attempt.as_str())
+    .bind(committed.schema_version as i32)
+    .bind(committed.manifest_id.as_str())
+    .bind(&committed.manifest.identity.target_store_id)
+    .bind(&committed.manifest.identity.bucket)
+    .bind(&committed.manifest.identity.key)
+    .bind(&committed.manifest.identity.version_id)
+    .bind(committed.manifest.content_sha256.as_str())
+    .bind(committed.manifest.size as i64)
+    .execute(&pool)
+    .await
+    .unwrap();
+
+    let mut after = None;
+    let mut observed = 0_u32;
+    loop {
+        let page = inventory
+            .page_members(&committed, after.as_deref(), PageSize::new(137).unwrap())
+            .await
+            .unwrap();
+        for member in &page.items {
+            assert_eq!(member.path(), format!("inventory/{observed:08}"));
+            observed += 1;
+        }
+        let Some(next) = page.next_after else {
+            break;
+        };
+        after = Some(next);
+    }
+    assert_eq!(observed, 2_500);
+}
+
+fn member_identity(attempt: &OpaqueId, object: &OpaqueId) -> ObjectIdentity {
+    ObjectIdentity {
+        target_store_id: "fixture-store".into(),
+        bucket: "fixture-bucket".into(),
+        key: format!("_remora/a/{}/o/{}", attempt.as_str(), object.as_str()),
+        version_id: None,
+    }
+}
+
+fn placeholder_publication(attempt: &OpaqueId) -> remora_result::PublicationInput {
+    let manifest_id = unique_id("manifest");
+    remora_result::PublicationInput {
+        attempt_id: attempt.clone(),
+        manifest_id: manifest_id.clone(),
+        schema_version: MANIFEST_SCHEMA_VERSION,
+        manifest: remora_target::StoredObject {
+            identity: ObjectIdentity {
+                target_store_id: "fixture-store".into(),
+                bucket: "fixture-bucket".into(),
+                key: remora_target::attempt_manifest_key(attempt, &manifest_id),
+                version_id: None,
+            },
+            content_sha256: ContentSha256::parse(
+                "sha256:0000000000000000000000000000000000000000000000000000000000000000",
+            )
+            .unwrap(),
+            size: 0,
+        },
+        member_count: 1,
+        total_bytes: 0,
+    }
+}

crates/remora-target/src/store.rs

@@ -258,6 +258,19 @@ impl TargetStore {
         })
     }
 
+    /// Stream and independently hash an exact identity without buffering it.
+    pub async fn verify_exact(
+        &self,
+        identity: &ObjectIdentity,
+    ) -> Result<ObjectMetadata, TargetError> {
+        let (content_sha256, size) = self.hash_exact(identity).await?;
+        Ok(ObjectMetadata {
+            identity: identity.clone(),
+            content_sha256: Some(content_sha256),
+            size,
+        })
+    }
+
     /// Issue a time-limited URL bound to one exact object identity.
     pub async fn presign_read(
         &self,