🧪 test(coverage): drive the error arms no test reached (#2178)

* 🧪 test(coverage): walk every arm in one binary

llvm-cov scores a function by its best instantiation, not by the union
across them, so an arm exercised only in a different compilation of the
same crate still reports as uncovered. Two functions were split that way
and neither showed up in lcov, because the merged file view lcov renders
does count those arms as covered.

TransportError::terminal_reason had four variants with no case in
transport_error_contract, leaving peryx-ha's own test compilation short
of the arms another compilation happened to walk. logging_layer had the
mirror image: the sink loop drove Journald and Syslog while Stdout and
File were reached only from elsewhere, so no single compilation covered
the whole match.

Both fixes extend an existing parameterised test rather than adding a
new one. A new test with a new type would add another partial view and
make the reported number worse.

* 🧪 test(coverage): drive the report error arms

A sort comparator never runs on a collection of fewer than two elements,
and every cache-inspection fixture had exactly one index and one driver.
So neither ordering the report documents had ever executed: not the
longest-first attribution that decides which index a key belongs to, nor
the by-name ecosystem order that keeps two runs diffable. Both now have
a fixture with something to reorder.

The ecosystem case needs four ecosystems rather than two. The registry
is a hash map, so with a single other ecosystem an unsorted report comes
out in the right order half the time and the test would pass against a
missing sort.

The OCI metadata check reported success on two failures it could not
survive: a manifest scan that cannot read the store, and a finding that
cannot be written out. Neither arm had a test because neither redb nor a
Vec writer fails on its own, so both go through a seam - the storage
backend fault for the read, a writer that refuses for the report.

* 🧪 test(coverage): pin the blob scan failure arm

A descriptor check reads the blob store to decide whether a layer is
present, and that read can fail for reasons that are not absence. The
arm that tells those apart had no test, because a filesystem blob store
does not fail on its own once its root exists.

Rooting the store at a regular file makes every path under it fail as
not a directory, which is a plain io error rather than a missing file.
The test asserts nothing was written as well as that the call failed:
the finding the check must not produce here is "missing blob", and an
empty report is what distinguishes a surfaced failure from a guess.

* 🧪 test(coverage): pin the tag scan failure arm

The tag scan is the second read the metadata check makes, so injecting a
storage failure at the first opportunity never reaches it - the manifest
scan consumes the injection. Sweeping the injection point finds the run
that fails during the tags instead.

Both reads fail with the same text, so the error cannot say which one
failed and the report has to. An empty report means the manifest scan
failed before writing anything; a report carrying the manifest findings
means those were written and the tag scan then failed.

A store handle does not survive its own injected failure, so each sweep
step reopens the retained pages rather than reusing one handle. Reusing
it leaves every step after the first failing in the manifest scan, which
looks exactly like "no injection point reaches the tags".

* 🧪 test(coverage): reject a partial summary count

The row count is assembled from two prefix scans, and neither scan's
failure arm had ever run, so nothing established what a failure partway
through returns. A count that comes back short is the dangerous shape
here: it reads as a real answer, and an operator cannot tell it from a
store that genuinely holds fewer rows.

Sweeping the injection point across both scans asserts the property
rather than one arm - every point either returns the whole count or an
error, and none returns a smaller one. The sweep also reaches both
failure arms, which one fixed injection point cannot.

pypi gains the fault-injection dev-dependencies the OCI crate already
carries. Resolver 3 keeps dev-dependency features out of the normal
build, so this stays test-only.

* 🧪 test(coverage): reject partial audit and repair results

The audit assembles its defect list from four scans and none of their
failure arms had run, so nothing established what a failure partway
through returns. A short list of defects is a false all-clear in the
same way a short count is a wrong answer: an operator reads two defects
as the whole truth and stops looking.

The repair has the matching shape one level down. Its write either lands
or it does not, and a store left with some rows rebuilt and some not
reports as a smaller set of defects on the next audit, which reads as
progress rather than as a repair that failed. The sweep asserts the
store settles at either its original defects or none, never between.

Both drive the property rather than the arm, which is also what reaches
every scan: a single fixed injection point can only fail one of them.

* 🧪 test(coverage): reject a short problem count

The metadata check sums problems across nine scans and the failure arm
of five of them had never run, so nothing established what a failure
partway through reports. A short problem count is the same false
all-clear as a short defect list: an operator reads "three problems" as
the whole truth and fixes three.

The sweep asserts the count and the report together, so a scan that
fails after writing part of its findings cannot pass by arriving at the
right number with the wrong lines. Sweeping is also what reaches all
five arms, since one injection point can only fail one scan.

* 🧪 test(coverage): drive the cache blob scan failure

Both cache reports end by scanning the blob store, and neither scan's
failure arm had run. The existing failure cases all come from the driver
seam, which returns before either report reaches its blob scan.

Rooting the store at a regular file does not work here, because a scan
skips a digest directory that does not exist and reports no blobs - the
seam would be silently ignored and the request would answer 200 with an
empty blob list. A regular file where the digest directory belongs is a
root that exists and cannot be read, which is what makes the scan fail.

The cases extend the existing parameterised failure test rather than
adding two of their own, so the failure seam stays in one place.

* 🧪 test(coverage): order the repair report by ecosystem

The repair report sorts its ecosystems because the driver registry is a
hash map, but every fixture registered one repairable ecosystem, so the
comparator never ran and the ordering it promises had never been checked.

Four ecosystems rather than two. With one other ecosystem an unsorted
report comes out in the right order half the time, so the test would
agree with a missing sort as often as not; four takes that to one in
twenty-four. The assertion is the whole report rather than a property of
it, so a wrong order fails on the line that is wrong.

The driver double gains the repair capability it lacked, which keeps the
test beside the driver set it exercises rather than rebuilding a plugin
registry to reach the same two calls.

* 🧪 test(coverage): reject a short journal page

The journal read propagates a store failure but nothing had ever driven
that arm, so nothing established what a failure partway through a page
returns. A short page is the damaging shape: a reader treats what it
gets as everything after its cursor, so entries dropped by a failed read
are never asked for again and the reader believes it is caught up.

The sweep asserts the whole snapshot rather than the entry count, since
a page that ends early still reports the serial it reached and a
count-only check would accept a page with the wrong cursor.

* 🧪 test(coverage): reject a short prefix scan

Both driver prefix scans propagate a store failure and neither arm had
run, so nothing established what a failure partway through a scan hands
back. The rows already gathered are the damaging answer: a caller that
picked a generation from this snapshot reads the short list as the whole
of it and works from a prefix it never finished reading.

The bounded and unbounded scans get one sweep each rather than one
covering both, since a handle does not survive its own injected failure
and the second call would run against a store the first already broke.

* 🧪 test(coverage): reject a short expired-upload scan

The reaper deletes whatever the expired-upload scan returns, so a scan
that fails partway and hands back the rows it had gathered leaves the
sessions it skipped holding their reserved bytes, with nothing arranged
to look at them again. The sweep asserts the scan returns the whole list
or an error.

A stored manifest checkpoint that cannot be decoded gets the matching
treatment one layer up. The decode is read to answer whether a request
already committed, so treating a corrupt record as absent would let that
request run a second time. The round trip is asserted beside it as the
control, since a decode that rejected everything would pass without it.

* 🧪 test(coverage): pin two blob store failure arms

A usage figure assembled from a walk that stopped early is not a usage
figure. Reporting the bytes it reached would understate what is on disk,
and a caller sizing a sweep from that number frees less than it believed
it would. The sweep already had this test for an unreadable store; the
usage walk beside it did not.

The multipart size check runs before the first request, so an object
past the protocol limit is refused against a bucket nobody contacted.
That refusal has to name the backend and the operation, or an operator
reads a bare size complaint with nothing to attach it to. The assertion
is the whole rendered message rather than the error kind, since the kind
alone is identical whether the context survived or was dropped.

* 🧪 test(coverage): match the source a routed 304 names

A routed upstream names which configured source answered, and only a
generation that same source published may be reused on a 304. Without
the match a source swapped in behind one index would have its "not
modified" applied to files a different source had published.

Every existing sync test drives a plain client, which leaves the
answering source absent and short-circuits the comparison before it
runs. Naming a source at all takes a router, so the test builds one.

The page refresh path has a sibling test for a 304 from a source that
never answered. This is the generation path, which reaches the same
question through different code.

* 🧪 test(pypi): serve a cached page through a rate limit

A rate-limited refresh is about the refresh, not about the copy peryx
already holds, so a page still inside its stale bound is served rather
than the failure reaching the caller. Every other rate-limit test runs
with no cached page at all, which settles the question before the bound
is ever read.

This does not close the branch that weighs the bound against a 429
directly. Measurement puts that line still uncovered: the client retries
a 429 and surfaces it as an upstream failure, so the arm matching an
Ok response with that status is reached another way, if at all.

* 🧪 test(pypi): weigh a cached page against a rate limit

A rate-limited refresh is about the refresh, not about the copy peryx
already holds. A page still inside its stale bound is served; once it
ages past the bound there is nothing to fall back on and the rate limit
reaches the caller as a rate limit, which says when to come back, rather
than as a gateway failure, which says nothing.

Three details decide which arm answers, and getting any of them wrong
lands on a neighbour that returns the same shape of failure. The
fixture is built here rather than through the stale-page helper, which
mounts a 503 on this same path and would be matched instead. The
retry-after has to exceed what the client will honour, or the response
is retried rather than handed back. And the request asks for HTML,
since the JSON path tries the stream first and logs its failure before
falling through.

* 🧪 test(pypi): keep a purge from preserving too little

A purge decides what to keep from a scan of the project's file rows, so
a scan that fails partway must not hand back a smaller preserved set.
Preserving less means removing more, and the rows it would take are the
ones a project populated by a catalog sync needs for a cold download.

The sweep runs as a dry run, so the store is identical at every
injection point and any difference in the report comes from the scan
rather than from what an earlier step deleted.

The stale-bound refusal beside it now names the status it expects. It
asserted only that the answer was not OK, which several arms satisfy,
and that looseness already sent one reading of this file wrong.

* 🧪 test(oci): commit no manifest without its checkpoint

The checkpoint is what a retry reads to decide the work is already
done, so a manifest committed without one is published again on the
next attempt as though it had never landed. Nothing had driven the
failure arm of that write, and every existing publish test passes no
operation at all, which skips the checkpoint entirely.

Each sweep step gets its own backend, because a publish mutates and the
next injection point would otherwise run against whatever the previous
one left behind. The fault is armed after the store opens so the count
applies to the publish rather than to creating the tables.

* 🧪 test(oci): refuse an oversized referrers document

A referrers document past the manifest bound is refused by size rather
than read in, so an upstream cannot spend peryx's memory on a body it
would reject anyway. The bound had no test on this path.

The assertion names the reason as well as the status. A failing proxy
answers with the same status from the arm beside this one, so a status
on its own would not tell a refused size from a refused upstream.

* 🧪 test(oci): leave no torn tag when a delete fails

A tag delete that cannot finish must leave the mapping either as it was
or gone, never pointing somewhere neither the write nor the delete put
it. Nothing had driven the failure arm of the trash write.

The first shape of this test asserted that a failed delete keeps the
tag, and that is not the guarantee: the delete is a soft one, so a
failure part way through can legitimately have trashed the tag already
and still answer 502. The assertion that holds is the narrower one
about what a pull could resolve afterwards.

* 🧪 test(oci): leave no torn tag when a restore fails

The mirror of the delete guarantee. A restore that cannot finish must
leave the tag either still trashed or pointing back at exactly what it
pointed at before, never at a digest neither the original write nor the
restore put there.

The delete that sets the tag up runs unfaulted, so the injection lands
on the restore rather than on the trashing that precedes it.

* 🧪 test(coverage): drive the arms these tests left behind

The gate measures test files too, and three of the harnesses added here
carried arms nothing ran. A coverage push that adds uncovered test code
moves the number the wrong way while looking like progress.

The repair report is now asserted for a rebuild as well as a preview,
which reaches the double's repair path and checks the ordering holds on
both. The report writer is a zero-capacity cursor rather than a type of
our own, since a hand-rolled one carries a flush this path never calls.

The publish sweep needed more than its dead branch removing. Every step
was failing because nothing had claimed the operation, so the injected
fault decided nothing and the success arm could not run. Claiming it
first makes the sweep exercise what it describes, and a control publish
proves the range contains a success at all. The invariant it asserts is
now the one the name claims: the manifest and its checkpoint are both
there or neither is, which holds at every injection point.

* 🧪 test(coverage): pin three worker and request failures

An export whose driver comes apart takes the worker with it, so nothing
sends the summary the request waits on. That wait has to end as an
interruption rather than hanging on a sender that will never write. A
driver returning without a snapshot does not reach it: the export
validates that itself and forwards a store error through the channel.

A driver that opens a second snapshot has already sent the first, so the
request is answered and the violation reaches the caller through the
body already streaming. Both doubles existed and were only ever driven
through the collect path, never through the export.

A request that cannot be built never reached a server, so it carries no
status; reporting one would attribute the failure to a response nobody
received. A scratch directory that cannot be created names the path it
failed on, since the profile write that follows depends on it.

* 🧪 test(peryx): name the backup config a restore cannot read

A restore reads the backup config before it verifies anything, so a
config that is not text is the first thing an operator hears about a
damaged backup. The failure has to name the file it was reading rather
than surfacing a bare encoding error with nothing attached to it.

The verify path reaches the same read, but only for a config whose hash
already matched, which bytes that fail to decode cannot do. The restore
path has no such check in front of it.

* 🧪 test(driver): tell the worker its request is gone

A client that abandons an export leaves nothing waiting for the summary,
so the send that opens the snapshot has nowhere to go. The worker has to
learn the request went rather than carry on producing a body no one will
read.

The order is forced rather than raced. The driver parks inside the plan
and says so, the test drops the request and waits for that drop to
finish, and only then is the driver released to send. Nothing here
depends on which thread wins.

Author: gaborbernat

crates/peryx-bench/tests/unit/tests.rs

@@ -923,3 +923,24 @@ fn metadata_start_failure_is_reported() {
         "cargo did not start"
     );
 }
