Merge watchdog idunn-topology-lock: no tick waits on another uid's lock

Eureka idunn-watchdog cut idunn-topology-lock, attempts 1-2. Soul verdict
idunn-watchdog:verdict:cut-idunn-topology-lock.s2 holds. Topology and
incident writes share one private-lock exchange that replaces a non-private
lock by rename; a contended publish stays in phase and retries. Measured
tick with a held lock: 0.09 s (was 5.07 s).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

deploy/idunn-yggdrasil.service

@@ -77,7 +77,10 @@ ReadWritePaths=/etc/ufw
 # idunn-projection is deliberately outside Idunn's own 0750 directory: the
 # topology store is Idunn's *published* surface, and every managed target must
 # read it to verify its own Expected incarnation against the Idunn anchor. It
-# was inside the private root, where no workload could traverse to it.
+# was inside the private root, where no workload could traverse to it. The
+# published topology store is read lock-free (Odin) and incidents.cc is read by
+# Bifrost's reader; managed targets get Expected from their runtime bundle. Only
+# the stores are readable: each .lock beside them is Idunn's alone, 0600.
 ReadWritePaths=/var/lib/gamecult/idunn /var/lib/gamecult/idunn-authority /var/lib/gamecult/idunn-projection
 ReadWritePaths=/srv/build
 # Each managed target gets only its bound roots. A target Idunn cannot write is

docs/migration.md

@@ -520,7 +520,7 @@ started, ran, and then died on a file it could not open:
 | `runtime_root` and every directory above it | traversable by the workload: `o+x`, or group-owned by the state group with `g+x` | the workload is a `DynamicUser` and owns nothing, and `ReadOnlyPaths=` binds the bundle's leaf into the namespace without granting traversal above it. Idunn now checks the whole chain and names the closed directory |
 | `runtime_root` | setgid, group-owned by the state group (`2750 root:<state_group>`) | `harden_root_authority_file` requires the write-lease record's group to equal its parent's, and only setgid makes a newly created record inherit it. Without it the record lands `root:root` and the workload cannot read its own lease |
 | the write-lease record | `root:<state_group>`, mode `0640` | follows from the setgid parent; the workload reads it through the state group. Idunn writes it, so fixing an existing wrong-group record means deleting it, and that is safe only when no transaction is live |
-| `/var/lib/gamecult/idunn-projection` | default ACL `u::rw-,g::r--,o::r--` | Idunn runs with `UMask=027`, so each publish creates the store and its `.lock` at `0640`. The code re-opens them after every write, but that leaves a window; a default ACL sets the mode at creation and closes it |
+| `/var/lib/gamecult/idunn-projection` | default ACL `u::rw-,g::r--,o::r--` | Idunn runs with `UMask=027`, so a store is created `0640` and each publish re-opens it `0644`; the default ACL sets the mode at creation and closes that window for stores. The `.lock` is not published and no reader opens it: Idunn creates it `0600` (replacing one found wider, so a descriptor another uid already holds is orphaned), and the ACL never decides its mode |
 | `/etc/nginx/idunn-stream-routes` | `root:root`, not group-writable | "route authority parent is not canonical root-owned and nonwritable". It was `root:idunn 0775` from the generation where Idunn ran unprivileged and needed to write it as a group member |
 
 Two shapes are worth stating plainly because they are easy to get backwards.

src/control_plane.rs

