receive: add ents-receive, the one write path every mutation frontend shares
commit
310e646receive: add ents-receive, the one write path every mutation frontend shares
Orchestration above traits that already exist: gate policy (mandatory hosted vs. advisory local, via a Mode receive() takes explicitly, since the same write path must behave differently by deployment), redaction enforcement at ingest, and effect-footprint matching plus enqueue. Never the gate’s own judgment (ents-gate) and never an executor (ents-effect, a later phase this crate does not and must not link).
Defines EventSink (null and in-memory-deduplicating reference impls) and reconcile(), the boot-time scan that rebuilds exactly the obligations incremental receive() calls would have enqueued from repository state alone, per the work set query.workset defines - durability is a performance property, never a correctness one.
receive: define the receive() entry point, Proposal/RefTransition, and the Mode/Outcome/TxResult vocabulary receive: define EventSink plus NullEventSink and MemoryEventSink receive: add the boot-time reconciliation scan and incremental enqueue Assisted-by: Claude:claude-sonnet-4-6
Reviews
No reviews of this commit yet — record a verdict below.
Start a review
Cargo.lock
@@ -430,6 +430,24 @@
"thiserror",
]
+[[package]]
+name = "ents-receive"
+version = "0.0.0"
+dependencies = [
+ "ents-gate",
+ "ents-model",
+ "ents-query",
+ "ents-testutil",
+ "facet",
+ "facet-git-tree",
+ "gix",
+ "gix-hash",
+ "gix-object",
+ "gix-ref-store",
+ "rstest",
+ "thiserror",
+]
+
[[package]]
name = "ents-testutil"
version = "0.0.0"
Cargo.toml
@@ -5,6 +5,7 @@
"crates/ents-gate",
"crates/ents-model",
"crates/ents-query",
+ "crates/ents-receive",
"crates/ents-testutil",
"crates/gix-ref-store",
]
@@ -22,6 +23,7 @@
ents-gate = { path = "crates/ents-gate" }
ents-model = { path = "crates/ents-model" }
ents-query = { path = "crates/ents-query" }
+ents-receive = { path = "crates/ents-receive" }
ents-testutil = { path = "crates/ents-testutil" }
gix-ref-store = { path = "crates/gix-ref-store" }
arborium = { version = "2.18", default-features = false, features = [
crates/ents-receive/Cargo.toml
@@ -1,0 +1,25 @@
+[package]
+name = "ents-receive"
+version = "0.0.0"
+edition.workspace = true
+publish.workspace = true
+license.workspace = true
+
+[dependencies]
+ents-gate = { workspace = true }
+ents-model = { workspace = true }
+ents-query = { workspace = true }
+facet-git-tree = { workspace = true }
+gix = { workspace = true }
+gix-hash = { workspace = true }
+gix-object = { workspace = true }
+gix-ref-store = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+ents-testutil = { workspace = true }
+facet = { workspace = true }
+rstest = { workspace = true }
+
+[lints]
+workspace = true
crates/ents-receive/src/error.rs
@@ -1,0 +1,54 @@
+//! `ents-receive`'s infrastructure error type.
+//!
+//! An [`Error`] means `receive` could not *reach* an outcome (a store read
+//! failed, an object is missing or undecodable, an [`crate::EventSink`]
+//! failed to durably enqueue) — as opposed to an ordinary refusal
+//! ([`ents_gate::Verdict::Fail`], [`crate::TxResult::Refused`],
+//! [`crate::TxResult::Rejected`], [`crate::TxResult::Redacted`]), which is a
+//! reached judgment carried inside [`crate::Outcome`], not an `Err`.
+
+use gix_hash::ObjectId;
+
+/// Everything that can prevent `receive` from reaching an outcome.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+ /// The gate could not evaluate a proposed transition.
+ #[error("gate evaluation failed: {0}")]
+ Gate(#[from] ents_gate::Error),
+
+ /// The ref store's read or write half failed.
+ #[error("ref store operation failed: {0}")]
+ Refs(#[from] gix_ref_store::Error),
+
+ /// The query evaluator could not compute an entry or work set.
+ #[error("query evaluation failed: {0}")]
+ Eval(#[from] ents_query::EvalError),
+
+ /// An `EventSink` failed to durably enqueue an obligation.
+ ///
+ /// `receive.never-blocks` still holds: this only ever wraps a genuine
+ /// sink failure (durable-queue I/O, hosted), never effect evaluation
+ /// itself, which this crate never performs.
+ #[error("event sink failed to enqueue ({effect}, {oid}): {detail}")]
+ Sink {
+ /// The effect the obligation was for.
+ effect: String,
+ /// The commit the obligation names.
+ oid: ObjectId,
+ /// What failed, human-readable.
+ detail: String,
+ },
+
+ /// An object could not be read or decoded while scanning
+ /// `refs/meta/effects/*` or `refs/meta/redactions/*`.
+ #[error("object {oid} could not be read: {detail}")]
+ Decode {
+ /// The undecodable object.
+ oid: ObjectId,
+ /// What failed, human-readable.
+ detail: String,
+ },
+}
+
+/// The `Result` alias every fallible `ents-receive` operation returns.
+pub type Result<T> = std::result::Result<T, Error>;
crates/ents-receive/src/lib.rs
@@ -1,0 +1,91 @@
+//! `receive`: the one write path every mutation frontend shares
+//! (`docs/spec/receive.sdoc`).
+//!
+//! This crate's single responsibility is orchestration above traits that
+//! already exist by the time it lands: gate policy (mandatory hosted,
+//! advisory local), redaction enforcement at ingest, and effect-footprint
+//! matching plus enqueue — never the gate's own judgment (`ents-gate`),
+//! never the query algebra (`ents-query`), and never an executor
+//! (`ents-effect`, a later phase this crate must never link,
+//! `arch.query-effect-split`).
+//!
+//! # Spec coverage
+//!
+//! From `docs/spec/receive.sdoc`:
+//!
+//! - `receive.unit`, `receive.shared-path` — [`receive`]: the sole
+//! mutation entry point, identical for every frontend; only the trait
+//! implementations and [`Mode`] differ.
+//! - `receive.proposal-shape` — [`Proposal`], [`RefTransition`],
+//! [`TransportAuth`].
+//! - `receive.refstore-seam` — [`receive`] takes `&dyn RefStore`, the full
+//! read/CAS seam (`arch.refstore-read-cas-split`).
+//! - `receive.object-access` — object access uses only `gix_object::Find`
+//! and `gix_object::Write`; see [`receive`]'s own doc for the one
+//! deliberate deviation (`gix_object::Exists` omitted — a fixture gap,
+//! not a design choice) and for the quarantine-directory note.
+//! - `receive.event-sink`, `receive.never-blocks` — [`EventSink`]; enqueue
+//! is the entire synchronous cost `receive` adds, and it is computed via
+//! each effect's static footprint, never a re-scan of every effect on
+//! every push.
+//! - `receive.dedup` — [`MemoryEventSink`]'s `(effect, oid)` set.
+//! - `receive.reconstructible` — [`reconcile`], the boot-time scan that
+//! rebuilds the exact obligations incremental `receive` calls would have
+//! enqueued, from repository state alone (`query.workset`).
+//! - `receive.redaction-admin-only` — a consequence of composition, not new
+//! code: `refs/meta/redactions/*` already falls through `ents-gate`'s
+//! default authorization arm, which requires admin-registered provenance
+//! for every namespace without its own carve-out; this crate's own test
+//! suite pins that composition at the `receive` level.
+//! - `receive.redaction-ingest` — [`receive`]'s first step: any proposal
+//! object matching a recorded redaction target refuses the whole batch.
+//!
+//! # Examples
+//!
+//! An end-to-end local write path: advisory gate, null sink — the shape
+//! `receive.adoc`'s phase-4 exit criterion runs.
+//!
+//! ```
+//! use ents_gate::Config;
+//! use ents_model::{Provenance, namespace};
+//! use ents_receive::{Mode, NullEventSink, Proposal, RefTransition, TxResult, receive};
+//! use ents_testutil::{Keypair, MemRefStore, ObjectStore, enroll_member, write_meta_entity};
+//!
+//! let refs = MemRefStore::default();
+//! let objects = ObjectStore::default();
+//! let admin = Keypair::from_seed(1);
+//!
+//! enroll_member(&refs, &objects, "admin", &admin, Provenance::AdminRegistered, 100);
+//! let config_ref: gix::refs::FullName = namespace::CONFIG_REF.try_into().expect("valid");
+//! let tip = write_meta_entity(
+//! &refs, &objects, config_ref.clone(), &Config { epoch: Some(200) }, Some(&admin), 200,
+//! );
+//!
+//! // The fixture already moved the ref; re-propose the same tip through
+//! // `receive` against a pre-write copy, the way a CLI would.
+//! let before = refs.fetched_copy();
+//! before.remove(config_ref.as_ref());
+//! let proposal = Proposal {
+//! transitions: vec![RefTransition { name: config_ref, old: None, new: Some(tip) }],
+//! objects: vec![tip],
+//! auth: None,
+//! };
+//!
+//! let outcome = receive(&before, &objects, &NullEventSink, &proposal, Mode::Advisory)
+//! .expect("evaluates");
+//! assert_eq!(outcome.result, TxResult::Applied);
+//! ```
+
+mod error;
+mod outcome;
+mod proposal;
+mod receive;
+mod reconcile;
+mod sink;
+
+pub use error::{Error, Result};
+pub use outcome::{Mode, Outcome, TxResult};
+pub use proposal::{Proposal, RefTransition, TransportAuth};
+pub use receive::receive;
+pub use reconcile::reconcile;
+pub use sink::{EventSink, MemoryEventSink, NullEventSink};
crates/ents-receive/src/outcome.rs
@@ -1,0 +1,94 @@
+//! `receive`'s gate policy (mandatory or advisory) and the outcome it
+//! reports: which verdict each proposed transition got, and whether the
+//! batch actually landed.
+
+use gix::refs::FullName;
+use gix_hash::ObjectId;
+
+use ents_gate::Verdict;
+
+/// Which of the two gate policies `receive.adoc` names governs one call:
+/// abort the whole batch on a failing verdict (`gate.mandatory-hosted`), or
+/// accept the write regardless and only annotate (`gate.advisory-local`).
+///
+/// The gate itself ([`ents_gate::verify`]) is one pure function evaluated
+/// identically either way (`gate.call-sites`); `Mode` is the policy
+/// [`crate::receive`] applies to a *failing* verdict, which is exactly the
+/// orchestration the development plan assigns to this crate — the gate
+/// crate never sees a `Mode`, and could not: it has no write path to gate.
+///
+/// # Examples
+///
+/// ```
+/// use ents_receive::Mode;
+///
+/// let mode = Mode::Advisory;
+/// assert_eq!(mode, Mode::Advisory);
+/// ```
+// @relation(gate.mandatory-hosted, gate.advisory-local, scope=file)
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Mode {
+ /// The hosted policy: a failing verdict against any transition in the
+ /// batch aborts the whole batch before any ref is updated
+ /// (`gate.mandatory-hosted`).
+ Mandatory,
+ /// The local policy: every transition is written regardless of its
+ /// verdict; a failing verdict only annotates the result, never blocks
+ /// it (`gate.advisory-local`).
+ Advisory,
+}
+
+/// What happened to one [`crate::Proposal`]'s ref-transaction batch.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TxResult {
+ /// Every transition in the batch was written atomically.
+ Applied,
+ /// [`Mode::Mandatory`] aborted the whole batch before attempting a
+ /// write, because at least one transition's verdict failed
+ /// (`gate.mandatory-hosted`). See [`crate::Outcome::verdicts`] for
+ /// which one and why.
+ Refused,
+ /// The underlying store rejected the compare-and-swap: `name`'s
+ /// current value no longer matched the precondition read at
+ /// evaluation time — a genuine race, reported in the gate's own
+ /// vocabulary (`Requirement::AtomicCas`) as the gate crate's docs
+ /// anticipate for exactly this caller.
+ Rejected {
+ /// The ref whose precondition was stale.
+ name: FullName,
+ },
+ /// The batch introduced an object matching a previously recorded
+ /// redaction target; the whole batch was refused before any verdict
+ /// was even evaluated, so a redacted hole cannot be silently refilled
+ /// by re-pushing the same bytes (`receive.redaction-ingest`).
+ Redacted {
+ /// The offending object id.
+ oid: ObjectId,
+ },
+}
+
+/// The result of one [`crate::receive`] call: every transition's verdict,
+/// and what happened to the batch as a whole.
+///
+/// # Examples
+///
+/// ```
+/// use ents_receive::{Outcome, TxResult};
+///
+/// let outcome = Outcome {
+/// verdicts: vec![],
+/// result: TxResult::Applied,
+/// };
+/// assert_eq!(outcome.result, TxResult::Applied);
+/// ```
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Outcome {
+ /// Each proposed transition's refname and the gate's verdict on it,
+ /// in proposal order. Present under both [`Mode`]s: mandatory callers
+ /// use it to see which refusal aborted the batch
+ /// (`gate.verdict-reason`); advisory callers render it to the user
+ /// regardless of [`Outcome::result`] (`gate.advisory-local`).
+ pub verdicts: Vec<(FullName, Verdict)>,
+ /// What happened to the batch.
+ pub result: TxResult,
+}
crates/ents-receive/src/proposal.rs
@@ -1,0 +1,97 @@
+//! The fourth argument's shape (`receive.proposal-shape`): the ref
+//! transitions a caller proposes, the new objects that accompany them, and
+//! any transport-auth evidence the frontend collected.
+
+use gix::refs::FullName;
+use gix_hash::ObjectId;
+
+/// One proposed ref transition: `(refname, old-oid, new-oid)`, exactly the
+/// triple `receive.proposal-shape` names.
+///
+/// `old` is the frontier the *proposal* claims — what a `git push` command
+/// line reports as its own base, or what a local UI last read. [`crate::receive`]
+/// never trusts it for admission: [`ents_gate::verify`] re-reads the actual
+/// current tip itself (the same snapshot every other check uses), and a
+/// mismatch between a claimed `old` and the store's real tip surfaces as an
+/// ordinary [`crate::TxResult::Rejected`] once the transaction is attempted,
+/// the same "someone moved this ref first" outcome a real push would see.
+///
+/// # Examples
+///
+/// ```
+/// use ents_receive::RefTransition;
+///
+/// let transition = RefTransition {
+/// name: "refs/meta/issues/1".try_into().expect("valid"),
+/// old: None,
+/// new: Some(gix_hash::ObjectId::null(gix_hash::Kind::Sha1)),
+/// };
+/// assert!(transition.old.is_none());
+/// ```
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RefTransition {
+ /// The ref being updated.
+ pub name: FullName,
+ /// The tip the proposal claims as its base, or `None` for creation.
+ pub old: Option<ObjectId>,
+ /// The proposed new tip, or `None` to delete the ref.
+ pub new: Option<ObjectId>,
+}
+
+/// Transport-level authentication evidence a frontend collected: a
+/// signed-push credential, a smart-HTTP session, or nothing for a frontend
+/// whose transport carries no separate authentication (`receive.proposal-shape`).
+///
+/// This is a connection-level ACL input for `refs/heads/*` only
+/// (`gate.principled-split`) — no such ACL policy is defined yet anywhere in
+/// the spec, so [`crate::receive`] accepts and threads this value through
+/// without interpreting it; it is never substituted for the tip invariant on
+/// a `refs/meta/*` update, which is the one thing `receive.proposal-shape`
+/// actually requires today. Defining and enforcing the `refs/heads/*` ACL
+/// itself is future work with no requirement id yet to hang it on.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TransportAuth {
+ /// Opaque evidence bytes: a signed-push certificate, a session token,
+ /// or whatever shape a future frontend needs. `receive` never parses
+ /// this; only a future `refs/heads/*` ACL check would.
+ pub evidence: Vec<u8>,
+}
+
+/// The fourth argument to [`crate::receive`]: every proposed ref transition
+/// this call attempts, the object ids the proposal introduces, and any
+/// transport-auth evidence (`receive.proposal-shape`).
+///
+/// # Examples
+///
+/// ```
+/// use ents_receive::{Proposal, RefTransition};
+///
+/// let proposal = Proposal {
+/// transitions: vec![RefTransition {
+/// name: "refs/meta/issues/1".try_into().expect("valid"),
+/// old: None,
+/// new: Some(gix_hash::ObjectId::null(gix_hash::Kind::Sha1)),
+/// }],
+/// objects: vec![gix_hash::ObjectId::null(gix_hash::Kind::Sha1)],
+/// auth: None,
+/// };
+/// assert_eq!(proposal.transitions.len(), 1);
+/// ```
+// @relation(receive.proposal-shape, scope=file)
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct Proposal {
+ /// The proposed ref transitions this call attempts, applied together
+ /// as one atomic batch (`arch.refstore-read-cas-split`,
+ /// `gate.atomic-cas`).
+ pub transitions: Vec<RefTransition>,
+ /// The object ids this proposal introduces, already durably present in
+ /// the object store [`crate::receive`] is handed (the frontend's job
+ /// per `receive.shared-path`: the CLI and local UI write directly, and
+ /// smart-HTTP unpacks the incoming pack, before `receive` is ever
+ /// called). `receive` checks each of these against the redaction list
+ /// at ingest time (`receive.redaction-ingest`).
+ pub objects: Vec<ObjectId>,
+ /// Transport-auth evidence the frontend collected, or `None`. See
+ /// [`TransportAuth`] for why `receive` does not interpret this today.
+ pub auth: Option<TransportAuth>,
+}
crates/ents-receive/src/receive.rs
@@ -1,0 +1,242 @@
+//! `receive`: the sole entry point through which a meta-ref or branch ref
+//! is mutated (`receive.unit`).
+
+use std::collections::HashSet;
+
+use ents_gate::{Update, verify};
+use ents_model::Redaction;
+use ents_query::Transition;
+use gix_hash::ObjectId;
+use gix_object::Find;
+use gix_ref_store::{Expected, RefEdit, RefStore, TxOutcome};
+
+use crate::error::Result;
+use crate::outcome::{Mode, Outcome, TxResult};
+use crate::proposal::Proposal;
+use crate::reconcile::{commit_tree, enqueue_matches};
+use crate::sink::EventSink;
+
+const REDACTIONS_PREFIX: &str = "refs/meta/redactions/";
+
+/// The sole entry point through which a meta-ref or branch ref is mutated
+/// (`receive.unit`).
+///
+/// Gate evaluation, redaction enforcement, effect-footprint matching, and
+/// enqueue all live here, above the `RefStore` (`refs`), object-store
+/// (`objects`), and [`EventSink`] (`events`) traits this function is
+/// handed — never duplicated in a caller (`receive.unit`,
+/// `arch.gate-receive-split`). Every mutation frontend (the CLI, the local
+/// UI, a hosted smart-HTTP hook) MUST call exactly this function
+/// in-process, with only the trait implementations and `mode` differing
+/// (`receive.shared-path`): a `LooseRefStore` and a null sink locally, a
+/// Postgres-backed store and a durable queue hosted.
+///
+/// # Order of operations
+///
+/// 1. **Redaction ingest** (`receive.redaction-ingest`): every object id
+/// `proposal.objects` introduces is checked against the redaction
+/// targets recorded under `refs/meta/redactions/*`. A match refuses the
+/// *entire* batch before any verdict is even evaluated — a redacted
+/// hole cannot be silently refilled by re-pushing the same bytes.
+/// 2. **Gate evaluation** (`receive.refstore-seam`): every proposed
+/// transition is judged by the identical [`ents_gate::verify`] every
+/// other call site uses (`gate.call-sites`), read against `refs`'s read
+/// half only.
+/// 3. **Gate policy** (`Mode`): [`Mode::Mandatory`] aborts the whole batch
+/// before writing anything if any verdict failed
+/// (`gate.mandatory-hosted`); [`Mode::Advisory`] writes every transition
+/// regardless of its verdict (`gate.advisory-local`) — the verdicts are
+/// still returned for the caller to render.
+/// 4. **Atomic write**: every transition lands as one
+/// [`RefStore::transaction`] call, so either the whole batch applies or
+/// none of it does (`gate.atomic-cas`). A stale precondition — another
+/// writer moved a ref between step 2 and here — surfaces as
+/// [`TxResult::Rejected`], in the gate's own vocabulary
+/// (`Requirement::AtomicCas`).
+/// 5. **Enqueue** (`receive.event-sink`, `receive.never-blocks`): once the
+/// batch is durably applied, every known effect's static footprint is
+/// matched against each transition and the entry set — `trigger −
+/// results(self, any)`, `query.workset` — is enqueued into `events`.
+/// This is the entire synchronous cost added to the write; no effect is
+/// evaluated here.
+///
+/// # Object access
+///
+/// Per `receive.object-access`, object access here uses only gitoxide's
+/// own traits — [`Find`] and [`gix_object::Write`] — never a private
+/// object-access trait. `gix_object::Exists` is gitoxide's third named
+/// trait for this seam; it is omitted from this signature because the
+/// shared fixture (`ents_testutil::ObjectStore`, from the external
+/// `facet-git-tree` crate) does not implement it — every existence check
+/// this crate needs goes through `Find` instead (`try_find(..).is_some()`),
+/// which is not a private trait and keeps `arch.no-object-store-trait`
+/// intact.
+///
+/// Which object directory `objects` resolves through (the common odb, never
+/// a git hook's quarantine directory, until its transaction commits) is the
+/// composition root's responsibility to wire, not this function's: `receive`
+/// only ever sees the seam it is handed.
+///
+/// # Errors
+///
+/// A [`crate::Error`] means `receive` could not reach an outcome at all
+/// (a store or object read failed) — distinct from every variant of
+/// [`Outcome`], which is a reached judgment.
+///
+/// # Examples
+///
+/// A minimal advisory, null-sink round trip: enroll an admin, set the
+/// epoch, then land a signed issue mutation.
+///
+/// ```
+/// use ents_gate::Config;
+/// use ents_model::{Issue, Provenance, namespace};
+/// use ents_receive::{Mode, NullEventSink, Proposal, RefTransition, TxResult, receive};
+/// use ents_testutil::{Keypair, MemRefStore, ObjectStore, enroll_member, write_meta_entity};
+///
+/// let refs = MemRefStore::default();
+/// let objects = ObjectStore::default();
+/// let admin = Keypair::from_seed(1);
+///
+/// enroll_member(&refs, &objects, "admin", &admin, Provenance::AdminRegistered, 100);
+/// let config_ref: gix::refs::FullName = namespace::CONFIG_REF.try_into().expect("valid");
+/// write_meta_entity(&refs, &objects, config_ref, &Config { epoch: Some(200) }, Some(&admin), 200);
+///
+/// let issue = Issue {
+/// title: "t".into(), body: "b".into(), state: "open".into(),
+/// assignees: vec![], labels: vec![],
+/// };
+/// let name: gix::refs::FullName = "refs/meta/issues/1".try_into().expect("valid");
+/// // write_meta_entity signs and lands the commit's object graph, but does
+/// // not move the ref through the gate — that is exactly receive's job.
+/// let tip = {
+/// let tree = facet_git_tree::serialize_into(&issue, &objects).expect("serializes");
+/// let trailers = ents_model::trailer::Trailers { ents_ref: Some(name.clone()), schema_version: None };
+/// let message = format!("Mutate {}\n\n{}", name.as_bstr(), trailers.render());
+/// ents_testutil::write_commit(&objects, &ents_testutil::CommitSpec {
+/// tree, parents: vec![], message, seconds: 300,
+/// }, Some(&admin))
+/// };
+///
+/// let proposal = Proposal {
+/// transitions: vec![RefTransition { name: name.clone(), old: None, new: Some(tip) }],
+/// objects: vec![tip],
+/// auth: None,
+/// };
+///
+/// let outcome = receive(&refs, &objects, &NullEventSink, &proposal, Mode::Advisory).expect("evaluates");
+/// assert_eq!(outcome.result, TxResult::Applied);
+/// assert!(outcome.verdicts[0].1.is_pass());
+/// ```
+// @relation(receive.unit, receive.shared-path, receive.refstore-seam, receive.object-access, scope=function)
+pub fn receive(
+ refs: &dyn RefStore,
+ objects: &(impl Find + gix_object::Write),
+ events: &dyn EventSink,
+ proposal: &Proposal,
+ mode: Mode,
+) -> Result<Outcome> {
+ // @relation(receive.redaction-ingest, scope=function)
+ if let Some(oid) = first_redacted(refs, objects, proposal)? {
+ return Ok(Outcome {
+ verdicts: Vec::new(),
+ result: TxResult::Redacted { oid },
+ });
+ }
+
+ let mut verdicts = Vec::with_capacity(proposal.transitions.len());
+ let mut edits = Vec::with_capacity(proposal.transitions.len());
+ let mut query_transitions = Vec::with_capacity(proposal.transitions.len());
+ let mut any_failed = false;
+
+ for transition in &proposal.transitions {
+ let old = refs.get(transition.name.as_ref())?;
+ // gate.call-sites: the identical function every other call site uses.
+ // receive.redaction-admin-only is a consequence of this composition:
+ // `verify` already refuses refs/meta/redactions/* to a non-admin
+ // signer via its default namespace-authorization arm, regardless of
+ // any refs/meta/config role rule, so no separate check is needed
+ // here.
+ // @relation(gate.call-sites, receive.redaction-admin-only, scope=function)
+ let verdict = verify(
+ refs,
+ objects,
+ &Update {
+ name: transition.name.clone(),
+ new: transition.new,
+ },
+ )?;
+ any_failed |= !verdict.is_pass();
+ verdicts.push((transition.name.clone(), verdict));
+
+ let expected = old.map_or(Expected::MustNotExist, Expected::MustExistAndMatch);
+ edits.push(RefEdit {
+ name: transition.name.clone(),
+ expected,
+ new: transition.new,
+ });
+ query_transitions.push(Transition {
+ name: transition.name.clone(),
+ old,
+ new: transition.new,
+ });
+ }
+
+ // gate.mandatory-hosted: abort the whole batch before writing anything.
+ // gate.advisory-local is the fallthrough: every transition is written
+ // below regardless of `any_failed`.
+ // @relation(gate.mandatory-hosted, scope=function)
+ if any_failed && mode == Mode::Mandatory {
+ return Ok(Outcome {
+ verdicts,
+ result: TxResult::Refused,
+ });
+ }
+
+ let result = if edits.is_empty() {
+ TxResult::Applied
+ } else {
+ match refs.transaction(&edits)? {
+ TxOutcome::Applied => TxResult::Applied,
+ TxOutcome::Rejected { name } => TxResult::Rejected { name },
+ }
+ };
+
+ // receive.event-sink, receive.never-blocks: enqueue is the entire
+ // synchronous cost; no effect is evaluated here.
+ if result == TxResult::Applied {
+ for transition in &query_transitions {
+ enqueue_matches(refs, objects, events, transition)?;
+ }
+ }
+
+ Ok(Outcome { verdicts, result })
+}
+
+/// The first object in `proposal.objects` that matches a target recorded
+/// under `refs/meta/redactions/*`, if any (`receive.redaction-ingest`).
+fn first_redacted(
+ refs: &dyn gix_ref_store::RefStoreRead,
+ objects: &impl Find,
+ proposal: &Proposal,
+) -> Result<Option<ObjectId>> {
+ if proposal.objects.is_empty() {
+ return Ok(None);
+ }
+ let mut targets = HashSet::new();
+ for entry in refs.iter_prefix(REDACTIONS_PREFIX)? {
+ let (_, tip) = entry?;
+ let Some(tree) = commit_tree(objects, tip)? else {
+ continue;
+ };
+ let Ok(redaction) = facet_git_tree::deserialize::<Redaction>(&tree, objects) else {
+ continue;
+ };
+ targets.insert(redaction.target());
+ }
+ Ok(proposal
+ .objects
+ .iter()
+ .copied()
+ .find(|oid| targets.contains(oid)))
+}
crates/ents-receive/src/reconcile.rs
@@ -1,0 +1,160 @@
+//! The boot-time reconciliation scan (`receive.reconstructible`): the
+//! reference proof that the obligation queue needs no state `receive`
+//! itself did not already have available in the repository.
+
+use ents_model::Effect;
+use ents_query::{Evaluator, Query};
+use gix::refs::FullName;
+use gix_hash::ObjectId;
+use gix_object::{CommitRef, Find, Kind};
+use gix_ref_store::RefStoreRead;
+
+use crate::error::{Error, Result};
+use crate::sink::EventSink;
+
+const EFFECTS_PREFIX: &str = "refs/meta/effects/";
+
+/// Every effect definition currently readable under `refs/meta/effects/*`,
+/// as `(name, parsed trigger)`.
+///
+/// An effect whose tree cannot be read, or whose `trigger` fails to parse,
+/// is skipped rather than failing the scan: `receive.validation`
+/// (`ents-model`, `ents-query`) is what keeps a *newly written* effect
+/// well-formed, but a scan reused across every future push must stay
+/// resilient to a pre-existing malformed one rather than let it take down
+/// every push after it (the same spirit as `receive.never-blocks`, applied
+/// to a scan that runs on `receive`'s own hot path).
+// @relation(receive.event-sink, scope=function)
+fn known_effects(refs: &dyn RefStoreRead, objects: &impl Find) -> Result<Vec<(String, Query)>> {
+ let mut effects = Vec::new();
+ for entry in refs.iter_prefix(EFFECTS_PREFIX)? {
+ let (name, tip) = entry?;
+ let Some(short) = short_effect_name(&name) else {
+ continue;
+ };
+ let Some(tree) = commit_tree(objects, tip)? else {
+ continue;
+ };
+ let Ok(effect) = facet_git_tree::deserialize::<Effect>(&tree, objects) else {
+ continue;
+ };
+ let Ok(trigger) = effect.trigger.parse::<Query>() else {
+ continue;
+ };
+ effects.push((short, trigger));
+ }
+ Ok(effects)
+}
+
+/// The effect name segment of a `refs/meta/effects/<name>` refname, or
+/// `None` for anything deeper or shallower (mirrors the results-namespace
+/// scan in `ents-query`'s evaluator).
+fn short_effect_name(name: &FullName) -> Option<String> {
+ let path = name.as_bstr().to_string();
+ let short = path.strip_prefix(EFFECTS_PREFIX)?;
+ (!short.is_empty() && !short.contains('/')).then(|| short.to_owned())
+}
+
+/// The tree of the commit at `oid`, or `None` if `oid` is missing or not a
+/// commit — treated as "this ref is unreadable", never a hard failure of
+/// the whole scan. Shared with [`crate::receive`]'s redaction-target scan.
+pub(crate) fn commit_tree(objects: &impl Find, oid: ObjectId) -> Result<Option<ObjectId>> {
+ let mut buf = Vec::new();
+ let Some(data) = objects
+ .try_find(&oid, &mut buf)
+ .map_err(|source| Error::Decode {
+ oid,
+ detail: source.to_string(),
+ })?
+ else {
+ return Ok(None);
+ };
+ if data.kind != Kind::Commit {
+ return Ok(None);
+ }
+ let Ok(commit) = CommitRef::from_bytes(data.data, oid.kind()) else {
+ return Ok(None);
+ };
+ Ok(Some(commit.tree()))
+}
+
+/// The full, reconciliation-grade obligation scan
+/// (`receive.reconstructible`): for every effect currently defined, compute
+/// its outstanding work set (`trigger − results(self, any)`,
+/// `query.workset`) against current ref state and enqueue every commit
+/// still owed a result.
+///
+/// A composition root calls this once at startup, before serving further
+/// pushes, so an `EventSink` that lost its queued events on crash (the null
+/// sink always; the in-memory reference sink after a restart) recovers
+/// exactly the same obligations incremental `receive` calls would have
+/// enqueued — the queue is reconstructible from repository state alone,
+/// with the dedup key (`receive.dedup`) unchanged by reconciliation.
+///
+/// # Errors
+///
+/// Fails only on a ref-store or object-store read failure, or a sink
+/// failure; a malformed individual effect definition is skipped, not an
+/// error (this module's private effect-scan helper treats an unreadable
+/// tree or an unparsable trigger as "no match", never a hard failure).
+///
+/// # Examples
+///
+/// ```
+/// use ents_model::Effect;
+/// use ents_receive::{MemoryEventSink, reconcile};
+/// use ents_testutil::{MemRefStore, ObjectStore, advance_ref, write_meta_entity};
+///
+/// let refs = MemRefStore::default();
+/// let objects = ObjectStore::default();
+/// let commits = advance_ref(&refs, &objects, "refs/heads/main", 1, 100);
+///
+/// let effect = Effect {
+/// trigger: "rev(refs/heads/main)".to_owned(),
+/// toolchains: vec![],
+/// run: "true".to_owned(),
+/// };
+/// let name: gix::refs::FullName = "refs/meta/effects/unit".try_into().expect("valid");
+/// write_meta_entity(&refs, &objects, name, &effect, None, 200);
+///
+/// let sink = MemoryEventSink::default();
+/// reconcile(&refs, &objects, &sink).expect("reconciles");
+/// assert_eq!(sink.pending(), vec![("unit".to_owned(), commits[0])]);
+/// ```
+// @relation(receive.reconstructible, query.workset, scope=function)
+pub fn reconcile(
+ refs: &dyn RefStoreRead,
+ objects: &impl Find,
+ events: &dyn EventSink,
+) -> Result<()> {
+ let evaluator = Evaluator::new(refs, objects);
+ for (name, trigger) in known_effects(refs, objects)? {
+ for oid in evaluator.outstanding(&name, &trigger)? {
+ enqueue(events, &name, oid)?;
+ }
+ }
+ Ok(())
+}
+
+/// Enqueue matches for `transition` against every known effect
+/// (`receive.event-sink`): the incremental counterpart to [`reconcile`],
+/// called by [`crate::receive`] once per successfully applied transition.
+// @relation(receive.event-sink, query.workset, scope=function)
+pub(crate) fn enqueue_matches(
+ refs: &dyn RefStoreRead,
+ objects: &impl Find,
+ events: &dyn EventSink,
+ transition: &ents_query::Transition,
+) -> Result<()> {
+ let evaluator = Evaluator::new(refs, objects);
+ for (name, trigger) in known_effects(refs, objects)? {
+ for oid in evaluator.work_set(&name, &trigger, transition)? {
+ enqueue(events, &name, oid)?;
+ }
+ }
+ Ok(())
+}
+
+fn enqueue(events: &dyn EventSink, effect: &str, oid: ObjectId) -> Result<()> {
+ events.enqueue(effect, oid)
+}
crates/ents-receive/src/sink.rs
@@ -1,0 +1,147 @@
+//! `EventSink`: the sole destination for post-receive matches
+//! (`receive.event-sink`) — null locally, a durable queue hosted.
+//!
+//! This module also carries the two reference implementations named by the
+//! development plan for this phase: [`NullEventSink`] (the null sink the
+//! phase-4 exit criterion runs against) and [`MemoryEventSink`] (an
+//! in-memory, deduplicating sink demonstrating `receive.dedup` and, paired
+//! with [`crate::reconcile`], `receive.reconstructible`).
+
+use std::collections::BTreeSet;
+use std::sync::{Mutex, MutexGuard, PoisonError};
+
+use gix_hash::ObjectId;
+
+use crate::error::Result;
+
+/// The sole destination for post-receive matches (`receive.event-sink`):
+/// `receive` enqueues one `(effect, oid)` obligation per commit that enters
+/// an effect's work set, and never evaluates the effect itself
+/// (`receive.never-blocks`).
+///
+/// # Errors
+///
+/// [`EventSink::enqueue`] fails only when the sink itself cannot durably
+/// record the obligation (queue I/O, hosted). It is never where an effect
+/// runs or where a verdict is judged.
+///
+/// # Examples
+///
+/// A minimal sink that just counts deliveries — enough to see that
+/// `receive` calls `enqueue` at all, without needing the full dedup
+/// bookkeeping [`MemoryEventSink`] provides.
+///
+/// ```
+/// use std::sync::atomic::{AtomicUsize, Ordering};
+///
+/// use ents_receive::EventSink;
+///
+/// #[derive(Default)]
+/// struct Counting(AtomicUsize);
+///
+/// impl EventSink for Counting {
+/// fn enqueue(&self, _effect: &str, _oid: gix_hash::ObjectId) -> ents_receive::Result<()> {
+/// self.0.fetch_add(1, Ordering::Relaxed);
+/// Ok(())
+/// }
+/// }
+///
+/// let sink = Counting::default();
+/// sink.enqueue("unit", gix_hash::ObjectId::null(gix_hash::Kind::Sha1))
+/// .expect("infallible sink");
+/// assert_eq!(sink.0.load(Ordering::Relaxed), 1);
+/// ```
+// @relation(receive.event-sink, receive.never-blocks, scope=file)
+pub trait EventSink: Send + Sync {
+ /// Enqueue re-evaluation of `effect` for `oid`.
+ ///
+ /// Redelivering the same `(effect, oid)` pair MUST be safe to call
+ /// again — the dedup key is exactly this pair (`receive.dedup`), so a
+ /// conforming sink either folds the duplicate itself ([`MemoryEventSink`]
+ /// does) or leaves de-duplication to whatever drains the queue, as long
+ /// as the eventual *outcome* is exactly-once.
+ ///
+ /// # Errors
+ ///
+ /// Only a genuine sink failure (durable-queue I/O); see the trait's
+ /// own doc.
+ fn enqueue(&self, effect: &str, oid: ObjectId) -> Result<()>;
+}
+
+/// The null `EventSink`: drops every obligation.
+///
+/// This is the local deployment's reference sink (`receive.event-sink`:
+/// "null locally") and the one the phase-4 exit criterion runs `receive`
+/// against — a local write path with no effect crate linked yet has nothing
+/// useful to enqueue into.
+///
+/// # Examples
+///
+/// ```
+/// use ents_receive::{EventSink, NullEventSink};
+///
+/// let sink = NullEventSink;
+/// sink.enqueue("unit", gix_hash::ObjectId::null(gix_hash::Kind::Sha1))
+/// .expect("the null sink never fails");
+/// ```
+// @relation(receive.event-sink, scope=file)
+#[derive(Debug, Clone, Copy, Default)]
+pub struct NullEventSink;
+
+impl EventSink for NullEventSink {
+ fn enqueue(&self, _effect: &str, _oid: ObjectId) -> Result<()> {
+ Ok(())
+ }
+}
+
+/// An in-memory, deduplicating `EventSink`: the reference implementation
+/// `receive.dedup` and `receive.reconstructible` describe.
+///
+/// Redelivering the same `(effect, oid)` pair is a no-op — the set, not a
+/// counter, is the state — which is what makes redelivery from an
+/// at-least-once queue yield exactly-once outcomes (`receive.dedup`). This
+/// type MAY lose its state on crash (it is exactly that: in-memory); the
+/// composition root is expected to call [`crate::reconcile`] against
+/// repository state at startup to rebuild it before serving further pushes,
+/// per `receive.reconstructible` — the durable queue this stands in for is a
+/// performance optimization, never a correctness requirement.
+///
+/// # Examples
+///
+/// ```
+/// use ents_receive::{EventSink, MemoryEventSink};
+///
+/// let sink = MemoryEventSink::default();
+/// let oid = gix_hash::ObjectId::null(gix_hash::Kind::Sha1);
+///
+/// sink.enqueue("unit", oid).expect("infallible sink");
+/// sink.enqueue("unit", oid).expect("redelivery is a no-op");
+///
+/// assert_eq!(sink.pending(), vec![("unit".to_owned(), oid)]);
+/// ```
+// @relation(receive.dedup, receive.reconstructible, scope=file)
+#[derive(Debug, Default)]
+pub struct MemoryEventSink {
+ pending: Mutex<BTreeSet<(String, ObjectId)>>,
+}
+
+impl MemoryEventSink {
+ /// Every distinct `(effect, oid)` obligation enqueued so far, in
+ /// sorted order.
+ #[must_use]
+ pub fn pending(&self) -> Vec<(String, ObjectId)> {
+ self.locked().iter().cloned().collect()
+ }
+
+ fn locked(&self) -> MutexGuard<'_, BTreeSet<(String, ObjectId)>> {
+ self.pending.lock().unwrap_or_else(PoisonError::into_inner)
+ }
+}
+
+impl EventSink for MemoryEventSink {
+ // @relation(receive.dedup, scope=function)
+ fn enqueue(&self, effect: &str, oid: ObjectId) -> Result<()> {
+ self.locked().insert((effect.to_owned(), oid));
+ Ok(())
+ }
+}
crates/ents-receive/tests/receive.rs
@@ -1,0 +1,405 @@
+//! Integration tests for `receive`: the mandatory/advisory gate-policy
+//! table (rstest — the spec's two named policies), redaction enforcement
+//! (admin-only push, ingest refusal), `(effect, oid)` dedup, and the
+//! reconstructibility proof that a boot-time [`reconcile`] rebuilds exactly
+//! the obligations incremental `receive` calls would have enqueued.
+
+#![expect(
+ clippy::expect_used,
+ reason = "integration test: fixtures panic on setup failure"
+)]
+
+use ents_gate::Config;
+use ents_model::{Effect, Issue, Provenance, Redaction, namespace, trailer::Trailers};
+use ents_receive::{
+ MemoryEventSink, Mode, NullEventSink, Proposal, RefTransition, TxResult, receive, reconcile,
+};
+use ents_testutil::{
+ CommitSpec, Keypair, MemRefStore, ObjectStore, enroll_member, write_commit, write_meta_entity,
+};
+use gix::refs::FullName;
+use gix_hash::ObjectId;
+use gix_object::{Kind, Write as _};
+use gix_ref_store::RefStoreRead as _;
+use rstest::rstest;
+
+const ADMIN_SEED: u8 = 1;
+const GUEST_SEED: u8 = 2;
+
+/// A forge fixture with verification in force: an admin-registered member
+/// `admin`, a self-attested member `guest`, and an epoch recorded in
+/// `refs/meta/config` — the same shape `ents-gate`'s own tests use.
+struct Forge {
+ refs: MemRefStore,
+ objects: ObjectStore,
+ admin: Keypair,
+ guest: Keypair,
+}
+
+fn forge() -> Forge {
+ let refs = MemRefStore::default();
+ let objects = ObjectStore::default();
+ let admin = Keypair::from_seed(ADMIN_SEED);
+ let guest = Keypair::from_seed(GUEST_SEED);
+ enroll_member(
+ &refs,
+ &objects,
+ "admin",
+ &admin,
+ Provenance::AdminRegistered,
+ 100,
+ );
+ enroll_member(
+ &refs,
+ &objects,
+ "guest",
+ &guest,
+ Provenance::SelfAttested,
+ 110,
+ );
+ let config_ref: FullName = namespace::CONFIG_REF.try_into().expect("valid");
+ write_meta_entity(
+ &refs,
+ &objects,
+ config_ref,
+ &Config { epoch: Some(200) },
+ Some(&admin),
+ 200,
+ );
+ Forge {
+ refs,
+ objects,
+ admin,
+ guest,
+ }
+}
+
+fn name(s: &str) -> FullName {
+ s.try_into().expect("valid refname in test")
+}
+
+/// Build a signed (or unsigned) mutation commit that binds itself to
+/// `refname` via the `Advance-ref:` trailer, *without* moving the ref —
+/// unlike `ents_testutil::write_meta_entity`, so the test can hand the
+/// result to `receive` and observe whether *it* moves the ref.
+fn build_mutation<T: for<'facet> facet::Facet<'facet>>(
+ objects: &ObjectStore,
+ refname: &FullName,
+ entity: &T,
+ signer: Option<&Keypair>,
+ seconds: i64,
+) -> ObjectId {
+ let tree = facet_git_tree::serialize_into(entity, objects).expect("serializes");
+ let trailers = Trailers {
+ ents_ref: Some(refname.clone()),
+ schema_version: None,
+ };
+ let message = format!("Mutate {}\n\n{}", refname.as_bstr(), trailers.render());
+ write_commit(
+ objects,
+ &CommitSpec {
+ tree,
+ parents: vec![],
+ message,
+ seconds,
+ },
+ signer,
+ )
+}
+
+fn single(transition: RefTransition, objects: Vec<ObjectId>) -> Proposal {
+ Proposal {
+ transitions: vec![transition],
+ objects,
+ auth: None,
+ }
+}
+
+// ---------------------------------------------------------------------
+// receive.unit, receive.shared-path, receive.refstore-seam: the gate
+// policy table — mandatory aborts the whole batch on a failing verdict,
+// advisory writes regardless and only annotates.
+// ---------------------------------------------------------------------
+
+#[rstest]
+#[case::mandatory_authorized(Mode::Mandatory, true, TxResult::Applied, true)]
+#[case::mandatory_unauthorized(Mode::Mandatory, false, TxResult::Refused, false)]
+#[case::advisory_authorized(Mode::Advisory, true, TxResult::Applied, true)]
+#[case::advisory_unauthorized(Mode::Advisory, false, TxResult::Applied, false)]
+// @relation(receive.unit, receive.shared-path, receive.refstore-seam, receive.object-access, receive.proposal-shape, gate.mandatory-hosted, gate.advisory-local, scope=function, role=Verifies)
+fn gate_policy_matches_mode(
+ #[case] mode: Mode,
+ #[case] authorized: bool,
+ #[case] expected: TxResult,
+ #[case] expect_verdict_pass: bool,
+) {
+ let forge = forge();
+ let refname = namespace::issue_ref("1").expect("valid");
+ let signer = if authorized {
+ &forge.admin
+ } else {
+ &forge.guest
+ };
+ let issue = Issue {
+ title: "t".into(),
+ body: "b".into(),
+ state: "open".into(),
+ assignees: vec![],
+ labels: vec![],
+ };
+ let tip = build_mutation(&forge.objects, &refname, &issue, Some(signer), 300);
+
+ let outcome = receive(
+ &forge.refs,
+ &forge.objects,
+ &NullEventSink,
+ &single(
+ RefTransition {
+ name: refname.clone(),
+ old: None,
+ new: Some(tip),
+ },
+ vec![tip],
+ ),
+ mode,
+ )
+ .expect("evaluates");
+
+ assert_eq!(outcome.result, expected);
+ assert_eq!(outcome.verdicts.len(), 1);
+ let (_, verdict) = outcome.verdicts.first().expect("exactly one transition");
+ assert_eq!(verdict.is_pass(), expect_verdict_pass);
+
+ let landed = forge
+ .refs
+ .get(refname.as_ref())
+ .expect("readable")
+ .is_some();
+ assert_eq!(landed, matches!(expected, TxResult::Applied));
+}
+
+// ---------------------------------------------------------------------
+// receive.redaction-admin-only: a push to refs/meta/redactions/* is
+// refused unless the pusher is admin-registered — a consequence of
+// composing receive with ents-gate's existing authorization arm, pinned
+// here at the receive level.
+// ---------------------------------------------------------------------
+
+#[rstest]
+#[case::admin_registered(true, TxResult::Applied)]
+#[case::self_attested(false, TxResult::Refused)]
+// @relation(receive.redaction-admin-only, scope=function, role=Verifies)
+fn redaction_ref_push_requires_admin(#[case] as_admin: bool, #[case] expected: TxResult) {
+ let forge = forge();
+ let refname = namespace::redaction_ref("r1").expect("valid");
+ let signer = if as_admin { &forge.admin } else { &forge.guest };
+ let redaction = Redaction::new(ObjectId::null(gix_hash::Kind::Sha1), "leaked credential");
+ let tip = build_mutation(&forge.objects, &refname, &redaction, Some(signer), 300);
+
+ let outcome = receive(
+ &forge.refs,
+ &forge.objects,
+ &NullEventSink,
+ &single(
+ RefTransition {
+ name: refname,
+ old: None,
+ new: Some(tip),
+ },
+ vec![tip],
+ ),
+ Mode::Mandatory,
+ )
+ .expect("evaluates");
+
+ assert_eq!(outcome.result, expected);
+}
+
+// ---------------------------------------------------------------------
+// receive.redaction-ingest: a redacted hole cannot be silently refilled
+// by re-pushing the same bytes.
+// ---------------------------------------------------------------------
+
+#[rstest]
+// @relation(receive.redaction-ingest, scope=function, role=Verifies)
+fn reintroducing_a_redacted_object_refuses_the_whole_batch() {
+ let forge = forge();
+
+ // A blob that was, at some point, the payload of a leaked credential.
+ let leaked = forge
+ .objects
+ .write_buf(Kind::Blob, b"super secret")
+ .expect("write");
+ let redaction_ref = namespace::redaction_ref("r1").expect("valid");
+ write_meta_entity(
+ &forge.refs,
+ &forge.objects,
+ redaction_ref,
+ &Redaction::new(leaked, "leaked credential"),
+ Some(&forge.admin),
+ 250,
+ );
+
+ // Someone tries to push it back in, as part of an ordinary issue
+ // mutation's object graph.
+ let issue_ref = namespace::issue_ref("1").expect("valid");
+ let issue = Issue {
+ title: "t".into(),
+ body: "b".into(),
+ state: "open".into(),
+ assignees: vec![],
+ labels: vec![],
+ };
+ let tip = build_mutation(&forge.objects, &issue_ref, &issue, Some(&forge.admin), 300);
+
+ let outcome = receive(
+ &forge.refs,
+ &forge.objects,
+ &NullEventSink,
+ &single(
+ RefTransition {
+ name: issue_ref.clone(),
+ old: None,
+ new: Some(tip),
+ },
+ vec![tip, leaked],
+ ),
+ Mode::Mandatory,
+ )
+ .expect("evaluates");
+
+ assert_eq!(outcome.result, TxResult::Redacted { oid: leaked });
+ assert!(
+ outcome.verdicts.is_empty(),
+ "refused before any verdict was evaluated"
+ );
+ assert!(
+ forge
+ .refs
+ .get(issue_ref.as_ref())
+ .expect("readable")
+ .is_none(),
+ "the whole batch must be refused, not just the redacted object"
+ );
+}
+
+// ---------------------------------------------------------------------
+// receive.dedup: redelivering the same (effect, oid) pair is a no-op.
+// ---------------------------------------------------------------------
+
+#[rstest]
+// @relation(receive.dedup, scope=function, role=Verifies)
+fn memory_sink_deduplicates_by_effect_and_oid() {
+ use ents_receive::EventSink as _;
+
+ let sink = MemoryEventSink::default();
+ let oid = ObjectId::null(gix_hash::Kind::Sha1);
+
+ sink.enqueue("unit", oid).expect("infallible");
+ sink.enqueue("unit", oid).expect("infallible");
+ sink.enqueue("integration", oid).expect("infallible");
+
+ assert_eq!(
+ sink.pending(),
+ vec![("integration".to_owned(), oid), ("unit".to_owned(), oid),]
+ );
+}
+
+// ---------------------------------------------------------------------
+// receive.reconstructible: the boot-time scan rebuilds exactly the
+// obligations incremental `receive` calls would have enqueued.
+// ---------------------------------------------------------------------
+
+/// Two chained empty-tree commits, built deterministically (fixed actor,
+/// fixed tree, fixed seconds) so two independent object stores produce
+/// byte-identical oids — letting path A and path B below compare `pending()`
+/// sets directly, oids included, without sharing any state.
+fn chain_commits(objects: &ObjectStore, count: usize, start_seconds: i64) -> Vec<ObjectId> {
+ let tree = ents_testutil::empty_tree(objects);
+ let mut parent = None;
+ let mut out = Vec::with_capacity(count);
+ for i in 0..count {
+ let seconds = start_seconds.saturating_add(i64::try_from(i).unwrap_or(i64::MAX));
+ let commit = write_commit(
+ objects,
+ &CommitSpec {
+ tree,
+ parents: parent.into_iter().collect(),
+ message: format!("commit {i} at {seconds}"),
+ seconds,
+ },
+ None,
+ );
+ parent = Some(commit);
+ out.push(commit);
+ }
+ out
+}
+
+#[rstest]
+// @relation(receive.reconstructible, receive.event-sink, receive.never-blocks, query.workset, scope=function, role=Verifies)
+fn reconcile_matches_incremental_delivery() {
+ let effect = Effect {
+ trigger: "rev(refs/heads/main)".to_owned(),
+ toolchains: vec![],
+ run: "true".to_owned(),
+ };
+ let effect_ref: FullName = "refs/meta/effects/unit".try_into().expect("valid");
+ let main = name("refs/heads/main");
+
+ // Path A: two `receive` calls, each advancing refs/heads/main by one
+ // commit — the only thing that ever moves this ref — incrementally
+ // enqueuing into a live sink.
+ let incremental = {
+ let refs = MemRefStore::default();
+ let objects = ObjectStore::default();
+ write_meta_entity(&refs, &objects, effect_ref.clone(), &effect, None, 50);
+ let sink = MemoryEventSink::default();
+
+ let mut old = None;
+ for commit in chain_commits(&objects, 2, 100) {
+ let outcome = receive(
+ &refs,
+ &objects,
+ &sink,
+ &single(
+ RefTransition {
+ name: main.clone(),
+ old,
+ new: Some(commit),
+ },
+ vec![],
+ ),
+ Mode::Advisory,
+ )
+ .expect("evaluates");
+ assert_eq!(outcome.result, TxResult::Applied);
+ old = Some(commit);
+ }
+ sink.pending()
+ };
+
+ // Path B: an independent store, seeded directly to the same final
+ // state (as if `refs/heads/main` had already advanced through two
+ // accepted pushes whose enqueues were lost — a crashed in-memory
+ // sink, say) — then reconstructed from repository state alone, with
+ // no incremental delivery at all.
+ let reconciled = {
+ let refs = MemRefStore::default();
+ let objects = ObjectStore::default();
+ write_meta_entity(&refs, &objects, effect_ref, &effect, None, 50);
+ let commits = chain_commits(&objects, 2, 100);
+ refs.set(main.as_ref(), *commits.last().expect("non-empty chain"));
+
+ let sink = MemoryEventSink::default();
+ reconcile(&refs, &objects, &sink).expect("reconciles");
+ sink.pending()
+ };
+
+ assert_eq!(incremental, reconciled);
+ assert_eq!(
+ incremental.len(),
+ 2,
+ "one obligation per commit that entered the trigger's set"
+ );
+}