+
+/// A scratch directory that cannot be created has to name the path it failed on. The profile write
+/// that follows depends on it, and a bare io error would leave whoever reads the failure guessing
+/// which directory the benchmark wanted.
+#[tokio::test]
+async fn machine_profile_publish_names_the_scratch_it_cannot_create() {
+    let dir = tempfile::tempdir().unwrap();
+    let blocked = dir.path().join("occupied");
+    std::fs::write(&blocked, b"not a directory").unwrap();
+    let scratch = blocked.join("scratch");
+
+    let error = MachineProfile::system(dir.path())
+        .publish(Some(&scratch))
+        .await
+        .unwrap_err();
+
+    assert_eq!(
+        format!("{error}"),
+        format!("cannot create benchmark scratch at {}", scratch.display())
+    );
+}

crates/peryx-driver/tests/driver_set_contract.rs

@@ -6,8 +6,8 @@ use async_trait::async_trait;
 use peryx_core::Ecosystem;
 use peryx_driver::serving::{
     BlobReferenceDriver, BrowseDriver, BrowseError, BrowseRequest, CacheDriver, CacheInspectDriver,
-    CapabilityRegistrar, EcosystemDriver, FsckDriver, ImportDriver, JobConfig, JobDriver, MetricsDriver, NameDriver,
-    RetentionDriver, TrashDriver,
+    CapabilityRegistrar, EcosystemDriver, FsckDriver, ImportDriver, JobConfig, JobDriver, MetadataRepairDriver,
+    MetricsDriver, NameDriver, RetentionDriver, TrashDriver,
 };
 use peryx_driver::serving::{CachePage, PurgeReport};
 use peryx_driver::{BlobReferenceScan, BlobReferenceScanError, DriverSet};