@@ -40,7 +40,7 @@ use crate::deployment_plan::{
 use crate::drivers::{
     ChallengeFailure, CultCacheTopologyDriver, CultCacheWriteLeaseDriver, DockerRunnerDriver, FrozenSourceReceipt,
     GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver,
-    ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators,
+    LockContended, ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators,
     RouteObservation, RoutePreflightReceipt, RunnerPort, SourcePort,
     SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort,
     WriteLeasePort,
@@ -5576,7 +5576,14 @@ impl Engine {
                 // Before the fence an abort that cannot finish a step is
                 // resumable, never a post-fence abort: that path refuses a
                 // pre-fence phase, so choosing it only wedged the tick.
-                if latest.value.pre_fencing_abort.is_none() {
+                //
+                // A publish that lost to a held lock is not a failed step: the
+                // holder lets go, so the transaction stays in its phase with
+                // the error recorded and the backoff above retries it. The
+                // phase deadline, not an abort, bounds the wait.
+                if latest.value.pre_fencing_abort.is_none()
+                    && error.downcast_ref::<LockContended>().is_none()
+                {
                     self.begin_pre_fencing_abort(latest, error)?;
                 } else {
                     self.record_resumable_error(latest, &error)?;
@@ -15496,7 +15503,7 @@ mod tests {
         Ok(())
     }
 
-    #[cfg(unix)]
+    #[cfg(target_os = "linux")]
     #[test]
     fn a_held_incident_lock_changes_nothing_continuity_decides() -> Result<()> {
         use crate::incident::tests::{STUCK_AFTER, hold_lock_of, run_while_held};
@@ -15510,7 +15517,8 @@ mod tests {
         let world = EngineFixture::with_workload(workload.clone())?;
         admit_incumbent(&world)?;
         let path = incident_path(&world);
-        let actual = run_while_held(hold_lock_of(&path), || {
+        // Three ticks reach the open of an exhausted target, one exchange each.
+        let actual = run_while_held(hold_lock_of(&path), 3, || {
             continuity_decisions(&world, &world.engine, &workload)
         })?;
         assert_eq!(actual, expected, "a held incident lock changed a continuity decision");
@@ -15540,6 +15548,171 @@ mod tests {
         assert_control_store_holds_no_incident(&world)
     }
 
+    /// A publish that could not land is not done, at the tick: the demotion
+    /// continuity needs before a restart is one exchange on the topology lock,
+    /// and another holder of it defers the restart instead of waiting for it.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_held_topology_lock_defers_continuity_and_mints_nothing() -> Result<()> {
+        use crate::incident::tests::{hold_lock_of, run_while_held};
+
+        let workload = SwitchWorkload::new();
+        let world = EngineFixture::with_workload(workload.clone())?;
+        let incumbent = admit_incumbent(&world)?;
+        project_admitted(&world, &incumbent)?;
+        let store = world.engine.options.topology_store.clone();
+        let projected = projected_under(&world, &incumbent.expected)?;
+        assert!(projected.contains(&IdunnRuntimeActivationRecord::TYPE.to_owned()));
+        let before = std::fs::read(&store)?;
+        workload.kill();
+
+        // One exchange: the demotion. The window is not spent, so continuity
+        // would restart if the projection could be demoted.
+        let moved = run_while_held(hold_lock_of(&store), 1, || {
+            world.engine.supervise_one_admitted_generation()
+        })?;
+        assert!(moved, "the deferral is a write the tick made");
+        assert_eq!(std::fs::read(&store)?, before, "a contended publish changed the projection");
+        let snapshot = ControlSnapshot::read(&world.state_store)?;
+        assert!(snapshot.transactions.is_empty(), "continuity minted over a projection it could not demote");
+        let recorded = snapshot.supervision_for("service").context("no deferral recorded")?.value.clone();
+        assert!(recorded.continuity_restarts.is_empty(), "a deferral spent an attempt");
+        let reason = recorded.continuity_deferral_reason.clone().context("no deferral reason")?;
+        assert!(reason.contains("locked by another holder"), "{reason}");
+        assert!(!reason.contains(&world.root.display().to_string()), "{reason}");
+
+        // The holder is gone and the deferral has run out: the demotion lands
+        // and the restart proceeds.
+        let mut ran_out = recorded;
+        ran_out.continuity_deferred_until = Some(1);
+        set_meters(&world, &ran_out)?;
+        assert!(world.engine.supervise_one_admitted_generation()?);
+        assert_eq!(ControlSnapshot::read(&world.state_store)?.transactions.len(), 1);
+        assert_eq!(projected_under(&world, &incumbent.expected)?, expected_only());
+        Ok(())
+    }
+
+    /// A deployment phase records a publication digest only after the publish
+    /// returned: the digest is the claim that it landed.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_contended_expected_publish_is_not_recorded_as_published() -> Result<()> {
+        use crate::incident::tests::{hold_lock_of, run_while_held};
+
+        let world = EngineFixture::with_workload(Arc::new(StillWorkload))?;
+        let seeded = transaction_at(&world, DeploymentPhase::Warming)?;
+        let stored = resident(&world)?;
+        let mut starting = seeded.clone();
+        starting.phase = DeploymentPhase::Starting;
+        starting.expected_publication_sha256 = None;
+        starting.activation = None;
+        starting.workload = None;
+        starting.activation_publication_sha256 = None;
+        starting.warming = None;
+        replace_transaction(&world.state_store, &stored, &starting)?;
+        let store = world.engine.options.topology_store.clone();
+        let stored = resident(&world)?;
+
+        let error = run_while_held(hold_lock_of(&store), 1, || {
+            world.engine.advance_transaction(&stored)
+        })
+        .expect_err("a contended publish must not advance the phase");
+        assert!(format!("{error:#}").contains("locked by another holder"), "{error:#}");
+        assert!(!store.exists(), "a contended publish wrote the projection");
+        let after = resident(&world)?.value;
+        assert_eq!(after.phase, DeploymentPhase::Starting);
+        assert_eq!(after.expected_publication_sha256, None);
+
+        // Free again: the same step publishes and records what it published.
+        world.engine.advance_transaction(&resident(&world)?)?;
+        let expected = seeded.expected.context("seeded transaction has no Expected")?;
+        assert_eq!(
+            resident(&world)?.value.expected_publication_sha256,
+            Some(expected.canonical_sha256()?)
+        );
+        Ok(())
+    }
+
+    /// A Continuity transaction parked in Starting with nothing published yet,
+    /// plus the stored record the tick will see.
+    #[cfg(target_os = "linux")]
+    fn parked_in_starting(world: &EngineFixture) -> Result<DeploymentTransaction> {
+        let seeded = transaction_at(world, DeploymentPhase::Warming)?;
+        let stored = resident(world)?;
+        let mut starting = seeded.clone();
+        starting.phase = DeploymentPhase::Starting;
+        starting.expected_publication_sha256 = None;
+        starting.activation = None;
+        starting.workload = None;
+        starting.activation_publication_sha256 = None;
+        starting.warming = None;
+        replace_transaction(&world.state_store, &stored, &starting)?;
+        Ok(seeded)
+    }
+
+    /// A publish that lost to a held lock is transient: before the fence the
+    /// transaction stays in its phase with the error recorded, no abort is
+    /// begun, and the next tick after the release lands the publish. A
+    /// continuity restart is a Continuity transaction on this same path, so
+    /// its attempt is minted once and a contended publish spends no second one.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_contended_starting_publish_is_retried_on_a_later_tick_not_aborted() -> Result<()> {
+        use crate::incident::tests::{hold_lock_of, run_while_held};
+
+        let world = EngineFixture::with_workload(Arc::new(StillWorkload))?;
+        let seeded = parked_in_starting(&world)?;
+        let mut meters = TargetSupervision::new("service");
+        meters.continuity_restarts = vec![now_millis()?];
+        set_meters(&world, &meters)?;
+        let store = world.engine.options.topology_store.clone();
+
+        for tick in 0..2 {
+            run_while_held(hold_lock_of(&store), 1, || world.engine.resume_one_transaction())?;
+            world.engine.resume_backoff.lock().unwrap().clear();
+            let held = resident(&world)?.value;
+            assert_eq!(held.phase, DeploymentPhase::Starting, "tick {tick}");
+            assert!(held.pre_fencing_abort.is_none(), "tick {tick}: a held lock aborted the transaction");
+            assert!(!held.is_terminal(), "tick {tick}");
+            assert_eq!(held.expected_publication_sha256, None, "tick {tick}");
+            let error = held.last_error.context("a contended publish left no last_error")?;
+            assert!(error.contains("locked by another holder"), "{error}");
+            assert!(!store.exists(), "a contended publish wrote the projection");
+        }
+
+        // Released: the next tick lands the publish and records its digest.
+        world.engine.resume_one_transaction()?;
+        let landed = resident(&world)?.value;
+        assert_eq!(landed.phase, DeploymentPhase::Starting);
+        assert!(landed.pre_fencing_abort.is_none());
+        assert_eq!(
+            landed.expected_publication_sha256,
+            Some(seeded.expected.context("seeded transaction has no Expected")?.canonical_sha256()?)
+        );
+        assert_eq!(meters_of(&world, "service")?.continuity_restarts.len(), 1);
+        Ok(())
+    }
+
+    /// Only contention is transient: a publish that fails for any other reason
+    /// still ends the transaction before the fence, as it always did.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_non_contention_publish_error_still_aborts_before_the_fence() -> Result<()> {
+        let world = EngineFixture::with_workload(Arc::new(StillWorkload))?;
+        parked_in_starting(&world)?;
+        // A directory where the store should be: the snapshot cannot be read.
+        std::fs::create_dir_all(&world.engine.options.topology_store)?;
+        world.engine.resume_one_transaction()?;
+        let after = resident(&world)?.value;
+        assert!(after.pre_fencing_abort.is_some(), "a non-contention error did not abort");
+        assert!(
+            !after.last_error.as_deref().unwrap_or_default().contains("locked by another holder"),
+            "{:?}",
+            after.last_error
+        );
+        Ok(())
+    }
+
     #[test]
     fn status_prints_an_open_incident_under_its_target() -> Result<()> {
         let workload = SwitchWorkload::new();

src/drivers.rs

@@ -4247,6 +4247,10 @@ impl WriteLeasePort for CultCacheWriteLeaseDriver {
 /// facts here. Service presence is absent by construction; only Odin may
 /// correlate these records with signed runtime observation into Present/Ready.
 ///
+/// The store is published 0644 and read lock-free by Odin and Idunn; its lock
+/// is Idunn's alone (`exchange_behind_private_lock`), so no other uid can hold
+/// it and stall a publish.
+///
 /// Every record is keyed by the incarnation it describes, never by the target
 /// alone. A target being replaced has two incarnations at once -- the admitted
 /// incumbent and the sealed candidate -- and both are projected side by side.
@@ -4390,22 +4394,28 @@ impl CultCacheTopologyDriver {
     /// Apply one whole-snapshot mutation with compare-and-swap. `mutate`
     /// returns `None` when the snapshot already has the shape it wants, and
     /// the replacement set otherwise; every writer below is one of these.
+    ///
+    /// Never waits: a publish is done exactly when this returns `Ok`. Only a
+    /// lost race (the store moved on) is retried at once; a held lock is an
+    /// error and a no-op, retried by the caller's own retry.
     fn mutate<F>(&self, mutate: F) -> Result<()>
     where
         F: Fn(&[CultCacheEnvelope]) -> Result<Option<Vec<CultCacheEnvelope>>>,
     {
-        if let Some(parent) = self.projection_store.parent() {
-            fs::create_dir_all(parent)?;
-        }
-        let store = SingleFileMessagePackBackingStore::new(&self.projection_store);
         for _ in 0..8 {
             let entries = self.snapshot()?;
             let Some(replacement) = mutate(&entries)? else {
                 return Ok(());
             };
-            if store.compare_exchange_snapshot(&entries, &replacement)? {
-                publish_projection_mode(&self.projection_store)?;
-                return Ok(());
+            match exchange_behind_private_lock(&self.projection_store, &entries, &replacement)? {
+                TryCompareExchangeSnapshotOutcome::Exchanged => {
+                    publish_file_mode(&self.projection_store)?;
+                    return Ok(());
+                }
+                TryCompareExchangeSnapshotOutcome::Mismatch => {}
+                TryCompareExchangeSnapshotOutcome::LockContended => {
+                    return Err(LockContended { store: "topology projection" }.into());
+                }
             }
         }
         bail!("CultCache projection changed repeatedly during publication")
@@ -5097,6 +5107,22 @@ pub enum RouteActuation {
     Survival,
 }
 
+/// A publish that lost to a held projection-directory lock. Transient by
+/// kind: the holder lets go, and the caller's own retry decides what a later
+/// attempt does. It names the store kind, never its path.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct LockContended {
+    pub(crate) store: &'static str,
+}
+
+impl std::fmt::Display for LockContended {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(formatter, "the {} is locked by another holder", self.store)
+    }
+}
+
+impl std::error::Error for LockContended {}
+
 /// A forward route actuation the ceiling refused. `reopens_at_unix_millis` is
 /// when the oldest counted actuation leaves the window.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -7710,19 +7736,95 @@ pub(crate) fn publish_file_mode(path: &Path) -> Result<()> {
     Ok(())
 }
 
-/// The topology store is Idunn's *published* surface: every managed target
-/// reads it to verify its own Expected incarnation against the Idunn anchor.
+/// The one write primitive for a store in the projection directory: one
+/// nonblocking compare-exchange of the whole store against the snapshot it was
+/// decided on, behind a lock only Idunn can hold.
 ///
-/// Idunn runs with `UMask=027`, which is right for its private state and wrong
-/// for this one file -- it lands `0640 root:root`, and a `DynamicUser` workload
-/// gets EACCES. Integrity here comes from the signatures over the records, not
-/// from the mode, so the published copy is readable. The lock sibling too:
-/// CultCache opens it alongside the store, so a 0640 lock denies the read just
-/// as surely as a 0640 store, and it is created fresh under Idunn's umask on
-/// every publish.
-fn publish_projection_mode(path: &Path) -> Result<()> {
-    publish_file_mode(path)?;
-    publish_file_mode(&authority_lock_path(path))
+/// Readers of a published store take no lock, so nobody but Idunn needs to
+/// open its lock, and a lock another uid can open is a lock another uid can
+/// hold to stall every publish. CultCache creates a lock with mode 0666, so
+/// the directory's default ACL would decide who may (Yggdrasil's lands it
+/// 0644). This makes the lock private first, by this rule: an absent lock is
+/// created `O_EXCL` 0600; a lock that is not (a regular file, ours, 0600) is
+/// replaced by a fresh 0600 file renamed over it, because a chmod does not
+/// revoke a descriptor another uid already holds, while the rename leaves that
+/// holder only an orphaned inode. A private lock is opened as it is. No
+/// existing lock is ever chmodded. Nothing here names a path in an error.
+pub(crate) fn exchange_behind_private_lock(
+    store: &Path,
+    expected: &[CultCacheEnvelope],
+    replacements: &[CultCacheEnvelope],
+) -> Result<TryCompareExchangeSnapshotOutcome> {
+    make_lock_private(store)?;
+    SingleFileMessagePackBackingStore::new(store)
+        .try_compare_exchange_snapshot(expected, replacements)
+}
+
+#[cfg(unix)]
+fn make_lock_private(store: &Path) -> Result<()> {
+    use std::os::unix::fs::{MetadataExt, PermissionsExt};
+
+    let lock = authority_lock_path(store);
+    if let Some(parent) = lock.parent() {
+        fs::create_dir_all(parent).context("creating the store directory")?;
+    }
+    // SAFETY: geteuid has no preconditions.
+    let euid = unsafe { libc::geteuid() };
+    // The daemon is the only writer, so a lost create race is defensive only.
+    for _ in 0..3 {
+        match fs::symlink_metadata(&lock) {
+            Ok(metadata)
+                if metadata.is_file()
+                    && metadata.uid() == euid
+                    && metadata.permissions().mode() & 0o777 == 0o600 =>
+            {
+                return Ok(());
+            }
+            Ok(_) => return replace_with_private_lock(&lock),
+            Err(error) if error.kind() == ErrorKind::NotFound => {
+                match create_private_file(&lock) {
+                    Ok(()) => return Ok(()),
+                    Err(error) if error.kind() == ErrorKind::AlreadyExists => {}
+                    Err(error) => return Err(error).context("creating a private store lock"),
+                }
+            }
+            Err(error) => return Err(error).context("inspecting a store lock"),
+        }
+    }
+    bail!("a store lock kept changing while it was made private")
+}
+
+#[cfg(not(unix))]
+fn make_lock_private(store: &Path) -> Result<()> {
+    let _ = store;
+    Ok(())
+}
+
+/// A new file nothing else has open, mode 0600 (the create mode is intersected
+/// with the umask and the directory's default ACL, which only narrow it).
+#[cfg(unix)]
+fn create_private_file(path: &Path) -> std::io::Result<()> {
+    use std::os::unix::fs::OpenOptionsExt;
+
+    OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .mode(0o600)
+        .open(path)
+        .map(drop)
+}
+
+/// Stage a fresh private file beside `lock` and rename it over the lock.
+#[cfg(unix)]
+fn replace_with_private_lock(lock: &Path) -> Result<()> {
+    let mut staged = lock.as_os_str().to_os_string();
+    staged.push(format!(".{}", Uuid::new_v4().simple()));
+    let staged = PathBuf::from(staged);
+    create_private_file(&staged).context("staging a private store lock")?;
+    fs::rename(&staged, lock).map_err(|error| {
+        let _ = fs::remove_file(&staged);
+        anyhow::Error::new(error).context("replacing a store lock with a private one")
+    })
 }
 
 fn build_machine_id(workspace: &Path) -> Result<String> {
@@ -9670,27 +9772,335 @@ Content-Le".to_vec()), |_| {}),
         ));
     }
 
-    #[cfg(unix)]
-    #[test]
-    fn the_published_projection_is_readable_by_a_workload() {
-        use std::os::unix::fs::PermissionsExt;
+    /// A native process that is the Expected's: what `publish_observed_activation`
+    /// checks the activation against.
+    #[cfg(target_os = "linux")]
+    fn observation_of(
+        expected: &IdunnExpectedIncarnationRecord,
+        activation: &IdunnRuntimeActivationRecord,
+    ) -> WorkloadObservation {
+        let mut host = host_audit_observation(1, 1);
+        host.runtime_instance_id = activation.runtime_instance_id.clone();
+        host.executable_sha256 = expected.artifact_sha256.clone();
+        WorkloadObservation::Host(host)
+    }
 
+    #[cfg(target_os = "linux")]
+    fn mode_of(path: &Path) -> u32 {
+        use std::os::unix::fs::PermissionsExt;
+        fs::metadata(path).unwrap().permissions().mode() & 0o777
+    }
+
+    #[cfg(target_os = "linux")]
+    fn inode_of(path: &Path) -> u64 {
+        use std::os::unix::fs::MetadataExt;
+        fs::metadata(path).unwrap().ino()
+    }
+
+    /// Idunn's `UMask=027` alone lands a file 0640: the store still has to be
+    /// published, and its lock has to stay private.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_published_store_is_readable_by_a_workload_and_its_lock_is_not() -> Result<()> {
+        use crate::incident::tests::under_service_umask;
+
+        let temp = tempfile::tempdir()?;
+        let (expected, _, _, anchor) = authenticated_warming(temp.path())?;
+        let driver = topology_driver(temp.path(), "topology");
+        under_service_umask(|| driver.publish_expected(&expected, &anchor))?;
+        assert_eq!(mode_of(&driver.projection_store), 0o644);
+        assert_eq!(mode_of(&authority_lock_path(&driver.projection_store)), 0o600);
+        Ok(())
+    }
+
+    /// The body's condition: the projection directory carries the default ACL
+    /// `u::rw g::r o::r`, under which the umask is ignored and a lock CultCache
+    /// creates lands 0644. Every writer publishes the store 0644 and leaves
+    /// the lock 0600, and another uid cannot open the lock.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn under_a_default_acl_the_projection_is_published_and_its_lock_is_private() -> Result<()> {
+        use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
+        use crate::incident::tests::{give_default_acl, readable_by_another_uid, under_service_umask};
+
+        let temp = tempfile::tempdir()?;
+        let (expected, activation, _, anchor) = authenticated_warming(temp.path())?;
+        let observation = observation_of(&expected, &activation);
+        let dir = temp.path().join("projection");
+        fs::create_dir(&dir)?;
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o755))?;
+        give_default_acl(&dir);
+        let driver = topology_driver(&dir, "topology");
+        let store = driver.projection_store.clone();
+        let lock = authority_lock_path(&store);
+
+        // Control: the lock CultCache would have made here. Without the ACL in
+        // force this reads 0640 and the assertions below prove nothing.
+        let control = dir.join("control.lock");
+        under_service_umask(|| {
+            OpenOptions::new().create(true).write(true).mode(0o666).open(&control)
+        })?;
+        assert_eq!(mode_of(&control), 0o644, "the default ACL is not in force");
+
+        let published = |step: &str| {
+            assert_eq!(mode_of(&store), 0o644, "the store after {step}");
+            assert_eq!(mode_of(&lock), 0o600, "the lock after {step}");
+        };
+        under_service_umask(|| driver.publish_expected(&expected, &anchor))?;
+        published("publish_expected");
+        let private_lock = inode_of(&lock);
+        under_service_umask(|| {
+            driver.publish_observed_activation(&expected, &activation, &observation)
+        })?;
+        published("publish_observed_activation");
+        under_service_umask(|| {
+            driver.demote_to_expected_only(&expected, &anchor, &activation, None)
+        })?;
+        published("demote_to_expected_only");
+        under_service_umask(|| driver.withdraw_incarnation(&expected, &anchor, None, None))?;
+        published("withdraw_incarnation");
+        assert_eq!(inode_of(&lock), private_lock, "a private lock was replaced");
+
+        // The consequence: another uid cannot open the lock, so it cannot hold
+        // it. Provable only as root with setpriv, and only if that uid can
+        // read the control file, which shows the switch itself works.
+        if readable_by_another_uid(&control) == Some(true) {
+            assert_eq!(readable_by_another_uid(&lock), Some(false));
+        } else {
+            eprintln!("UNPROVEN: no uid switch here; the lock modes above stand alone");
+        }
+        Ok(())
+    }
+
+    /// Another process holds the lock: the publish is an error and a no-op, and
+    /// it lands once the lock is free. Nothing waits (`run_while_held` counts
+    /// the opens of the lock), and the error names the store kind, not a path.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_topology_publish_never_waits_on_a_held_lock_and_lands_when_freed() -> Result<()> {
+        use crate::incident::tests::{eventually, hold_lock_of, run_while_held};
+
+        let temp = tempfile::tempdir()?;
+        let (expected, activation, _, anchor) = authenticated_warming(temp.path())?;
+        let observation = observation_of(&expected, &activation);
+        let driver = topology_driver(temp.path(), "topology");
+
+        let error = run_while_held(hold_lock_of(&driver.projection_store), 1, || {
+            driver.publish_expected(&expected, &anchor)
+        })
+        .unwrap_err();
+        assert_eq!(error.to_string(), "the topology projection is locked by another holder");
+        assert!(!driver.projection_store.exists(), "a contended publish wrote the store");
+
+        eventually(|| driver.publish_expected(&expected, &anchor));
+        let landed = fs::read(&driver.projection_store)?;
+
+        // An existing store is unchanged by a contended write of any kind.
+        let error = run_while_held(hold_lock_of(&driver.projection_store), 1, || {
+            driver.publish_observed_activation(&expected, &activation, &observation)
+        })
+        .unwrap_err();
+        assert_eq!(error.to_string(), "the topology projection is locked by another holder");
+        assert_eq!(fs::read(&driver.projection_store)?, landed);
+        eventually(|| driver.publish_observed_activation(&expected, &activation, &observation));
+        assert_ne!(fs::read(&driver.projection_store)?, landed);
+        Ok(())
+    }
+
+    /// The load-bearing rule (a chmod does not revoke a holder): a wider lock
+    /// that another process already holds is replaced by a fresh private one,
+    /// so the holder keeps only an orphaned inode and cannot block a publish.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_holder_of_a_wider_lock_cannot_block_a_publish() -> Result<()> {
+        use std::os::unix::fs::PermissionsExt;
+        use crate::incident::tests::hold_lock_of;
+
+        let temp = tempfile::tempdir()?;
+        let (expected, _, _, anchor) = authenticated_warming(temp.path())?;
+        let driver = topology_driver(temp.path(), "topology");
+        let lock = authority_lock_path(&driver.projection_store);
+        fs::write(&lock, b"")?;
+        fs::set_permissions(&lock, fs::Permissions::from_mode(0o644))?;
+        let wider = inode_of(&lock);
+        let holder = hold_lock_of(&driver.projection_store);
+
+        driver.publish_expected(&expected, &anchor)?;
+        assert_eq!(mode_of(&lock), 0o600);
+        assert_ne!(inode_of(&lock), wider, "the wider lock was chmodded, not replaced");
+        assert!(holder.is_held(), "the publish ended the holder instead of outliving it");
+        assert_eq!(driver.snapshot()?.len(), 2);
+        Ok(())
+    }
+
+    /// Every lock that is not a regular file, ours, mode 0600 is replaced, and
+    /// a private one is left alone.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn every_lock_that_is_not_private_is_replaced_and_a_private_one_is_not() -> Result<()> {
+        use std::os::unix::fs::{MetadataExt, PermissionsExt};
+
+        let temp = tempfile::tempdir()?;
+        let (expected, activation, _, anchor) = authenticated_warming(temp.path())?;
+        let observation = observation_of(&expected, &activation);
+        let driver = topology_driver(temp.path(), "topology");
+        let lock = authority_lock_path(&driver.projection_store);
+        driver.publish_expected(&expected, &anchor)?;
+        // Each call changes the projection, so each one exchanges.
+        let toggle = || -> Result<()> {
+            if driver.projected_activation_is_present(&expected)? {
+                driver
+                    .demote_to_expected_only(&expected, &anchor, &activation, None)
+                    .map(drop)
+            } else {
+                driver
+                    .publish_observed_activation(&expected, &activation, &observation)
+                    .map(drop)
+            }
+        };
+        // SAFETY: geteuid has no preconditions.
+        let euid = unsafe { libc::geteuid() };
+
+        for (shape, mode, owner) in [
+            ("0640", 0o640, None),
+            ("0666", 0o666, None),
+            ("0700", 0o700, None),
+            ("0400", 0o400, None),
+            ("0604", 0o604, None),
+            ("0600 owned by another uid", 0o600, Some(65534)),
+        ] {
+            if owner.is_some() && euid != 0 {
+                eprintln!("UNPROVEN: not root, so a lock owned by another uid cannot be made");
+                continue;
+            }
+            fs::set_permissions(&lock, fs::Permissions::from_mode(mode))?;
+            if let Some(uid) = owner {
+                std::os::unix::fs::chown(&lock, Some(uid), Some(uid))?;
+            }
+            let before = inode_of(&lock);
+            toggle()?;
+            assert_ne!(inode_of(&lock), before, "a {shape} lock was kept");
+            assert_eq!(mode_of(&lock), 0o600, "after a {shape} lock");
+            assert_eq!(fs::metadata(&lock)?.uid(), euid, "after a {shape} lock");
+        }
+
+        // A lock that is a symlink is replaced, never followed.
+        let target = temp.path().join("elsewhere");
+        fs::write(&target, b"")?;
+        fs::set_permissions(&target, fs::Permissions::from_mode(0o600))?;
+        fs::remove_file(&lock)?;
+        std::os::unix::fs::symlink(&target, &lock)?;
+        toggle()?;
+        assert!(!fs::symlink_metadata(&lock)?.file_type().is_symlink());
+        assert_eq!(mode_of(&lock), 0o600);
+        assert!(target.exists());
+
+        // A private lock is opened as it is: two more exchanges, one inode.
+        let kept = inode_of(&lock);
+        toggle()?;
+        toggle()?;
+        assert_eq!(inode_of(&lock), kept);
+        Ok(())
+    }
+
+    /// A lock that cannot be inspected is an inspection error, and one that
+    /// cannot be created is a creation error: neither is mistaken for the
+    /// other, and neither is retried as if another creator had won.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_lock_that_cannot_be_inspected_or_created_says_which() {
         let temp = tempfile::tempdir().unwrap();