@@ -72,6 +72,28 @@ impl FsckDriver for Driver {
     }
 }
 
+impl MetadataRepairDriver for Driver {
+    fn preview_metadata_repair(
+        &self,
+        _: &peryx_storage::meta::MetaStore,
+        _: &[peryx_index::Index],
+        out: &mut dyn std::io::Write,
+    ) -> Result<u64, String> {
+        writeln!(out, "metadata\t{}\twould rebuild", self.ecosystem.as_str()).map_err(|error| error.to_string())?;
+        Ok(1)
+    }
+
+    fn repair_metadata(
+        &self,
+        _: &peryx_storage::meta::MetaStore,
+        _: &[peryx_index::Index],
+        out: &mut dyn std::io::Write,
+    ) -> Result<u64, String> {
+        writeln!(out, "metadata\t{}\trebuilt", self.ecosystem.as_str()).map_err(|error| error.to_string())?;
+        Ok(1)
+    }
+}
+
 impl RetentionDriver for Driver {
     fn validate_retention(&self, _policy: &peryx_policy::RetentionPolicy) -> Result<(), String> {
         Ok(())
@@ -461,3 +483,35 @@ async fn driver_set_dispatches_browse_capability() {
         Err(BrowseError::Internal("browse".to_owned()))
     );
 }
+
+/// The driver registry is a hash map, so the repair report needs an order of its own or two runs
+/// list the ecosystems differently and an operator diffing them reads the swap as change.
+///
+/// Four ecosystems rather than two: with one other ecosystem an unsorted report lands in the right
+/// order half the time by luck, and the test would pass against a missing sort.
+#[rstest::rstest]
+#[case::preview(false, "would rebuild", "planned")]
+#[case::apply(true, "rebuilt", "repaired")]
+fn cache_repair_reports_ecosystems_in_name_order(#[case] apply: bool, #[case] verb: &str, #[case] label: &str) {
+    let mut set = DriverSet::default();
+    for name in ["zulu", "mike", "core", "alpha"] {
+        let ecosystem = Ecosystem::new(name);
+        set.register_metadata_repair(ecosystem.clone(), Arc::new(Driver { ecosystem }));
+    }
+    let directory = tempfile::tempdir().unwrap();
+    let meta = peryx_storage::meta::MetaStore::open(directory.path().join("peryx.redb")).unwrap();
+    let mut out = Vec::new();
+
+    peryx_driver::cache_inspection::write_cache_repair(&set, &meta, &[], apply, &mut out).unwrap();
+
+    assert_eq!(
+        String::from_utf8(out).unwrap(),
+        format!(
+            "metadata\talpha\t{verb}\n\
+             metadata\tcore\t{verb}\n\
+             metadata\tmike\t{verb}\n\
+             metadata\tzulu\t{verb}\n\
+             {label}\t4\n"
+        )
+    );
+}

crates/peryx-driver/tests/unit/retention/tests.rs

@@ -655,3 +655,135 @@ fn test_plan_error_display_names_each_failure() {
             .contains("boom")
     );
 }