-        let path = temp.path().join("topology.cc");
-        std::fs::write(&path, b"x").unwrap();
-        // What Idunn's UMask=027 leaves behind.
-        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o640)).unwrap();
-        let lock = temp.path().join("topology.cc.lock");
-        std::fs::write(&lock, b"").unwrap();
-        std::fs::set_permissions(&lock, std::fs::Permissions::from_mode(0o640)).unwrap();
-        publish_projection_mode(&path).unwrap();
-        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
-        assert_eq!(
-            mode, 0o644,
-            "every target must be able to read the projection"
-        );
-        let lock_mode = std::fs::metadata(&lock).unwrap().permissions().mode() & 0o777;
-        assert_eq!(lock_mode, 0o644, "the lock is opened alongside the store");
+        // The lock's name is one byte past NAME_MAX: inspecting it fails with
+        // an error that is not "not found".
+        let long = temp.path().join("a".repeat(251));
+        let error = exchange_behind_private_lock(&long, &[], &[]).unwrap_err();
+        assert!(format!("{error:#}").contains("inspecting a store lock"), "{error:#}");
+
+        // Absent, but nothing can be created beside it: procfs refuses.
+        let error = exchange_behind_private_lock(Path::new("/proc/self/idunn-store.cc"), &[], &[])
+            .unwrap_err();
+        assert!(format!("{error:#}").contains("creating a private store lock"), "{error:#}");
+    }
+
+    /// Many first publishers race to create the same lock. Whoever loses finds
+    /// a private lock and carries on: no call fails for having lost.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn publishers_racing_to_create_the_lock_all_find_it_private() {
+        use std::sync::Barrier;
+
+        const PUBLISHERS: usize = 8;
+        for round in 0..100 {
+            let temp = tempfile::tempdir().unwrap();
+            let store = temp.path().join("topology.cc");
+            let start = Barrier::new(PUBLISHERS);
+            let outcomes = thread::scope(|scope| {
+                (0..PUBLISHERS)
+                    .map(|_| {
+                        scope.spawn(|| {
+                            start.wait();
+                            exchange_behind_private_lock(&store, &[], &[]).map(drop)
+                        })
+                    })
+                    .collect::<Vec<_>>()
+                    .into_iter()
+                    .map(|publisher| publisher.join().unwrap())
+                    .collect::<Vec<_>>()
+            });
+            for outcome in outcomes {
+                outcome.unwrap_or_else(|error| panic!("round {round}: {error:#}"));
+            }
+            assert_eq!(mode_of(&authority_lock_path(&store)), 0o600);
+        }
+    }
+
+    /// A lost race is retried at once, against the store as it now is; a race
+    /// lost every time is an error rather than a loop.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn a_lost_race_is_retried_against_the_new_store_and_never_forever() -> Result<()> {
+        use std::cell::Cell;
+
+        let temp = tempfile::tempdir()?;
+        let driver = topology_driver(temp.path(), "topology");
+        let rival = |key: &str| CultCacheEnvelope {
+            key: key.into(),
+            r#type: "test.rival".into(),
+            payload: vec![0x90],
+            stored_at: "2026-01-01T00:00:00+00:00".into(),
+            schema_id: Some("test.rival.v1".into()),
+        };
+
+        // Lost once: a rival writes between the snapshot and the exchange.
+        let calls = Cell::new(0);
+        driver.mutate(|entries| {
+            calls.set(calls.get() + 1);
+            if calls.get() == 1 {
+                upsert_record(&driver.projection_store, rival("first")).unwrap();
+            }
+            let mut next = entries.to_vec();
+            next.push(rival(&format!("mine-{}", calls.get())));
+            Ok(Some(next))
+        })?;
+        assert_eq!(calls.get(), 2);
+        let keys = driver.snapshot()?.into_iter().map(|e| e.key).collect::<Vec<_>>();
+        assert_eq!(keys, ["first", "mine-2"]);
+
+        // Lost every time: bounded, and the store is the rival's.
+        let calls = Cell::new(0);
+        let error = driver
+            .mutate(|entries| {
+                calls.set(calls.get() + 1);
+                upsert_record(&driver.projection_store, rival(&format!("rival-{}", calls.get()))).unwrap();
+                let mut next = entries.to_vec();
+                next.push(rival("never"));
+                Ok(Some(next))
+            })
+            .unwrap_err();
+        assert!(error.to_string().contains("changed repeatedly"), "{error}");
+        assert_eq!(calls.get(), 8);
+        assert!(driver.snapshot()?.iter().all(|entry| entry.key != "never"));
+        Ok(())
     }
 
     #[cfg(unix)]