+
+struct PanickingDriver;
+
+impl RetentionDriver for PanickingDriver {
+    fn validate_retention(&self, _policy: &RetentionPolicy) -> Result<(), String> {
+        Ok(())
+    }
+
+    fn plan_retention(
+        &self,
+        _scan: &crate::serving::RetentionScan<'_>,
+        _start: &mut dyn FnMut(peryx_policy::RetentionSummary) -> Result<(), String>,
+        _emit: &mut dyn FnMut(RetentionDecision) -> Result<(), String>,
+    ) -> Result<(), String> {
+        panic!("the plugin driver came apart");
+    }
+}
+
+/// A driver that comes apart takes the worker with it, so nothing ever sends the summary the request
+/// is waiting on. The wait has to end as an interruption rather than hanging on a sender that will
+/// never write. A driver returning without a snapshot does not reach this: the export validates that
+/// itself and forwards a store error through the channel.
+#[tokio::test]
+async fn test_export_body_ends_the_wait_when_the_worker_dies() {
+    let (_dir, meta) = store();
+    let driver: Arc<dyn RetentionDriver> = Arc::new(PanickingDriver);
+    let gates = super::RetentionGates::new(1);
+    let permit = gates.try_enter("alpha").unwrap();
+    let export = super::RetentionExport {
+        index: "alpha".to_owned(),
+        ecosystem: "example".to_owned(),
+        policy: empty_policy(),
+        now: None,
+        after: 0,
+        expect: None,
+    };
+
+    let error = super::export_body(driver, meta, export, permit).await.unwrap_err();
+
+    assert!(format!("{error:?}").contains("export worker stopped"), "{error:?}");
+}
+
+/// A driver that opens a second snapshot has already sent the first, so the request is answered and
+/// the violation has to reach the caller through the body it is already streaming.
+#[tokio::test]
+async fn test_export_body_reports_a_second_snapshot_through_the_stream() {
+    let (_dir, meta) = store();
+    let driver: Arc<dyn RetentionDriver> = Arc::new(InvalidDriver(SnapshotViolation::Repeated));
+    let gates = super::RetentionGates::new(1);
+    let permit = gates.try_enter("alpha").unwrap();
+    let export = super::RetentionExport {
+        index: "alpha".to_owned(),
+        ecosystem: "example".to_owned(),
+        policy: empty_policy(),
+        now: None,
+        after: 0,
+        expect: None,
+    };
+
+    let (_, body) = super::export_body(driver, meta, export, permit).await.unwrap();
+    let error = axum::body::to_bytes(body, usize::MAX).await.unwrap_err();
+
+    assert!(format!("{error:?}").contains("more than one snapshot"), "{error:?}");
+}
+
+/// Parks inside the plan until the test releases it, so the request can be abandoned first and the
+/// summary send lands on a receiver that is already gone.
+struct GatedDriver {
+    entered: tokio::sync::mpsc::UnboundedSender<()>,
+    release: std::sync::Mutex<Option<std::sync::mpsc::Receiver<()>>>,
+    saw: tokio::sync::mpsc::UnboundedSender<String>,
+}
+
+impl RetentionDriver for GatedDriver {
+    fn validate_retention(&self, _policy: &RetentionPolicy) -> Result<(), String> {
+        Ok(())
+    }
+
+    fn plan_retention(
+        &self,
+        scan: &crate::serving::RetentionScan<'_>,
+        start: &mut dyn FnMut(peryx_policy::RetentionSummary) -> Result<(), String>,
+        _emit: &mut dyn FnMut(RetentionDecision) -> Result<(), String>,
+    ) -> Result<(), String> {
+        self.entered.send(()).unwrap();
+        // Blocking is correct here: the plan already runs on the blocking pool.
+        let release = self.release.lock().unwrap().take().expect("released once");
+        release.recv().unwrap();
+        let summary = peryx_policy::RetentionSummary {
+            policy_version: scan.policy.version(),
+            frontier: peryx_policy::RetentionFrontier::default(),
+        };
+        start(summary).inspect_err(|reason| self.saw.send(reason.clone()).unwrap())
+    }
+}
+
+/// A client that abandons an export leaves nothing waiting for the summary, so the send that opens
+/// the snapshot has nowhere to go. The worker has to learn the request is gone rather than carry on
+/// producing a body no one will read.
+///
+/// The order is forced rather than raced: the driver parks inside the plan, the test drops the
+/// request and waits for that drop to complete, and only then is the driver released.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_export_body_tells_the_worker_when_the_request_is_gone() {
+    let (_dir, meta) = store();
+    let (entered, mut entered_rx) = tokio::sync::mpsc::unbounded_channel();
+    let (saw, mut saw_rx) = tokio::sync::mpsc::unbounded_channel();
+    let (release, release_rx) = std::sync::mpsc::channel();
+    let driver: Arc<dyn RetentionDriver> = Arc::new(GatedDriver {
+        entered,
+        release: std::sync::Mutex::new(Some(release_rx)),
+        saw,
+    });
+    let gates = super::RetentionGates::new(1);
+    let permit = gates.try_enter("alpha").unwrap();
+    let export = super::RetentionExport {
+        index: "alpha".to_owned(),
+        ecosystem: "example".to_owned(),
+        policy: empty_policy(),
+        now: None,
+        after: 0,
+        expect: None,
+    };
+    let request = tokio::spawn(async move { super::export_body(driver, meta, export, permit).await.map(|_| ()) });
+
+    entered_rx.recv().await.expect("the plan parked");
+    request.abort();
+    assert!(request.await.unwrap_err().is_cancelled());
+    release.send(()).unwrap();
+
+    assert_eq!(saw_rx.recv().await.as_deref(), Some("export request gone"));
+}

crates/peryx-ecosystem-oci/tests/unit/quota/tests.rs

@@ -1,8 +1,8 @@
 use peryx_storage::meta::{MetaStore, QuotaLimit, QuotaLimits, QuotaReservationState, QuotaUsage};
 
 use super::{
-    ManifestCommit, ReserveOutcome, commit_blob_membership, finalize, publish_manifest, quota_reservation,
-    release_blob_membership, reserve,
+    ManifestCheckpoint, ManifestCommit, ManifestOperation, ReserveOutcome, commit_blob_membership, finalize,
+    publish_manifest, quota_reservation, release_blob_membership, reserve,
 };
 use crate::name::Reference;
 use crate::registry::ServeError;
@@ -246,3 +246,100 @@ fn test_manifest_tag_replacement_commits_a_new_allocation() {
         (8, 1)
     );
 }
+
+/// A stored checkpoint that cannot be decoded has to fail the replay it was read for. Treating it as
+/// absent would let a request that already committed run a second time.
+///
+/// The round trip is the control: without it a `decode` that rejected everything would pass.
+#[test]
+fn test_manifest_checkpoint_reports_a_record_it_cannot_decode() {
+    let checkpoint = ManifestCheckpoint {
+        reference: "sha256:aa".to_owned(),
+        epoch: 3,
+        serial: 9,
+    };
+
+    assert_eq!(ManifestCheckpoint::decode(&checkpoint.encode()).unwrap(), checkpoint);
+    assert!(matches!(
+        ManifestCheckpoint::decode(b"not a checkpoint"),
+        Err(ServeError::Transport(_))
+    ));
+}
+
+/// A publish that cannot record its idempotency checkpoint must not leave the manifest behind. The
+/// checkpoint is what a retry reads to decide the work is already done, so a manifest committed
+/// without one is published again on the next attempt as though it had never landed.
+///
+/// Each step gets its own backend, because a publish mutates and the next injection point would
+/// otherwise run against whatever the previous one left. The fault is armed after the store opens
+/// so the count applies to the publish rather than to creating the tables.
+#[test]
+fn test_publish_manifest_never_commits_without_its_checkpoint() {
+    let manifest = Manifest {
+        media_type: "application/vnd.oci.image.manifest.v1+json".to_owned(),
+        bytes: b"{}".to_vec(),
+    };
+    let reference = Reference::Tag("stable".to_owned());
+    // The control: with nothing injected the publish lands and the manifest is readable. Without it
+    // the sweep below would pass against a publish that always failed.
+    let (pages, fault) = peryx_test_support::fault::backend();
+    let meta = MetaStore::open_backend(peryx_test_support::fault::faulted(&pages, &fault)).unwrap();
+    meta.claim_operation("op-1", None, 100).unwrap();
+    publish_manifest(&meta, checkpoint_commit(&manifest, &reference)).unwrap();
+    drop(meta);
+    let meta = MetaStore::reopen_backend(peryx_test_support::fault::faulted(&pages, &fault)).unwrap();
+    assert!(meta.get_driver_value("oci\u{0}m\u{0}sha256:a").unwrap().is_some());
+    drop(meta);
+
+    let mut failed = 0_u32;
+    for fail_after in 0..192 {
+        let (pages, fault) = peryx_test_support::fault::backend();
+        let meta = MetaStore::open_backend(peryx_test_support::fault::faulted(&pages, &fault)).unwrap();
+        // The checkpoint refuses an operation nobody claimed, so without this every step fails for
+        // that reason instead of the injected one and the sweep proves nothing.
+        meta.claim_operation("op-1", None, 100).unwrap();
+        fault.arm(fail_after);
+        let published = publish_manifest(&meta, checkpoint_commit(&manifest, &reference));
+        fault.disable();
+        drop(meta);
+
+        let meta = MetaStore::reopen_backend(peryx_test_support::fault::faulted(&pages, &fault)).unwrap();
+        let stored = meta.get_driver_value("oci\u{0}m\u{0}sha256:a").unwrap().is_some();
+        // Empty while pending, so a recorded response is what says the checkpoint landed. Presence
+        // of the record alone would only prove the claim above ran.
+        let checkpointed = meta
+            .operation_outcome("op-1")
+            .unwrap()
+            .is_some_and(|record| !record.response.is_empty());
+        failed += u32::from(published.is_err());
+        assert_eq!(
+            stored,
+            checkpointed,
+            "injecting after {fail_after} reads left manifest={stored} and checkpoint={checkpointed} \
+             disagreeing, published_ok={}",
+            published.is_ok()
+        );
+    }
+
+    assert!(failed > 0, "no injection point reached the publish");
+}
+
+fn checkpoint_commit<'a>(manifest: &'a Manifest, reference: &'a Reference) -> ManifestCommit<'a> {
+    ManifestCommit {
+        index: "store",
+        repo: "app",
+        canonical: "sha256:a",
+        manifest,
+        reference,
+        referrer: None,
+        reservation: None,
+        journal: true,
+        webhook: None,
+        operation: Some(ManifestOperation {
+            id: "op-1",
+            reference: "stable",
+            epoch: 1,
+            now: 100,
+        }),
+    }
+}