src/incident.rs

@@ -20,7 +20,7 @@ use cultcache_rs::{
 use serde::{Deserialize, Serialize};
 
 use crate::control_plane::{decode_record, require_id, typed_envelope};
-use crate::drivers::publish_file_mode;
+use crate::drivers::{LockContended, exchange_behind_private_lock, publish_file_mode};
 
 const INCIDENT_SCHEMA: &str = "idunn.operator_incident.v1";
 
@@ -177,8 +177,9 @@ impl Snapshot {
 /// scheduler tick, so no write waits: each is one nonblocking compare-exchange
 /// of the whole file against the snapshot it was decided on. A lost race or a
 /// held lock is an error and a no-op, retried on a later pass. The lock beside
-/// the file is not published and is created 0600 (`secure_lock`): the readers
-/// of `incidents.cc` take no lock, so only Idunn can ever hold it.
+/// the file is not published and is Idunn's alone
+/// (`exchange_behind_private_lock`): the readers of `incidents.cc` take no
+/// lock, so only Idunn can ever hold it.
 pub(crate) struct IncidentStore {
     path: PathBuf,
 }
@@ -231,16 +232,13 @@ impl IncidentStore {
         expected: &[CultCacheEnvelope],
         replacements: &[CultCacheEnvelope],
     ) -> Result<()> {
-        secure_lock(path)?;
-        match SingleFileMessagePackBackingStore::new(path)
-            .try_compare_exchange_snapshot(expected, replacements)?
-        {
+        match exchange_behind_private_lock(path, expected, replacements)? {
             TryCompareExchangeSnapshotOutcome::Exchanged => Ok(()),
             TryCompareExchangeSnapshotOutcome::Mismatch => {
                 bail!("the incident store changed while it was being written")
             }
             TryCompareExchangeSnapshotOutcome::LockContended => {
-                bail!("the incident store is locked by another holder")
+                Err(LockContended { store: "incident store" }.into())
             }
         }
     }
@@ -335,46 +333,6 @@ impl IncidentStore {
     }
 }
 
-/// Make the lock beside `store` Idunn's alone, before CultCache opens it.
-///
-/// CultCache creates a lock with mode 0666, so the published directory decides
-/// who may open it: under Yggdrasil's default ACL (`u::rw g::r o::r`) that is
-/// 0644, and any uid could hold the lock and keep every incident from being
-/// recorded. Created here with 0600 the create mode is intersected with the
-/// ACL and still holds, and no wider file ever exists. A lock left wider by an
-/// earlier run is tightened on the descriptor just opened (a descriptor some
-/// other uid already holds is not revoked). Readers of the store take no lock,
-/// so nothing but Idunn needs to open this file.
-fn secure_lock(store: &Path) -> Result<()> {
-    #[cfg(unix)]
-    {
-        use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
-
-        let mut name = store
-            .file_name()
-            .context("the incident store has no file name")?
-            .to_owned();
-        name.push(".lock");
-        let lock = store.with_file_name(name);
-        if let Some(parent) = lock.parent() {
-            std::fs::create_dir_all(parent).context("creating the incident store directory")?;
-        }
-        std::fs::OpenOptions::new()
-            .create(true)
-            .read(true)
-            .write(true)
-            .truncate(false)
-            .mode(0o600)
-            .open(&lock)
-            .context("opening the incident lock")?
-            .set_permissions(std::fs::Permissions::from_mode(0o600))
-            .context("making the incident lock private")?;
-    }
-    #[cfg(not(unix))]
-    let _ = store;
-    Ok(())
-}
-
 /// The lines `idunn status` prints for one target's incidents: each open one,
 /// and each closed one still inside retention.
 pub(crate) fn render_incidents(records: &[IncidentRecord], target: &str, now: u64) -> Vec<String> {
@@ -399,7 +357,7 @@ pub(crate) fn render_incidents(records: &[IncidentRecord], target: &str, now: u6
 
 #[cfg(test)]
 pub(crate) mod tests {
-    #[cfg(unix)]
+    #[cfg(target_os = "linux")]
     use std::time::{Duration, Instant};
 
     use tempfile::TempDir;
@@ -644,7 +602,7 @@ pub(crate) mod tests {
     /// Idunn runs with `UMask=027`. The umask is process-wide, so it is held
     /// only around each call and only one test at a time may hold it.
     #[cfg(target_os = "linux")]
-    fn under_service_umask<T>(write: impl FnOnce() -> T) -> T {
+    pub(crate) fn under_service_umask<T>(write: impl FnOnce() -> T) -> T {
         static UMASK: std::sync::Mutex<()> = std::sync::Mutex::new(());
         let _held = UMASK.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
         // SAFETY: umask only swaps the process file-creation mask.
@@ -658,7 +616,7 @@ pub(crate) mod tests {
     /// `u::rw g::r o::r` that Yggdrasil's idunn-projection directory has. Under
     /// it the umask is ignored and a file created with mode 0666 lands 0644.
     #[cfg(target_os = "linux")]
-    fn give_default_acl(dir: &Path) {
+    pub(crate) fn give_default_acl(dir: &Path) {
         use std::os::unix::ffi::OsStrExt;
 
         // posix_acl_xattr: version 2, then (tag, perm, id) entries. The tags
@@ -691,7 +649,7 @@ pub(crate) mod tests {
     /// Whether uid 65534 can read `path`; `None` when the uid cannot be
     /// switched (not root, or no setpriv), which leaves the question unproven.
     #[cfg(target_os = "linux")]
-    fn readable_by_another_uid(path: &Path) -> Option<bool> {
+    pub(crate) fn readable_by_another_uid(path: &Path) -> Option<bool> {
         std::process::Command::new("setpriv")
             .args(["--reuid=65534", "--regid=65534", "--clear-groups", "cat"])
             .arg(path)
@@ -758,47 +716,154 @@ pub(crate) mod tests {
 
     #[cfg(unix)]
     #[test]
-    fn a_lock_left_wider_by_an_earlier_run_is_tightened_before_it_is_used() {
-        use std::os::unix::fs::PermissionsExt;
+    fn a_lock_left_wider_by_an_earlier_run_is_replaced_before_it_is_used() {
+        use std::os::unix::fs::{MetadataExt, PermissionsExt};
 
         let temp = TempDir::new().unwrap();
         let path = temp.path().join("incidents.cc");
         let lock = temp.path().join("incidents.cc.lock");
         std::fs::write(&lock, b"").unwrap();
         std::fs::set_permissions(&lock, std::fs::Permissions::from_mode(0o666)).unwrap();
+        let before = std::fs::metadata(&lock).unwrap().ino();
 
         assert!(
             IncidentStore::new(&path)
                 .open(IncidentCondition::ContinuityExhausted, "service", OPENED)
                 .unwrap()
         );
-        let mode = std::fs::metadata(&lock).unwrap().permissions().mode() & 0o777;
-        assert_eq!(mode, 0o600);
+        let after = std::fs::metadata(&lock).unwrap();
+        assert_eq!(after.permissions().mode() & 0o777, 0o600);
+        assert_ne!(after.ino(), before, "the wider lock was chmodded, not replaced");
     }
 
-    /// An exclusive lock on a store's lock file, held through another file
-    /// description. flock conflicts between descriptions, so the holder need
-    /// not be another process or another uid: the test does not depend on who
-    /// it runs as.
-    #[cfg(unix)]
-    pub(crate) struct HeldLock(#[allow(dead_code)] std::fs::File);
-
-    #[cfg(unix)]
-    pub(crate) fn hold_lock_of(store: &Path) -> HeldLock {
-        use std::os::unix::io::AsRawFd;
-
+    /// The lock beside a store, spelled independently of the code under test.
+    #[cfg(target_os = "linux")]
+    fn lock_beside(store: &Path) -> PathBuf {
         let mut name = store.file_name().unwrap().to_owned();
         name.push(".lock");
-        let file = std::fs::OpenOptions::new()
+        store.with_file_name(name)
+    }
+
+    /// An exclusive flock on a store's lock, held by a SEPARATE PROCESS
+    /// (`flock -x <lock> sleep`), so the holder shares no descriptor, thread
+    /// or address space with the code under test. A watch on the lock counts
+    /// every open of it from the moment it is held.
+    #[cfg(target_os = "linux")]
+    pub(crate) struct HeldLock {
+        holder: std::sync::Arc<std::sync::Mutex<std::process::Child>>,
+        watch: std::os::fd::OwnedFd,
+    }
+
+    /// Hold the lock of `store`. An absent lock is created as production keeps
+    /// it (0600, ours); a lock a test put there first is held as it is, never
+    /// chmodded.
+    #[cfg(target_os = "linux")]
+    pub(crate) fn hold_lock_of(store: &Path) -> HeldLock {
+        use std::os::fd::{FromRawFd, OwnedFd};
+        use std::os::unix::ffi::OsStrExt;
+        use std::os::unix::fs::OpenOptionsExt;
+        use std::os::unix::io::AsRawFd;
+
+        let lock = lock_beside(store);
+        std::fs::OpenOptions::new()
             .create(true)
-            .read(true)
             .write(true)
             .truncate(false)
-            .open(store.with_file_name(name))
+            .mode(0o600)
+            .open(&lock)
             .unwrap();
-        // SAFETY: flock on a descriptor this function owns.
-        assert_eq!(unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) }, 0);
-        HeldLock(file)
+        let mut holder = std::process::Command::new("flock")
+            .args(["--no-fork", "--exclusive"])
+            .arg(&lock)
+            .args(["sleep", "3600"])
+            .spawn()
+            .expect("flock (util-linux) must be installed to hold a lock from another process");
+
+        // Held once a nonblocking exclusive flock from here is refused.
+        let deadline = Instant::now() + STUCK_AFTER;
+        loop {
+            assert!(holder.try_wait().unwrap().is_none(), "the lock holder exited");
+            let probe = std::fs::File::open(&lock).unwrap();
+            // SAFETY: flock on a descriptor this function owns.
+            let refused = unsafe { libc::flock(probe.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0;
+            if refused {
+                assert_eq!(
+                    std::io::Error::last_os_error().kind(),
+                    std::io::ErrorKind::WouldBlock
+                );
+                break;
+            }
+            drop(probe);
+            assert!(Instant::now() < deadline, "the lock was never held");
+            std::thread::sleep(Duration::from_millis(10));
+        }
+
+        // Open and close alternate, so two opens are never identical adjacent
+        // events for the kernel to coalesce into one.
+        // SAFETY: plain syscalls on descriptors and strings this function owns.
+        let watch = unsafe {
+            let fd = libc::inotify_init1(libc::IN_CLOEXEC | libc::IN_NONBLOCK);
+            assert!(fd >= 0, "{}", std::io::Error::last_os_error());
+            let watch = OwnedFd::from_raw_fd(fd);
+            let path = std::ffi::CString::new(lock.as_os_str().as_bytes()).unwrap();
+            let wd = libc::inotify_add_watch(
+                fd,
+                path.as_ptr(),
+                libc::IN_OPEN | libc::IN_CLOSE_WRITE | libc::IN_CLOSE_NOWRITE,
+            );
+            assert!(wd >= 0, "{}", std::io::Error::last_os_error());
+            watch
+        };
+        HeldLock {
+            holder: std::sync::Arc::new(std::sync::Mutex::new(holder)),
+            watch,
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    impl HeldLock {
+        /// Whether the holder process is still running and so still holds.
+        pub(crate) fn is_held(&self) -> bool {
+            self.holder.lock().unwrap().try_wait().unwrap().is_none()
+        }
+
+        /// How many times the lock has been opened since it was held.
+        fn opens(&self) -> usize {
+            use std::os::unix::io::AsRawFd;
+
+            let mut opens = 0;
+            let mut buffer = [0u8; 4096];
+            loop {
+                // SAFETY: reads into a live buffer of the stated length.
+                let read = unsafe {
+                    libc::read(self.watch.as_raw_fd(), buffer.as_mut_ptr().cast(), buffer.len())
+                };
+                if read <= 0 {
+                    return opens;
+                }
+                let read = usize::try_from(read).unwrap();
+                let mut at = 0;
+                // struct inotify_event: wd i32, mask u32, cookie u32, len u32, name[len].
+                while at + 16 <= read {
+                    let word = |offset: usize| {
+                        u32::from_ne_bytes(buffer[at + offset..at + offset + 4].try_into().unwrap())
+                    };
+                    if word(4) & libc::IN_OPEN != 0 {
+                        opens += 1;
+                    }
+                    at += 16 + usize::try_from(word(12)).unwrap();
+                }
+            }
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    impl Drop for HeldLock {
+        fn drop(&mut self) {
+            let mut holder = self.holder.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+            let _ = holder.kill();
+            let _ = holder.wait();
+        }
     }
 
     /// How long a test lets a held lock, or a released lock's last straggler,
@@ -806,40 +871,57 @@ pub(crate) mod tests {
     /// spends it: the code under test finishes in milliseconds, and the limit
     /// is only what stops a blocked write from hanging the suite. No
     /// assertion compares elapsed time to a bound the machine's load can move.
-    #[cfg(unix)]
+    #[cfg(target_os = "linux")]
     pub(crate) const STUCK_AFTER: Duration = Duration::from_secs(30);
 
-    /// Run `work` while `held` is held, and fail if it waited on it. The holder
-    /// lets go only when `work` has returned, or after `STUCK_AFTER`: work that
-    /// waits on the lock is freed by that second release and is then seen to
-    /// have waited, so the test fails instead of hanging.
-    #[cfg(unix)]
-    pub(crate) fn run_while_held<T>(held: HeldLock, work: impl FnOnce() -> T) -> T {
+    /// Run `work` while `held` is held, and fail if it waited on it. The rule
+    /// is observed where it is decided, at the lock, not on a clock: every
+    /// exchange opens the lock exactly once, so `exchanges` is the number of
+    /// exchanges `work` makes against this lock, and a wait of any length (a
+    /// sleep and retry, a poll, a second attempt) opens it more often. A wait
+    /// that never ends is freed by the holder's release after `STUCK_AFTER`,
+    /// and is then seen to have waited, so the test fails instead of hanging.
+    /// The holder is still holding when `work` returns, and is released on the
+    /// way out.
+    #[cfg(target_os = "linux")]
+    pub(crate) fn run_while_held<T>(
+        held: HeldLock,
+        exchanges: usize,
+        work: impl FnOnce() -> T,
+    ) -> T {
         use std::sync::atomic::{AtomicBool, Ordering};
         use std::sync::{Arc, mpsc};
 
         let (finished, work_done) = mpsc::channel::<()>();
         let gave_up = Arc::new(AtomicBool::new(false));
-        let holder = {
+        let watchdog = {
             let gave_up = Arc::clone(&gave_up);
+            let holder = Arc::clone(&held.holder);
             std::thread::spawn(move || {
                 if work_done.recv_timeout(STUCK_AFTER) == Err(mpsc::RecvTimeoutError::Timeout) {
                     gave_up.store(true, Ordering::SeqCst);
+                    let _ = holder.lock().unwrap().kill();
                 }
-                drop(held);
             })
         };
         let result = work();
+        let still_held = held.is_held();
         let _ = finished.send(());
-        holder.join().unwrap();
+        watchdog.join().unwrap();
         assert!(!gave_up.load(Ordering::SeqCst), "the work waited on a held lock");
+        assert!(still_held, "the holder let go before the work returned");
+        assert_eq!(
+            held.opens(),
+            exchanges,
+            "the work opened the held lock other than once per exchange: it waited or retried"
+        );
         result
     }
 
-    /// Retry `attempt` until it lands. A released lock can stay held for a
-    /// moment by a child another test forked while it was held.
-    #[cfg(unix)]
-    fn eventually<T>(mut attempt: impl FnMut() -> Result<T>) -> T {
+    /// Retry `attempt` until it lands. A lock just released can take a moment
+    /// to be seen free.
+    #[cfg(target_os = "linux")]
+    pub(crate) fn eventually<T>(mut attempt: impl FnMut() -> Result<T>) -> T {
         let deadline = Instant::now() + STUCK_AFTER;
         loop {
             match attempt() {
@@ -850,7 +932,7 @@ pub(crate) mod tests {
         }
     }
 
-    #[cfg(unix)]
+    #[cfg(target_os = "linux")]
     #[test]
     fn no_incident_write_waits_on_a_held_lock_and_each_lands_when_it_is_freed() {
         let temp = TempDir::new().unwrap();
@@ -863,15 +945,17 @@ pub(crate) mod tests {
         let before = std::fs::read(&path).unwrap();
         let retired = OPENED + 1 + INCIDENT_RETENTION_MILLIS + 1;
 
-        let outcomes = run_while_held(hold_lock_of(&path), || {
+        // One exchange each: open, close, and the retirement's store delete.
+        let outcomes = run_while_held(hold_lock_of(&path), 3, || {
             (
                 store.open(condition, "new", CLOSED),
                 store.close(condition, "open", CloseReason::Recovered, CLOSED),
                 store.retire_closed(retired),
             )
         });
-        for outcome in [outcomes.0, outcomes.1, outcomes.2] {
-            assert!(outcome.is_err());
+        for outcome in [outcomes.0.map(drop), outcomes.1.map(drop), outcomes.2.map(drop)] {
+            let error = outcome.unwrap_err().to_string();
+            assert_eq!(error, "the incident store is locked by another holder");
         }
         assert_eq!(std::fs::read(&path).unwrap(), before, "a contended write changed the file");
 
@@ -885,7 +969,7 @@ pub(crate) mod tests {
         assert_eq!(history.pull_all_read_only_snapshot().unwrap().len(), 1);
     }
 
-    #[cfg(unix)]
+    #[cfg(target_os = "linux")]
     #[test]
     fn a_held_history_lock_keeps_the_incident_where_it_is() {
         let temp = TempDir::new().unwrap();
@@ -898,7 +982,7 @@ pub(crate) mod tests {
         let before = std::fs::read(&path).unwrap();
         let retired = OPENED + 1 + INCIDENT_RETENTION_MILLIS + 1;
 
-        let outcome = run_while_held(hold_lock_of(&history), || store.retire_closed(retired));
+        let outcome = run_while_held(hold_lock_of(&history), 1, || store.retire_closed(retired));
         assert!(outcome.is_err());
         assert_eq!(std::fs::read(&path).unwrap(), before);
         assert!(eventually(|| store.retire_closed(retired)));