net/adapter/net/behavior/org_revocation.rs
1//! Restart-persistent organization revocation maxima — OA-1 §1.5 of
2//! `docs/internal/plans/ORG_CAPABILITY_AUTH_PLAN.md`.
3//!
4//! An in-memory monotone merge of
5//! [`OrgRevocationBundle`] floors
6//! is insufficient: if config management replaces the operator's
7//! bundle file with an OLDER (still validly signed) bundle and the
8//! node restarts, there is no prior maximum left to compare
9//! against, and the fleet silently rolls back to weaker floors.
10//! The minimum fix — deliberately NOT the deferred WAL/replication
11//! system — is one small atomic local file of merged maxima
12//! (`revocation-state.json` in the node's authority config
13//! directory).
14//!
15//! # Locked reload order
16//!
17//! ```text
18//! verify incoming bundle signature
19//! → merge maxima with PERSISTED state (monotone; lower never wins)
20//! → atomically write merged maxima
21//! (write temp → fsync temp → atomic rename → fsync parent dir)
22//! → ONLY THEN publish the new live view
23//! ```
24//!
25//! [`OrgRevocationStore::apply_bundle`] implements exactly this
26//! order. Failure handling is asymmetric by design:
27//!
28//! - **Corrupt incoming bundle** → keep the persisted last-good
29//! state, log loudly, return a typed error. Live view untouched.
30//! - **Corrupt persisted maxima file** → LOUD startup failure
31//! ([`OrgRevocationStore::open_existing`] refuses) — protected
32//! verification never starts against silently weaker floors. A
33//! *missing* file at startup is equally loud: absence IS silently
34//! weaker floors. Only `net node adopt`
35//! ([`OrgRevocationStore::init`]) may create the file.
36//!
37//! Unlike the sdk's `RevocationStore`, the parent-directory fsync
38//! here is **not** best-effort: the plan's locked order makes it
39//! part of the durability boundary, and the live view must not
40//! publish a floor the filesystem could forget on crash.
41//!
42//! # Writer model
43//!
44//! The node owns its own maxima file; bundle files distributed by
45//! the operator are inputs, never this file. Same-file writers —
46//! whether a second store instance, a concurrent `net node adopt`,
47//! or another process — are ENFORCED serial (review-8 §5): every
48//! reload holds an exclusive advisory lock on the stable `.lock`
49//! sidecar and rereads the persisted maxima under that lock before
50//! merging, so no writer's floors can be rolled out of the file by
51//! a staler writer's in-memory snapshot.
52//!
53//! # One path, one security view
54//!
55//! Within a process, every [`OrgRevocationStore`] handle backed by
56//! the same NORMALIZED pathname shares one `StoreCore` (review-9
57//! addendum): one live floor view, one reload/publish transaction
58//! lock, one publish generation, and one subscriber registry. A
59//! same-path sibling therefore observes a raise the instant it is
60//! published — one backing file is never modeled as several
61//! independent security views glued to a shared poison boolean.
62//! Opens ALWAYS serialize behind the interprocess state lock (no
63//! pre-lock poison fast path), and durability recovery rereads and
64//! republishes the persisted state through the shared core BEFORE
65//! the path-wide poison bit clears.
66
67use std::collections::BTreeMap;
68use std::path::{Path, PathBuf};
69use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
70use std::sync::{Arc, Weak};
71
72use parking_lot::{Condvar, Mutex, MutexGuard, RwLock};
73use serde::{Deserialize, Serialize};
74
75use super::org::{OrgError, OrgId, OrgRevocationBundle};
76use crate::adapter::net::identity::EntityId;
77
78/// Format version of `revocation-state.json`. Bump requires an
79/// explicit migration; an unknown version is a loud startup
80/// failure, never a silent re-init.
81pub const ORG_REVOCATION_STATE_VERSION: u32 = 1;
82
83/// Merged revocation-floor maxima: for each `(org, member)`, the
84/// highest `minimum_generation` any verified bundle has ever
85/// asserted on this node. Monotone: a merge can only raise floors.
86#[derive(Debug, Clone, Default, PartialEq, Eq)]
87pub struct OrgRevocationState {
88 floors: BTreeMap<(OrgId, EntityId), u32>,
89}
90
91impl OrgRevocationState {
92 /// The empty state (fresh adopt).
93 pub fn empty() -> Self {
94 Self::default()
95 }
96
97 /// Test seam: a state with explicit floors, as a merged bundle would
98 /// leave it, without constructing and signing a bundle. Test-only — never
99 /// a supported downstream constructor for synthetic authority state.
100 #[cfg(test)]
101 pub(crate) fn from_floors_for_test(floors: BTreeMap<(OrgId, EntityId), u32>) -> Self {
102 Self { floors }
103 }
104
105 /// The current floor for `(org, member)`. Absent keys floor at
106 /// 0 — every generation is admissible until a bundle says
107 /// otherwise.
108 pub fn floor_for(&self, org: &OrgId, member: &EntityId) -> u32 {
109 self.floors
110 .get(&(*org, member.clone()))
111 .copied()
112 .unwrap_or(0)
113 }
114
115 /// Number of tracked `(org, member)` floors.
116 pub fn len(&self) -> usize {
117 self.floors.len()
118 }
119
120 /// `true` iff no floors are tracked.
121 pub fn is_empty(&self) -> bool {
122 self.floors.is_empty()
123 }
124
125 /// Iterate floors in canonical `(org, member)` order.
126 pub fn iter(&self) -> impl Iterator<Item = (&(OrgId, EntityId), &u32)> {
127 self.floors.iter()
128 }
129
130 /// Monotone merge: raise each `(bundle.org_id, member)` floor
131 /// to the bundle's value where higher; lower values never win.
132 /// Returns how many floors rose.
133 ///
134 /// Does NOT verify the bundle — the caller does (the store's
135 /// locked order verifies before merging; state-level callers
136 /// such as tests must do the same).
137 pub fn merge_bundle(&mut self, bundle: &OrgRevocationBundle) -> usize {
138 let mut raised = 0;
139 for (member, floor) in bundle.floors() {
140 // §14: a floor of 0 is the IMPLICIT default — `floor_for` returns 0
141 // for an absent key — so materializing an entry for it says
142 // nothing and never expires. `or_insert(0)` used to create a key
143 // for EVERY member a bundle named, unchanged or zero alike, and
144 // `floors` is never pruned, so `revocation-state.json` accumulated
145 // semantically-null `floor: 0` rows permanently. Those rows are
146 // not merely disk noise: `install_org_revocation_store_locked`
147 // walks the whole snapshot and calls `retract_floored_ownership`
148 // per entry, and that takes an EXCLUSIVE fold write lock — so an
149 // org that has named 2,000 members over its lifetime made every
150 // subsequent authority install take 2,000 sequential exclusive
151 // acquisitions, most of which can retract nothing
152 // (`generation < 0` is unsatisfiable for u32), stalling every
153 // concurrent `may_execute` / `has_local_capability` / discovery
154 // query on the node.
155 if *floor == 0 {
156 continue;
157 }
158 let entry = self
159 .floors
160 .entry((bundle.org_id, member.clone()))
161 .or_insert(0);
162 if *floor > *entry {
163 *entry = *floor;
164 raised += 1;
165 }
166 }
167 raised
168 }
169
170 /// Serialize to the versioned on-disk JSON form (sorted by the
171 /// map's canonical order, so the file is deterministic).
172 fn to_file_bytes(&self) -> Result<Vec<u8>, OrgRevocationError> {
173 let file = PersistedStateFile {
174 version: ORG_REVOCATION_STATE_VERSION,
175 floors: self
176 .floors
177 .iter()
178 .map(|((org, member), floor)| PersistedFloor {
179 org: *org,
180 member: member.clone(),
181 floor: *floor,
182 })
183 .collect(),
184 };
185 serde_json::to_vec_pretty(&file).map_err(|e| OrgRevocationError::Io {
186 path: String::new(),
187 reason: format!("serialize revocation state: {e}"),
188 })
189 }
190
191 /// Strict read of a persisted state file that may not exist
192 /// yet: `Ok(None)` when absent, loud typed errors on anything
193 /// unparseable. The adoption ceremony uses this to validate
194 /// candidate floors BEFORE creating any durable state
195 /// (review-8 §7/§8).
196 pub fn load_if_exists(path: &Path) -> Result<Option<Self>, OrgRevocationError> {
197 match read_regular_nofollow(path) {
198 Ok(bytes) => Self::from_file_bytes(&bytes, path).map(Some),
199 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
200 Err(e) => Err(OrgRevocationError::Io {
201 path: path.display().to_string(),
202 reason: e.to_string(),
203 }),
204 }
205 }
206
207 /// Strict parse of the on-disk form. Unknown fields, an
208 /// unsupported version, or duplicate `(org, member)` keys are
209 /// all corruption — loud typed errors, never best-effort
210 /// recovery (a "recovered" state could be a weaker one).
211 fn from_file_bytes(bytes: &[u8], path: &Path) -> Result<Self, OrgRevocationError> {
212 let file: PersistedStateFile =
213 serde_json::from_slice(bytes).map_err(|e| OrgRevocationError::CorruptState {
214 path: path.display().to_string(),
215 detail: e.to_string(),
216 })?;
217 if file.version != ORG_REVOCATION_STATE_VERSION {
218 return Err(OrgRevocationError::UnsupportedVersion {
219 path: path.display().to_string(),
220 found: file.version,
221 });
222 }
223 let mut floors = BTreeMap::new();
224 for entry in file.floors {
225 // §20 — the §14 zero-floor rule is enforced HERE too, not just in
226 // `merge_bundle`.
227 //
228 // A floor of 0 is the IMPLICIT default (`floor_for` returns 0 for
229 // an absent key), so a materialized zero row says nothing, never
230 // expires, and is carried forward by every subsequent write
231 // (`merged = disk.clone()`). Enforcing the invariant at only one of
232 // three entry points left the install-sweep pathology §14 describes
233 // re-openable by any state file that already contained zero rows —
234 // hand-edited, produced by a build predating §14, or grown through
235 // `publish`.
236 //
237 // Dropped rather than rejected: a zero row is semantically
238 // identical to absence, so refusing the whole file would turn a
239 // no-op into an outage.
240 if entry.floor == 0 {
241 continue;
242 }
243 if floors
244 .insert((entry.org, entry.member), entry.floor)
245 .is_some()
246 {
247 return Err(OrgRevocationError::CorruptState {
248 path: path.display().to_string(),
249 detail: "duplicate (org, member) floor entry".to_string(),
250 });
251 }
252 }
253 // §21 — floors are never pruned, and CANNOT be: dropping a floor
254 // un-revokes the member it retired. So this is a soft signal, not a
255 // cap. Refusing at a limit would be worse than the cost it avoids
256 // (the node would fail to load its own revocation state), and evicting
257 // would silently re-admit revoked certificates.
258 //
259 // The cost is real but operator-driven, not attacker-driven:
260 // `apply_bundle` is reachable only from the adopt ceremony, never from
261 // the network. Every raise re-serializes the whole map to pretty JSON
262 // under the cross-process lock, and
263 // `install_org_revocation_store_locked` walks the snapshot taking one
264 // EXCLUSIVE fold write lock per entry. At a few thousand entries that
265 // is a visible stall on every authority install.
266 //
267 // Surfacing it is what an operator can act on: retire the org's
268 // certificate generation and re-issue, so historical floors become
269 // redundant and the state file can be replaced wholesale.
270 if floors.len() >= FLOOR_COUNT_ADVISORY {
271 tracing::warn!(
272 floors = floors.len(),
273 path = %path.display(),
274 "org revocation: the persisted floor set is large; every raise \
275 re-serializes it under the interprocess lock and every authority \
276 install takes one exclusive fold lock per entry. Consider rolling \
277 the org certificate generation so historical floors can be retired.",
278 );
279 }
280 Ok(Self { floors })
281 }
282}
283
284/// Floor count at which [`OrgRevocationState::from_file_bytes`] warns (§21).
285///
286/// Deliberately an ADVISORY threshold rather than a cap. Floors are never
287/// pruned and must not be: dropping one un-revokes the member it retired, and
288/// refusing to load past a limit would take the node down rather than slow it.
289/// Sized well above any plausible steady state so it fires on genuine
290/// accumulation, not on a normally-operating org.
291const FLOOR_COUNT_ADVISORY: usize = 4_096;
292
293/// On-disk shape of `revocation-state.json`. `deny_unknown_fields`:
294/// an entry this node doesn't understand could be a floor it is
295/// about to drop — corruption, not forward compatibility.
296#[derive(Serialize, Deserialize)]
297#[serde(deny_unknown_fields)]
298struct PersistedStateFile {
299 version: u32,
300 floors: Vec<PersistedFloor>,
301}
302
303#[derive(Serialize, Deserialize)]
304#[serde(deny_unknown_fields)]
305struct PersistedFloor {
306 org: OrgId,
307 member: EntityId,
308 floor: u32,
309}
310
311/// Errors from the persisted revocation store.
312#[derive(Debug)]
313pub enum OrgRevocationError {
314 /// The incoming bundle failed signature or structural
315 /// verification. Persisted last-good state is retained.
316 InvalidBundle(OrgError),
317 /// No persisted maxima file at startup. Absence is silently
318 /// weaker floors, so startup must not proceed; only
319 /// `net node adopt` creates the file.
320 MissingState {
321 /// Where the state file was expected.
322 path: String,
323 },
324 /// The persisted maxima file exists but cannot be trusted
325 /// (parse failure, duplicate keys). LOUD startup failure.
326 CorruptState {
327 /// The state file's path.
328 path: String,
329 /// What failed to parse or validate.
330 detail: String,
331 },
332 /// The persisted file's format version is unknown to this
333 /// build.
334 UnsupportedVersion {
335 /// The state file's path.
336 path: String,
337 /// The version the file declares.
338 found: u32,
339 },
340 /// Filesystem failure while reading or durably writing. When
341 /// raised from `apply_bundle` this is always PRE-rename: the
342 /// old file and old live view are both intact.
343 Io {
344 /// The path being read or written.
345 path: String,
346 /// The underlying I/O error.
347 reason: String,
348 },
349 /// The rename LANDED but the parent-directory fsync failed —
350 /// the directory entry may or may not survive a crash, so disk
351 /// and memory can no longer be proven synchronized. The store
352 /// publishes the merged (never-weaker) live view, then poisons
353 /// the BACKING PATH: same-path operations are refused until
354 /// recovery — a locked reread republished through the shared
355 /// core plus a SUCCESSFUL parent-directory fsync —
356 /// re-establishes ground truth (review-8 §13, review-9). A
357 /// restart is one route to that recovery, not the contract.
358 DurabilityUncertain {
359 /// The state file's path.
360 path: String,
361 /// The underlying fsync error.
362 reason: String,
363 },
364 /// A previous apply ended post-rename durability-uncertain
365 /// (see [`Self::DurabilityUncertain`]) and recovery has not yet
366 /// succeeded; same-path reloads and opens are refused until a
367 /// locked reread plus a successful parent-directory fsync
368 /// clears the uncertainty.
369 Poisoned {
370 /// The state file's path.
371 path: String,
372 },
373 /// A running node refused to swap its installed revocation
374 /// store for one whose live view is lower on some `(org,
375 /// member)` key — an installed floor never lowers (review-8
376 /// §4). Reload higher floors through
377 /// [`OrgRevocationStore::apply_bundle`] instead of replacing
378 /// the store.
379 NonMonotonicReplacement {
380 /// The candidate store's state-file path.
381 path: String,
382 },
383 /// R2-4: this backing path is already bound, for the lifetime of a
384 /// live core, to a DIFFERENT `.lock` sidecar identity than the one
385 /// just opened — the sidecar was recreated or replaced underneath a
386 /// core that same-path siblings still hold. Joining under the new
387 /// identity would fork the path into two independent security views,
388 /// so it is refused loudly.
389 BackingIdentityConflict {
390 /// The normalized state-file path whose sidecar identity changed.
391 path: String,
392 },
393}
394
395impl std::fmt::Display for OrgRevocationError {
396 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
397 match self {
398 Self::InvalidBundle(e) => write!(f, "revocation bundle rejected: {e}"),
399 Self::MissingState { path } => write!(
400 f,
401 "revocation state file missing at {path}; refusing to start with \
402 implicitly empty floors — run `net node adopt` to provision"
403 ),
404 Self::CorruptState { path, detail } => write!(
405 f,
406 "revocation state file at {path} is corrupt ({detail}); refusing to \
407 start against silently weaker floors"
408 ),
409 Self::UnsupportedVersion { path, found } => write!(
410 f,
411 "revocation state file at {path} has unsupported version {found} \
412 (this build supports {ORG_REVOCATION_STATE_VERSION})"
413 ),
414 Self::Io { path, reason } => write!(f, "revocation state I/O at {path}: {reason}"),
415 Self::DurabilityUncertain { path, reason } => write!(
416 f,
417 "revocation state at {path}: rename landed but the parent-directory \
418 fsync failed ({reason}); disk and memory can no longer be proven \
419 synchronized — path poisoned until a locked reread and a successful \
420 parent-directory fsync recover it"
421 ),
422 Self::Poisoned { path } => write!(
423 f,
424 "revocation store path {path} is poisoned after a durability-uncertain \
425 write; recovery requires a locked reread republished through the \
426 shared store plus a successful parent-directory fsync (restarting the \
427 process is one route, not the requirement)"
428 ),
429 Self::NonMonotonicReplacement { path } => write!(
430 f,
431 "refusing to replace the installed revocation store with {path}: its \
432 live view is lower on at least one (org, member) floor — an installed \
433 floor never lowers; apply a bundle instead"
434 ),
435 Self::BackingIdentityConflict { path } => write!(
436 f,
437 "revocation store path {path} is bound to a different .lock sidecar \
438 identity than the one just opened — the sidecar was recreated or \
439 replaced while a same-path core is still live; refusing to fork the \
440 path into two independent security views"
441 ),
442 }
443 }
444}
445
446impl std::error::Error for OrgRevocationError {}
447
448/// One floor raise observed by [`OrgRevocationStore::apply_bundle`]
449/// relative to the store's previously published live view —
450/// `(org, member, new_floor)`. Fed to the raise callback so a
451/// running node can retract stale ownership projections
452/// immediately (review-8 §9).
453pub type RaisedFloor = (OrgId, EntityId, u32);
454
455/// Callback invoked after a reload publishes floors higher than the
456/// previously enforced view.
457type FloorsRaisedCallback = Arc<dyn Fn(&[RaisedFloor]) + Send + Sync>;
458
459/// The process-wide state shared by every [`OrgRevocationStore`]
460/// handle backed by one normalized path (review-9 addendum): ONE
461/// live view, ONE reload/publish transaction lock, ONE publish
462/// generation, ONE subscriber registry. Handles are cheap facades;
463/// the core is the security object.
464struct StoreCore {
465 /// The NORMALIZED backing path — used for reads / writes / the
466 /// interprocess lock. NOT the registry key (that is [`BackingId`],
467 /// so case-aliases collapse — AV-9).
468 path: PathBuf,
469 /// Stable backing-file identity (the `.lock` sidecar inode) — the
470 /// key in the core and poison registries (AV-9).
471 backing_id: BackingId,
472 /// Serializes merge→persist→publish transactions in-process
473 /// (the sidecar file lock serializes across processes). Also
474 /// exposed to the node as [`PublishGuard`] so store
475 /// replacement and authority installation can pin the live
476 /// view across their check-then-swap sections.
477 reload: Mutex<()>,
478 /// The one published live view every same-path handle shares.
479 /// Never ahead of the durably persisted state.
480 live: RwLock<Arc<OrgRevocationState>>,
481 /// Bumped on every publish; lets callers order publications.
482 ///
483 /// Advanced with `checked_add`, NEVER wrapping — see
484 /// [`StoreCore::generation_exhausted`].
485 generation: AtomicU64,
486 /// Terminal: the publication generation space is exhausted.
487 ///
488 /// A wrapping counter is not a currentness signal. Once it wraps, a NEW
489 /// floor view carries a generation a consumer has already seen, so evidence
490 /// built against the OLD view compares equal to the new one. Rather than
491 /// wrap, the generation freezes and this latches, and every consumer that
492 /// uses the generation for currentness must fail closed on it (Kyra
493 /// OLB-2B-E3c).
494 ///
495 /// Distinct from POISON, which means "durability uncertain" and can be
496 /// cleared by a successful locked reread. Exhaustion is not recoverable
497 /// in-process: clearing it would hand out an identity already in use.
498 generation_exhausted: AtomicBool,
499 /// Serializes POISON transitions on this core against readers that must hold
500 /// poison immobile.
501 ///
502 /// Poison is a path-registry write, not a view publication, so `live` does
503 /// not order it. A consumer whose decision is itself load-bearing — the
504 /// routing commit pin, whose `Current` causes `Healthy` — must be able to
505 /// hold poison still across its validation AND its settlement, or the two
506 /// remain independently interleavable (Kyra OLB-2B-E3c).
507 ///
508 /// FROZEN ORDER: `poison_gate` → `live`. Every poison transition on a live
509 /// core takes this before any later `live.write()`, and every pin takes it
510 /// before `live.read()`, so no cycle is reachable.
511 ///
512 /// BOTH directions. A recovery CLEAR is as load-bearing as a mark: a pin
513 /// that validated `poisoned == true` and then watched the clear land mid
514 /// settlement reports `Current` for an authority that is already gone. Every
515 /// live-core transition therefore goes through [`StoreCore::mark_poisoned`]
516 /// / [`StoreCore::clear_poison`], never the raw path-registry helpers
517 /// (Kyra OLB-2B-E3c closure).
518 poison_gate: Mutex<()>,
519 /// Test-only: fired ONLY when a publish's `live.try_write()` has actually
520 /// FAILED, immediately before it blocks on `write()`.
521 ///
522 /// Named `contended`, not `blocking`, for the same reason the poison gate
523 /// distinguishes the two: this is EVIDENCE, and it is only evidence because
524 /// the acquisition provably lost. It was previously fired before the attempt,
525 /// which an independent RED pass showed proves nothing — with the settlement
526 /// pin wrongly released, a publisher could signal, acquire the lock, and have
527 /// the observer read a proxy flag before the publisher stored it, so the gap
528 /// witness passed while a publication was occupying the gap it claims is
529 /// closed (Kyra, independent E3c RED pass 2026-07-27).
530 #[cfg(test)]
531 publish_contended_hook: Mutex<Option<Arc<dyn Fn() + Send + Sync>>>,
532 /// Test-only: the same acknowledgment for `poison_gate` — fired immediately
533 /// before a poison transition attempts the lock. Elapsed time is not
534 /// evidence that a contender reached the gate.
535 #[cfg(test)]
536 poison_blocking_hook: Mutex<Option<Arc<dyn Fn() + Send + Sync>>>,
537 /// Test-only, one-shot: force `apply_bundle`'s next state write to report a
538 /// POST-rename durability failure while leaving the file at its prior
539 /// bytes — the exact uncertainty PostRename names. On Windows the phase is
540 /// otherwise unreachable (write-through rename, §13), so the poison-mark
541 /// wake (E3c blockers §1) has no other witness route there.
542 #[cfg(test)]
543 force_post_rename: AtomicBool,
544 /// Test-only: fired by [`StoreCore::lock_poison_gate`] ONLY when its try
545 /// OBSERVED the gate held, immediately before blocking (E3c blockers §3).
546 /// Contrast [`StoreCore::poison_blocking_hook`], which fires before the
547 /// attempt regardless of contention.
548 #[cfg(test)]
549 poison_contended_hook: Mutex<Option<Arc<dyn Fn() + Send + Sync>>>,
550 /// Raise subscribers, each with a removable token. A REGISTRY,
551 /// not a single slot (review-9 addendum): registering a second
552 /// observer must never silently steal the first one's
553 /// notifications.
554 subscribers: RwLock<Vec<(u64, FloorsRaisedCallback)>>,
555 /// Token source for [`Self::subscribers`].
556 next_subscriber: AtomicU64,
557 /// Test-only pause fired ONCE, from inside [`Self::publish`],
558 /// AFTER the live-view swap and BEFORE the generation bump, while
559 /// `live.write()` is still held. Lets a witness deterministically
560 /// occupy the exact "new view installed, old generation still
561 /// present" window the barriered readers must not observe.
562 /// Always `None` in production (armed only by
563 /// [`OrgRevocationStore::arm_publish_pause_for_test`], a
564 /// `#[doc(hidden)]` seam mirroring the review-11 `*_paused_for_test`
565 /// hooks); the per-publish check is an uncontended `Mutex::take`.
566 ///
567 /// §19 — gated behind `cfg(test)`/`feature = "fixtures"`. It was plain
568 /// `pub` (only `#[doc(hidden)]`), and `run_publish_pause_hook` blocks on
569 /// an mpsc `recv()` WHILE `live.write()` is held: any code linked against
570 /// this crate could arm the pause, never send the resume token, and
571 /// permanently wedge `barriered_generation()` and
572 /// `snapshot_with_generation()` — i.e. every admission decision in
573 /// `verify_provider_authority`, plus every other process blocked on the
574 /// interprocess lock. Compiled out of consumer builds entirely.
575 #[cfg(any(test, feature = "fixtures"))]
576 publish_pause: parking_lot::Mutex<Option<PublishPauseHook>>,
577}
578
579/// The one-shot hook a test installs to pause [`StoreCore::publish`]
580/// between the view swap and the generation bump.
581#[cfg(any(test, feature = "fixtures"))]
582struct PublishPauseHook {
583 /// Signalled once the view is swapped and the pause begins.
584 swapped: std::sync::mpsc::Sender<()>,
585 /// Blocks the publisher until the test releases it.
586 resume: std::sync::mpsc::Receiver<()>,
587}
588
589impl StoreCore {
590 /// Swap the live view to `next`, returning every floor that
591 /// rose relative to the previously published view. `next` is
592 /// always a monotone superset under the locked reload order;
593 /// the per-key max with the outgoing view makes "an installed
594 /// floor never lowers" structural rather than assumed.
595 fn publish(&self, mut next: OrgRevocationState) -> Vec<RaisedFloor> {
596 // The acknowledgement fires ONLY after a `try_write` has actually
597 // failed, never merely before attempting the write.
598 //
599 // The earlier form — signal, then block on `write()` — was the forbidden
600 // "hook before the actual synchronization barrier" evidence pattern, and
601 // an independent RED pass proved it: with the settlement pin wrongly
602 // released, the publisher could signal, ACQUIRE the write lock, and have
603 // the observer inspect its proxy flag before the publisher stored the
604 // result — so the witness passed while a publication was occupying the
605 // very gap it claims is closed (Kyra, independent E3c RED pass
606 // 2026-07-27, at `80bb06b5a`).
607 //
608 // A failed `try_write` is the only evidence that cannot be faked by
609 // scheduling: it means some other holder is provably there right now.
610 // Under the same mutation `try_write` SUCCEEDS, no acknowledgement is
611 // ever sent, and the witness fails at its wait instead of passing.
612 #[cfg(test)]
613 let mut live = match self.live.try_write() {
614 // Uncontended. Deliberately silent: acknowledging here would
615 // reintroduce exactly the defect above.
616 Some(guard) => guard,
617 None => {
618 let hook = self.publish_contended_hook.lock().clone();
619 if let Some(hook) = hook {
620 hook();
621 }
622 self.live.write()
623 }
624 };
625 #[cfg(not(test))]
626 let mut live = self.live.write();
627 // §15 — the per-key max keeps the LIVE view safe, but silently
628 // absorbing a weaker incoming state hides the fact that DISK is now
629 // behind what this node is enforcing. `apply_bundle` then builds its
630 // merge from `disk` alone and re-persists that weaker base, so the
631 // divergence becomes permanent and only surfaces at the next restart —
632 // as a floor rollback.
633 //
634 // Nothing here can repair it (this is the in-memory publish, under the
635 // live write lock, with no file lock held), so it is SURFACED instead:
636 // an operator seeing this has a state file that needs restoring or a
637 // bundle re-applied, and would otherwise have no signal at all until
638 // the rollback landed.
639 //
640 // §20 — and the max no longer materializes a zero row for a live key
641 // that is absent from `next`: `or_insert(0)` created exactly the
642 // never-expiring, says-nothing entry §14 removed from `merge_bundle`.
643 let mut regressed = 0usize;
644 for ((org, member), floor) in live.iter() {
645 if *floor == 0 {
646 continue;
647 }
648 let entry = next.floors.entry((*org, member.clone())).or_insert(*floor);
649 if *floor > *entry {
650 *entry = *floor;
651 regressed += 1;
652 }
653 }
654 if regressed > 0 {
655 tracing::error!(
656 keys = regressed,
657 "org revocation: the persisted state is BEHIND the enforced view \
658 for {regressed} floor(s); the live view is preserved, but disk \
659 will re-persist the weaker base and a restart would roll those \
660 floors back. Restore the state file or re-apply the bundles \
661 that raised them.",
662 );
663 }
664 let raised: Vec<RaisedFloor> = next
665 .iter()
666 .filter(|((org, member), floor)| **floor > live.floor_for(org, member))
667 .map(|((org, member), floor)| (*org, member.clone(), *floor))
668 .collect();
669 *live = Arc::new(next);
670 // Occupy the "new view installed, old generation still present"
671 // window while `live` (the write guard) is held, so a witness
672 // can prove the barriered readers never observe it. A no-op
673 // (one uncontended `Mutex::take`) unless a test armed the hook.
674 #[cfg(any(test, feature = "fixtures"))]
675 self.run_publish_pause_hook();
676 // CHECKED, never wrapping: at the ceiling the generation freezes and the
677 // exhaustion latch is set, so a consumer comparing generations for
678 // currentness fails closed instead of matching a reused identity.
679 let current = self.generation.load(Ordering::Acquire);
680 match current.checked_add(1) {
681 Some(next) => self.generation.store(next, Ordering::Release),
682 None => {
683 if !self.generation_exhausted.swap(true, Ordering::AcqRel) {
684 tracing::error!(
685 "org revocation: publication generation space exhausted; \
686 the generation is frozen and every generation-based \
687 currentness check must now fail closed"
688 );
689 }
690 }
691 }
692 raised
693 }
694
695 /// Fire the one-shot publish pause hook if a test installed one.
696 /// Runs while the caller holds `live.write()`.
697 #[cfg(any(test, feature = "fixtures"))]
698 fn run_publish_pause_hook(&self) {
699 if let Some(hook) = self.publish_pause.lock().take() {
700 let _ = hook.swapped.send(());
701 let _ = hook.resume.recv();
702 }
703 }
704
705 /// Notify every subscriber of `raised`. Callers invoke this
706 /// OUTSIDE both the file lock and the reload lock — re-entrant
707 /// callbacks must not deadlock (review-9).
708 fn notify(&self, raised: &[RaisedFloor]) {
709 if raised.is_empty() {
710 return;
711 }
712 let subscribers: Vec<FloorsRaisedCallback> = self
713 .subscribers
714 .read()
715 .iter()
716 .map(|(_, callback)| callback.clone())
717 .collect();
718 for callback in subscribers {
719 callback(raised);
720 }
721 }
722
723 /// Wake every subscriber for an authority change that raised NO floor.
724 ///
725 /// [`Self::notify`] returns immediately on an empty raise set, which is
726 /// right for a publication that changed nothing. A POISON CLEAR is not that:
727 /// recovery republishes the same durable view, so it raises no floor, yet
728 /// what this node is permitted to serve just went from "nothing" back to the
729 /// real material. Without an explicit wake the routing registry stays
730 /// reconciled to obsolete `Unserved` facts until some reader happens to trip
731 /// the lazy epoch check (Kyra OLB-2B-E3c closure).
732 ///
733 /// Callers MUST invoke this with no file lock, reload guard, `poison_gate`
734 /// or `live` guard held — a subscriber takes the routing authority gate, and
735 /// a routing settlement holding that gate takes `poison_gate` + `live.read`.
736 fn notify_authority_changed(&self) {
737 let subscribers: Vec<FloorsRaisedCallback> = self
738 .subscribers
739 .read()
740 .iter()
741 .map(|(_, callback)| callback.clone())
742 .collect();
743 for callback in subscribers {
744 callback(&[]);
745 }
746 }
747
748 /// Mark this live core's path poisoned, under `poison_gate`, returning
749 /// whether this was the false→true TRANSITION (E3c blockers §1).
750 ///
751 /// The ONLY way production marks a path that has a live core. Calling the
752 /// raw path-registry helper instead would let the mark land inside a
753 /// [`PublicationPin`]'s validate-then-settle window.
754 fn mark_poisoned(&self) -> bool {
755 let _gate = self.lock_poison_gate();
756 mark_poisoned(&self.backing_id, &self.path)
757 }
758
759 /// Clear this live core's path poison, under `poison_gate`.
760 ///
761 /// The inverse of [`Self::mark_poisoned`] and exactly as load-bearing: a pin
762 /// that validated `poisoned == true` and then watched a raw clear land would
763 /// settle `Current` over a reconstruction built for an authority that no
764 /// longer exists (Kyra OLB-2B-E3c closure).
765 ///
766 /// Deliberately does NOT notify: every caller still holds the interprocess
767 /// file lock here. The wake is [`Self::notify_authority_changed`], invoked
768 /// once the caller has released everything.
769 fn clear_poison(&self) {
770 let _gate = self.lock_poison_gate();
771 clear_poison(&self.backing_id, &self.path);
772 }
773
774 /// Acquire `poison_gate` for a poison TRANSITION (mark or clear) — never
775 /// used by [`PublicationPin`], whose acquisition is the thing transitions
776 /// contend with. Try-then-block, with two distinct test acknowledgements
777 /// (E3c blockers §3):
778 ///
779 /// - the BLOCKING hook fires before the acquisition is attempted — the
780 /// placement rendezvous a witness uses to hold a transition at the gate
781 /// while it stages a pin;
782 /// - the CONTENDED hook fires ONLY when the try observed the gate held,
783 /// immediately before blocking. It is the acknowledgement the gap
784 /// witnesses wait on: an ack that fires regardless of contention proves
785 /// only that the contender was scheduled, and a negative assertion
786 /// sequenced after it can pass vacuously under a slow scheduler with the
787 /// protection broken. An ack that required `try_lock` to FAIL proves the
788 /// exclusion was actually met.
789 fn lock_poison_gate(&self) -> MutexGuard<'_, ()> {
790 self.run_poison_blocking_hook();
791 match self.poison_gate.try_lock() {
792 Some(guard) => guard,
793 None => {
794 #[cfg(test)]
795 {
796 let hook = self.poison_contended_hook.lock().clone();
797 if let Some(hook) = hook {
798 hook();
799 }
800 }
801 self.poison_gate.lock()
802 }
803 }
804 }
805
806 /// Fire the poison-gate acknowledgment hook if a test installed one. Runs
807 /// immediately BEFORE the blocking acquisition, so a witness proves a
808 /// contender reached the gate rather than inferring it from elapsed time.
809 fn run_poison_blocking_hook(&self) {
810 #[cfg(test)]
811 {
812 let hook = self.poison_blocking_hook.lock().clone();
813 if let Some(hook) = hook {
814 hook();
815 }
816 }
817 }
818
819 /// Remove the subscriber registered under `token`. Unknown tokens
820 /// are a no-op. Called by [`RaiseSubscription`]'s Drop through a
821 /// `Weak<StoreCore>` (R2-2), so a subscription is retired
822 /// deterministically by dropping its guard — never dependent on a
823 /// facade `Drop` a capture cycle could keep from running.
824 fn remove_subscriber(&self, token: u64) {
825 self.subscribers.write().retain(|(t, _)| *t != token);
826 }
827}
828
829/// Run `mutate` with the POISON GATE of the live core backing `id`, if one
830/// exists.
831///
832/// The construction paths can poison a path BEFORE they have joined its core —
833/// but a sibling handle may already hold one, with pins running against it. This
834/// finds that core through the registry, drops the registry lock (so the gate is
835/// never taken beneath it), and holds only the gate across the mutation. No live
836/// core means no pin can exist, so the raw mutation is already exclusive.
837fn with_live_poison_gate<R>(id: &BackingId, mutate: impl FnOnce() -> R) -> R {
838 let existing = {
839 let guard = core_registry().lock();
840 guard.cores.get(id).and_then(std::sync::Weak::upgrade)
841 };
842 match existing {
843 Some(core) => {
844 let _gate = core.lock_poison_gate();
845 mutate()
846 }
847 None => mutate(),
848 }
849}
850
851/// The exclusion lease shared between one raise subscription's wrapped
852/// callback and its [`RaiseSubscription`] guard (R2-3). It is the
853/// re-entrancy-safe "in-flight lease drained by teardown" variant: the
854/// wrapped callback registers itself as in-flight for the *duration of the
855/// user callback* (never holding the lease's own lock across it, so a
856/// re-entrant `apply_bundle` cannot self-deadlock), and teardown marks the
857/// lease dead and BLOCKS until every in-flight callback has left.
858///
859/// Guarantees, jointly:
860/// - a callback that has passed the liveness check and is mid-mutation
861/// keeps teardown blocked until it finishes (no torn retraction);
862/// - once teardown has marked the lease dead, no *new* callback body runs
863/// — including one already snapshotted by [`StoreCore::notify`] outside
864/// the registry lock, or a re-entrant one.
865struct SubscriptionLease {
866 state: Mutex<LeaseState>,
867 /// Signalled when `in_flight` reaches zero, so a draining teardown
868 /// wakes exactly when the last in-flight callback leaves.
869 drained: Condvar,
870}
871
872struct LeaseState {
873 /// Set once by teardown; gates every subsequent callback entry.
874 dead: bool,
875 /// Count of callback bodies currently executing under this lease.
876 in_flight: usize,
877}
878
879thread_local! {
880 /// Leases whose callback body the CURRENT thread is executing (R3-4).
881 /// Pushed by the wrapped callback on entry, popped on leave. A guard
882 /// dropped from INSIDE its own callback consults this so
883 /// [`SubscriptionLease::kill_and_drain`] does not wait for the very
884 /// frame that is dropping it (which would self-deadlock). Raw pointers
885 /// are only compared for identity and are only ever present while the
886 /// callback holds a live `Arc` to that lease, so there is no
887 /// use-after-free.
888 static ACTIVE_LEASES: std::cell::RefCell<Vec<*const SubscriptionLease>> =
889 const { std::cell::RefCell::new(Vec::new()) };
890}
891
892impl SubscriptionLease {
893 fn new() -> Arc<Self> {
894 Arc::new(Self {
895 state: Mutex::new(LeaseState {
896 dead: false,
897 in_flight: 0,
898 }),
899 drained: Condvar::new(),
900 })
901 }
902
903 /// Enter the callback body: `true` if admitted (caller MUST pair with
904 /// [`Self::leave`]), `false` if the lease is dead (caller returns
905 /// without running the user callback). The lease lock is held only for
906 /// this check-and-count, never across the user callback itself.
907 fn enter(self: &Arc<Self>) -> bool {
908 {
909 let mut st = self.state.lock();
910 if st.dead {
911 return false;
912 }
913 st.in_flight += 1;
914 }
915 // R3-4: record this thread as executing under this lease, so a
916 // self-drop from inside the callback does not drain-wait for its
917 // own frame.
918 let ptr = Arc::as_ptr(self);
919 ACTIVE_LEASES.with(|a| a.borrow_mut().push(ptr));
920 true
921 }
922
923 /// Leave the callback body, waking a draining teardown if this was the
924 /// last in-flight callback.
925 fn leave(self: &Arc<Self>) {
926 let ptr = Arc::as_ptr(self);
927 ACTIVE_LEASES.with(|a| {
928 let mut v = a.borrow_mut();
929 if let Some(i) = v.iter().rposition(|&p| p == ptr) {
930 v.remove(i);
931 }
932 });
933 let mut st = self.state.lock();
934 st.in_flight -= 1;
935 if st.in_flight == 0 {
936 self.drained.notify_all();
937 }
938 }
939
940 /// Teardown: mark the lease dead so no NEW callback body starts, then —
941 /// only when the caller holds none of this lease's own frames — block
942 /// until every in-flight callback has left.
943 ///
944 /// - External teardown (the common case: the guard is dropped from a
945 /// thread that is NOT inside this callback) has `own_frames == 0`, so
946 /// it BLOCKS until `in_flight` reaches zero — the strong guarantee that
947 /// no callback is in flight when the guard's `Drop` returns. `leave`
948 /// signals `drained` at exactly that boundary.
949 /// - Self-unsubscription (the guard is dropped from INSIDE one or more of
950 /// this lease's own callback frames, `own_frames > 0`) does NOT wait at
951 /// all. Waiting would be wrong for two independent reasons (R3-4): this
952 /// thread's own frame(s) cannot reach their `LeaveOnDrop` until this drop
953 /// returns, so waiting for them self-deadlocks; and a callback of the
954 /// SAME lease running on ANOTHER thread may be blocked on a user lock
955 /// THIS callback still holds, so waiting for that foreign frame would
956 /// deadlock across threads. Setting `dead` first stops every new
957 /// (including re-entrant) callback; the current frame's `LeaveOnDrop`
958 /// retires the subscription and each in-flight frame — own and foreign —
959 /// retires through its own `LeaveOnDrop` once it finishes. Return
960 /// immediately: non-blocking and non-reentrant.
961 fn kill_and_drain(self: &Arc<Self>) {
962 let ptr = Arc::as_ptr(self);
963 let own_frames = ACTIVE_LEASES.with(|a| a.borrow().iter().filter(|&&p| p == ptr).count());
964 let mut st = self.state.lock();
965 st.dead = true;
966 if own_frames > 0 {
967 // Self-unsubscription: never wait (see doc — self- and cross-thread
968 // deadlock). `dead` gates new entries; live frames self-retire.
969 return;
970 }
971 while st.in_flight > 0 {
972 self.drained.wait(&mut st);
973 }
974 }
975}
976
977/// An externally-owned RAII handle to one raise subscription (R2-2 +
978/// R2-3). Dropping it:
979/// 1. marks the exclusion lease dead and drains any in-flight callback
980/// (`SubscriptionLease::kill_and_drain`), then
981/// 2. removes the callback from the shared core's registry via a
982/// `Weak<StoreCore>`.
983///
984/// Because removal goes through the `Weak` — not the owning
985/// [`OrgRevocationStore`] facade's `Drop` — a
986/// `core → callback → Arc<store> → core` capture cycle that keeps the
987/// facade alive can no longer strand the callback in the core: whoever
988/// holds this guard (the node, a sibling handle) retires the subscription
989/// by dropping it.
990#[must_use = "dropping the RaiseSubscription immediately unsubscribes and drains the callback"]
991pub struct RaiseSubscription {
992 core: Weak<StoreCore>,
993 token: u64,
994 lease: Arc<SubscriptionLease>,
995}
996
997impl Drop for RaiseSubscription {
998 fn drop(&mut self) {
999 // R2-3: block until no callback observed live is still mutating,
1000 // and stop any snapshotted-but-not-yet-run callback, BEFORE the
1001 // token is removed.
1002 self.lease.kill_and_drain();
1003 // R2-2: retire through the Weak core, independent of the facade.
1004 if let Some(core) = self.core.upgrade() {
1005 core.remove_subscriber(self.token);
1006 }
1007 }
1008}
1009
1010/// A stable identity for a store's backing file, derived from the
1011/// OPENED `.lock` sidecar's inode (AV-9 item 9). The sidecar is created
1012/// once and NEVER renamed — only the state file is rename-replaced by
1013/// `write_atomic`, so the sidecar's inode is stable across every write,
1014/// and two differently-cased path aliases (`revocation-state.json` vs
1015/// `REVOCATION-STATE.JSON`) resolve to the SAME sidecar inode on a
1016/// case-insensitive filesystem. Keying the core and poison registries
1017/// on this — rather than the literal-cased normalized path — collapses
1018/// those aliases to one core (shared live view + publish lock) and one
1019/// poison entry, while `normalize_backing_path` / `open_lock_file`
1020/// still refuse a symlinked or non-regular final component.
1021#[derive(Debug, Clone, PartialEq, Eq, Hash)]
1022enum BackingId {
1023 /// Filesystem file-identity: the unix `(device, inode)` or windows
1024 /// `(volume_serial, file_index)` of the OPENED `.lock` sidecar. Two
1025 /// differently-cased path aliases of one sidecar share this.
1026 FileId { device: u64, inode: u64 },
1027 /// Last-resort key on a platform with no file-identity API (or if a Unix
1028 /// `fstat` fails): the FULL normalized path. R2-4 — never a lossy 64-bit
1029 /// path hash, so two distinct paths can NEVER collide onto one core or
1030 /// poison entry (the pre-R2-4 `DefaultHasher` fallback could, at the
1031 /// ~2^32 birthday bound). Never constructed on Windows: a missing file
1032 /// identity there fails loud (see [`BackingId::of`]) rather than degrading
1033 /// to a case-sensitive literal path.
1034 #[cfg_attr(windows, allow(dead_code))]
1035 Path(PathBuf),
1036}
1037
1038impl BackingId {
1039 /// Derive the identity from the opened lock sidecar. `path` (already
1040 /// normalized by the caller) backs the fallback key on a platform with
1041 /// no file-identity API, or if a Unix `fstat` somehow fails.
1042 ///
1043 /// On Windows the identity comes from the stable `GetFileInformationByHandle`
1044 /// Win32 call: the `std` `MetadataExt::{volume_serial_number, file_index}`
1045 /// accessors require the unstable `windows_by_handle` feature and do NOT
1046 /// build on stable Rust. A Windows identity read that fails is a LOUD
1047 /// error — never a silent degradation to a case-sensitive literal path,
1048 /// which would reopen AV-9 by letting two differently-cased aliases of one
1049 /// sidecar key two DISTINCT cores.
1050 fn of(lock: &std::fs::File, path: &Path) -> Result<Self, OrgRevocationError> {
1051 #[cfg(unix)]
1052 {
1053 use std::os::unix::fs::MetadataExt;
1054 // `fstat` on an open fd effectively never fails; degrade to the
1055 // full-path fallback key (R2-4) if it somehow does.
1056 Ok(match lock.metadata() {
1057 Ok(meta) => BackingId::FileId {
1058 device: meta.dev(),
1059 inode: meta.ino(),
1060 },
1061 Err(_) => BackingId::Path(path.to_path_buf()),
1062 })
1063 }
1064 #[cfg(windows)]
1065 {
1066 // Stable Win32 `(dwVolumeSerialNumber, nFileIndex)` identity; a
1067 // read failure fails loud rather than degrading to a literal path.
1068 match windows_file_identity(lock) {
1069 Ok((device, inode, _links)) => Ok(BackingId::FileId { device, inode }),
1070 Err(e) => Err(OrgRevocationError::Io {
1071 path: path.display().to_string(),
1072 reason: format!("state lock: cannot read Windows file identity: {e}"),
1073 }),
1074 }
1075 }
1076 // Fallback key (R2-4: the FULL normalized path, never a lossy 64-bit
1077 // hash) on a platform with no file-identity API.
1078 #[cfg(not(any(unix, windows)))]
1079 {
1080 let _ = lock;
1081 Ok(BackingId::Path(path.to_path_buf()))
1082 }
1083 }
1084}
1085
1086/// Read the stable Win32 `BY_HANDLE_FILE_INFORMATION` for an open handle and
1087/// return `(volume serial, file index, hard-link count)`.
1088///
1089/// `std`'s equivalents (`MetadataExt::volume_serial_number` / `file_index`,
1090/// and any link count at all) require the unstable `windows_by_handle`
1091/// feature and do not build on stable Rust, and there is no Win32 bindings
1092/// crate in this workspace — so we declare the one call we need directly, the
1093/// same hand-rolled `extern "system"` idiom the crate already uses elsewhere.
1094#[cfg(windows)]
1095fn windows_file_identity(file: &std::fs::File) -> std::io::Result<(u64, u64, u32)> {
1096 use std::os::windows::io::AsRawHandle;
1097
1098 // `BY_HANDLE_FILE_INFORMATION`; `FILETIME` is two `DWORD`s. `#[repr(C)]`
1099 // so the field offsets match the Win32 ABI exactly.
1100 #[repr(C)]
1101 #[derive(Default)]
1102 struct ByHandleFileInformation {
1103 dw_file_attributes: u32,
1104 ft_creation_time: [u32; 2],
1105 ft_last_access_time: [u32; 2],
1106 ft_last_write_time: [u32; 2],
1107 dw_volume_serial_number: u32,
1108 n_file_size_high: u32,
1109 n_file_size_low: u32,
1110 n_number_of_links: u32,
1111 n_file_index_high: u32,
1112 n_file_index_low: u32,
1113 }
1114
1115 extern "system" {
1116 fn GetFileInformationByHandle(
1117 h_file: *mut std::ffi::c_void,
1118 lp_file_information: *mut ByHandleFileInformation,
1119 ) -> i32;
1120 }
1121
1122 let mut info = ByHandleFileInformation::default();
1123 // SAFETY: `file` owns a valid, open handle for the duration of this call,
1124 // and `info` is a live, correctly-sized, writable output buffer.
1125 // `as_raw_handle()` is already `*mut c_void` — the exact `h_file` type.
1126 let ok = unsafe { GetFileInformationByHandle(file.as_raw_handle(), &mut info) };
1127 if ok == 0 {
1128 return Err(std::io::Error::last_os_error());
1129 }
1130 let volume = u64::from(info.dw_volume_serial_number);
1131 let index = (u64::from(info.n_file_index_high) << 32) | u64::from(info.n_file_index_low);
1132 Ok((volume, index, info.n_number_of_links))
1133}
1134
1135/// Process-wide core registry (AV-9 + R2-4). `cores` maps a backing
1136/// file's stable [`BackingId`] to its live core (`Weak`, so a backing
1137/// file whose handles all dropped releases its core; the POISON registry
1138/// is separate precisely because poison must outlive every handle).
1139///
1140/// `bindings` (R2-4) maps a normalized backing PATH to the sidecar
1141/// identity currently bound to it. It is GC'd in lockstep with dead cores
1142/// (a binding whose id has no live core is dropped), so a *surviving*
1143/// binding always names a LIVE core — a path resolving to a different
1144/// identity while its binding survives means the sidecar was recreated or
1145/// replaced under a still-held core, which
1146/// [`join_or_create_core`] refuses loudly.
1147struct CoreRegistry {
1148 cores: std::collections::HashMap<BackingId, std::sync::Weak<StoreCore>>,
1149 bindings: std::collections::HashMap<PathBuf, BackingId>,
1150}
1151
1152static CORES: std::sync::OnceLock<Mutex<CoreRegistry>> = std::sync::OnceLock::new();
1153
1154fn core_registry() -> &'static Mutex<CoreRegistry> {
1155 CORES.get_or_init(|| {
1156 Mutex::new(CoreRegistry {
1157 cores: std::collections::HashMap::new(),
1158 bindings: std::collections::HashMap::new(),
1159 })
1160 })
1161}
1162
1163/// Join the existing core for `path`, republishing the state just
1164/// reread from disk through it (a same-path sibling's live view
1165/// advances BEFORE any poison clears — review-9 addendum), or
1166/// create a fresh core seeded with that state. The caller MUST
1167/// hold the interprocess state lock, which is what makes the
1168/// reread current.
1169///
1170/// The republish through an EXISTING core takes that core's
1171/// `reload` lock (review-11 P1): every `StoreCore::publish` — not
1172/// only `apply_bundle`'s — must hold `reload`, or a replacement
1173/// holding [`PublishGuard`] could be racing an opener that
1174/// publishes a stronger floor between the guard's dominance
1175/// comparison and its swap. The canonical order is interprocess
1176/// file lock (already held by the caller) OUTER, `reload` INNER;
1177/// [`OrgRevocationStore::apply_bundle`] obeys the same order, and
1178/// a replacement holds only `reload` (never the file lock), so no
1179/// cycle exists. The registry lock is released before `reload` is
1180/// acquired so no `registry → reload` nesting can form.
1181fn join_or_create_core(
1182 backing_id: BackingId,
1183 path: &Path,
1184 disk: OrgRevocationState,
1185) -> Result<(Arc<StoreCore>, Vec<RaisedFloor>), OrgRevocationError> {
1186 let mut guard = core_registry().lock();
1187 // Reborrow as `&mut CoreRegistry` so disjoint field borrows (immutable
1188 // `cores`, mutable `bindings`) are allowed — a `MutexGuard`'s Deref
1189 // would otherwise borrow the whole guard.
1190 let reg = &mut *guard;
1191 // GC dead cores AND the bindings that named them, in lockstep: after
1192 // this, every surviving binding points at a LIVE core.
1193 reg.cores.retain(|_, weak| weak.strong_count() > 0);
1194 let live_ids = ®.cores;
1195 reg.bindings.retain(|_, id| live_ids.contains_key(id));
1196 // R2-4 binding check: if this path is already bound (to a still-live
1197 // core) under a DIFFERENT sidecar identity, the sidecar was recreated
1198 // or replaced underneath that core — refuse loudly rather than fork
1199 // the path into two independent security views. A legitimate
1200 // recreation (the old core fully dropped) left no surviving binding,
1201 // so it falls through and rebinds.
1202 if let Some(bound) = reg.bindings.get(path) {
1203 if *bound != backing_id {
1204 return Err(OrgRevocationError::BackingIdentityConflict {
1205 path: path.display().to_string(),
1206 });
1207 }
1208 }
1209 reg.bindings.insert(path.to_path_buf(), backing_id.clone());
1210 let existing = reg
1211 .cores
1212 .get(&backing_id)
1213 .and_then(std::sync::Weak::upgrade);
1214 if let Some(core) = existing {
1215 drop(guard);
1216 let raised = {
1217 let _reload = core.reload.lock();
1218 core.publish(disk)
1219 };
1220 return Ok((core, raised));
1221 }
1222 // Fresh core: nobody else can observe it until we insert, so
1223 // its first publish races nothing. Hold the registry lock
1224 // across the check-and-insert so two openers cannot both create.
1225 let core = Arc::new(StoreCore {
1226 path: path.to_path_buf(),
1227 backing_id: backing_id.clone(),
1228 reload: Mutex::new(()),
1229 live: RwLock::new(Arc::new(disk)),
1230 generation: AtomicU64::new(0),
1231 generation_exhausted: AtomicBool::new(false),
1232 poison_gate: Mutex::new(()),
1233 #[cfg(test)]
1234 publish_contended_hook: Mutex::new(None),
1235 #[cfg(test)]
1236 poison_blocking_hook: Mutex::new(None),
1237 #[cfg(test)]
1238 force_post_rename: AtomicBool::new(false),
1239 #[cfg(test)]
1240 poison_contended_hook: Mutex::new(None),
1241 subscribers: RwLock::new(Vec::new()),
1242 next_subscriber: AtomicU64::new(0),
1243 #[cfg(any(test, feature = "fixtures"))]
1244 publish_pause: parking_lot::Mutex::new(None),
1245 });
1246 reg.cores.insert(backing_id, Arc::downgrade(&core));
1247 Ok((core, Vec::new()))
1248}
1249
1250/// Exclusive guard over one or two stores' publish transactions.
1251/// While held, no reload can publish a new live view through the
1252/// guarded core(s) — from ANY same-path handle, including an
1253/// opener joining the core (review-11 P1). The node holds this
1254/// across its replacement dominance comparison and swap, and
1255/// across authority verification and publication, so the installed
1256/// floor view cannot rise between a check and the publication that
1257/// depends on it.
1258///
1259/// When two DISTINCT cores must be pinned (a cross-core store
1260/// replacement or authority install — the topology review-10
1261/// supports), [`publish_guard_pair`] acquires their `reload` locks
1262/// in a canonical order (normalized path order) so two nodes
1263/// performing opposite swaps cannot deadlock ABBA (review-11 P1).
1264/// Callbacks are never invoked under this guard (raises notify
1265/// outside the reload lock), so holding it cannot deadlock against
1266/// notification work.
1267pub(crate) struct PublishGuard<'a> {
1268 _guards: Vec<parking_lot::MutexGuard<'a, ()>>,
1269}
1270
1271/// Pin BOTH stores' publish transactions in a canonical, ABBA-free
1272/// order (review-11 P1). Same-core stores dedup to a single lock
1273/// (parking_lot mutexes are not reentrant, so locking one core
1274/// twice would self-deadlock). Distinct cores lock in normalized
1275/// path order, so every caller that pins the same two cores — from
1276/// any node — acquires them in the same sequence.
1277pub(crate) fn publish_guard_pair<'a>(
1278 a: &'a OrgRevocationStore,
1279 b: &'a OrgRevocationStore,
1280) -> PublishGuard<'a> {
1281 if Arc::ptr_eq(&a.core, &b.core) {
1282 return PublishGuard {
1283 _guards: vec![a.core.reload.lock()],
1284 };
1285 }
1286 let (first, second) = if a.core.path <= b.core.path {
1287 (a, b)
1288 } else {
1289 (b, a)
1290 };
1291 let g1 = first.core.reload.lock();
1292 let g2 = second.core.reload.lock();
1293 PublishGuard {
1294 _guards: vec![g1, g2],
1295 }
1296}
1297/// Holds this store's authority IMMOBILE.
1298///
1299/// While alive, both floor publication and poison transitions are blocked:
1300/// publication needs `live.write()`, and every poison transition on a live core
1301/// takes `poison_gate`. That is what lets a consumer validate and then act as ONE
1302/// operation rather than two interleavable steps (Kyra OLB-2B-E3c).
1303pub struct PublicationPin<'a> {
1304 core: &'a StoreCore,
1305 _poison: parking_lot::MutexGuard<'a, ()>,
1306 _live: parking_lot::RwLockReadGuard<'a, Arc<OrgRevocationState>>,
1307}
1308
1309impl PublicationPin<'_> {
1310 /// The generation this pin is holding still, or `Err` if exhausted.
1311 pub fn generation(&self) -> Result<BarrieredGeneration, GenerationExhausted> {
1312 OrgRevocationStore::sample_generation(self.core)
1313 }
1314
1315 /// Live poison, re-read through the guard.
1316 pub fn poisoned(&self) -> bool {
1317 is_poisoned(&self.core.backing_id, &self.core.path)
1318 }
1319}
1320
1321/// A publication generation sampled coherently with its terminal state.
1322///
1323/// Deliberately opaque. The raw `u64` is only meaningful next to the exhaustion
1324/// latch it was sampled with, so handing out a bare integer invites exactly the
1325/// aliasing this type exists to prevent — a consumer comparing frozen values and
1326/// concluding "unchanged". Compare these, do not unwrap them.
1327#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
1328pub struct BarrieredGeneration(u64);
1329
1330impl BarrieredGeneration {
1331 /// The raw value. For stamping and logging only — a currentness DECISION
1332 /// must compare `BarrieredGeneration`s, so the exhaustion latch they were
1333 /// sampled with cannot be dropped on the floor.
1334 pub fn get(self) -> u64 {
1335 self.0
1336 }
1337
1338 /// Test-only: fabricate a generation for stamp-comparison unit tests that
1339 /// never touch a real store.
1340 #[doc(hidden)]
1341 #[cfg(any(test, feature = "fixtures"))]
1342 pub fn from_raw_for_test(raw: u64) -> Self {
1343 Self(raw)
1344 }
1345}
1346
1347/// The publication generation space is exhausted: the counter is frozen, so it
1348/// can no longer distinguish floor views.
1349///
1350/// TERMINAL and fail-closed. Every consumer that uses the generation as a
1351/// currentness discriminator must refuse rather than proceed — a frozen counter
1352/// makes a post-exhaustion publication look identical to no publication at all
1353/// (Kyra OLB-2B-E3c). Returned as an `Err` precisely so no consumer can ignore
1354/// it by accident.
1355#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1356pub struct GenerationExhausted;
1357
1358impl std::fmt::Display for GenerationExhausted {
1359 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1360 f.write_str("org revocation publication generation space is exhausted")
1361 }
1362}
1363
1364impl std::error::Error for GenerationExhausted {}
1365
1366/// The node-local persisted revocation maxima plus its published
1367/// live view. See the module docs for the locked reload order and
1368/// failure semantics.
1369///
1370/// The store is org-agnostic (keys are `(OrgId, EntityId)`); WHICH
1371/// bundles get fed to [`Self::apply_bundle`] is the caller's trust
1372/// decision — in OA-1 the adopt/startup wiring feeds only the
1373/// node's owner-org bundle.
1374///
1375/// # Multi-writer safety (review-8 §5)
1376///
1377/// Same-file writers (a second store instance, a concurrent
1378/// `net node adopt`) are serialized through an exclusive advisory
1379/// lock on a stable `.lock` sidecar, and every reload REREADS the
1380/// persisted maxima under that lock before merging — an instance's
1381/// in-memory snapshot is never trusted as the merge base, so a
1382/// stale writer cannot roll another writer's floors out of the
1383/// file. Because every writer follows reread-merge-write, the disk
1384/// state only ever grows, and republishing the reread state can
1385/// never lower a live view.
1386///
1387/// Within one process, same-path handles additionally share one
1388/// `StoreCore` — one live view, one publish transaction, one
1389/// subscriber registry (review-9 addendum). See the module docs.
1390pub struct OrgRevocationStore {
1391 /// The shared per-path core.
1392 ///
1393 /// R3-4: the facade holds NO subscription of its own. A raise
1394 /// subscription is always owned EXTERNALLY through the
1395 /// [`RaiseSubscription`] guard returned by
1396 /// [`Self::subscribe_floors_raised`] — the node's install path holds
1397 /// it, a test holds it. The removed `set_on_floors_raised` stored its
1398 /// guard inside the facade, which a callback capturing `Arc<Self>`
1399 /// (`core → callback → Arc<store> → own_subscription → …`) could keep
1400 /// alive forever, so the facade's own drop never ran and the callback
1401 /// leaked. Whoever holds the external guard breaks that cycle by
1402 /// dropping it.
1403 core: Arc<StoreCore>,
1404}
1405
1406/// Caller-supplied evidence about whether a revocation state file is expected
1407/// to exist already — the difference between a first-ever adopt and a lost
1408/// state file, which are IDENTICAL on disk from this module's point of view.
1409///
1410/// This exists because "absence" is the one input the store cannot interpret on
1411/// its own, and the two readings sit at opposite ends of the safety spectrum:
1412/// creating an empty state on a genuine fresh adopt is correct, and creating
1413/// one because the file went missing silently discards every floor.
1414#[derive(Clone, Copy, PartialEq, Eq, Debug)]
1415pub enum ProvisioningExpectation {
1416 /// Nothing else at this authority path implies prior provisioning, so a
1417 /// missing state file is a first adopt and is created empty.
1418 MayBeFresh,
1419 /// Other authority artifacts are already present, so this store MUST exist
1420 /// too. A missing state file is data loss and is refused.
1421 MustExist,
1422}
1423
1424impl OrgRevocationStore {
1425 /// Adopt-time entry point: load the existing file if present
1426 /// (re-adoption preserves maxima — monotonicity survives even
1427 /// an operator re-running adopt), otherwise durably create an
1428 /// empty state file. Runs under the interprocess lock so two
1429 /// concurrent adoptions cannot race the create.
1430 ///
1431 /// # Absence is not automatically "fresh"
1432 ///
1433 /// This entry point and [`Self::open_existing`] used to read a missing
1434 /// state file in exactly OPPOSITE ways: `open_existing` raised
1435 /// [`OrgRevocationError::MissingState`] and refused, while `init` wrote an
1436 /// empty state and continued — silently resetting every floor, and thereby
1437 /// re-admitting every membership certificate the org had revoked.
1438 ///
1439 /// The permissive reading sat on the path an operator reaches for when
1440 /// something already looks wrong (`net node adopt`); the strict one sat on
1441 /// the path that merely reopens. And if a same-path core was still live,
1442 /// `publish`'s per-key max kept the RUNNING node enforcing the old floors,
1443 /// so the rollback did not surface until the next restart.
1444 ///
1445 /// Two independent signals now have to agree before an empty state is
1446 /// created:
1447 ///
1448 /// 1. `expect`, supplied by the caller, which can see the rest of the
1449 /// authority directory (a node holding a membership certificate has
1450 /// demonstrably been provisioned before);
1451 /// 2. the `.lock` sidecar, which is created beside the state file and
1452 /// survives its deletion. It is probed BEFORE `lock_state_file` can
1453 /// create it — the previous ordering destroyed exactly the evidence
1454 /// needed here.
1455 ///
1456 /// Residual, deliberately not closed: deleting BOTH the state file and its
1457 /// sidecar is indistinguishable from a fresh adopt at this layer. Signal 1
1458 /// is what covers that case, which is why it is a caller obligation and not
1459 /// merely an internal check.
1460 pub fn init(
1461 path: impl Into<PathBuf>,
1462 expect: ProvisioningExpectation,
1463 ) -> Result<Self, OrgRevocationError> {
1464 let path = path.into();
1465 if let Some(parent) = path.parent() {
1466 if !parent.as_os_str().is_empty() {
1467 std::fs::create_dir_all(parent).map_err(|e| OrgRevocationError::Io {
1468 path: path.display().to_string(),
1469 reason: e.to_string(),
1470 })?;
1471 }
1472 }
1473 let path = normalize_backing_path(&path)?;
1474 // Probe the sidecar BEFORE `lock_state_file` can create it. Its
1475 // presence beside a MISSING state file means this store was
1476 // provisioned and its state has since been removed — the one piece of
1477 // evidence that distinguishes loss from a first adopt, and the
1478 // previous ordering unconditionally `create(true)`'d it away before
1479 // anything could look.
1480 let sidecar_predates_us = {
1481 let mut lock_path = path.as_os_str().to_os_string();
1482 lock_path.push(".lock");
1483 std::fs::symlink_metadata(PathBuf::from(lock_path)).is_ok()
1484 };
1485 let lock = lock_state_file(&path)?;
1486 // AV-9: identity is the stable `.lock` inode, so case-aliases
1487 // share one core + poison entry.
1488 let backing_id = BackingId::of(&lock, &path)?;
1489 let was_poisoned = is_poisoned(&backing_id, &path);
1490 if was_poisoned {
1491 prove_entry_durable(&path)?;
1492 }
1493 let state = match read_regular_nofollow(&path) {
1494 Ok(bytes) => OrgRevocationState::from_file_bytes(&bytes, &path)?,
1495 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
1496 // Fail CLOSED when either signal says this store already
1497 // existed. Creating an empty state here would durably reset
1498 // every floor and re-admit every revoked certificate.
1499 if expect == ProvisioningExpectation::MustExist || sidecar_predates_us {
1500 let err = OrgRevocationError::MissingState {
1501 path: path.display().to_string(),
1502 };
1503 tracing::error!(
1504 sidecar_predates_us,
1505 ?expect,
1506 "{err}; refusing to re-create it as EMPTY — that would \
1507 discard every revocation floor and re-admit every \
1508 certificate this org has revoked. Restore the state \
1509 file from backup, or remove the whole authority \
1510 directory to provision deliberately from scratch."
1511 );
1512 return Err(err);
1513 }
1514 // §18 — poison on a post-rename durability failure, exactly as
1515 // `apply_bundle` does.
1516 //
1517 // `write_atomic` maps `PostRename` to `DurabilityUncertain` but
1518 // omits the `mark_poisoned` call, and its docstring scopes it to
1519 // "callers whose files carry no published live view". The state
1520 // file is precisely the file that DOES carry one: this is the
1521 // path that creates it. Without the poison a retry, or a
1522 // same-process `open_existing`, sees a clean path and proceeds
1523 // over a directory entry that was never proven durable.
1524 let state = OrgRevocationState::empty();
1525 match write_atomic_phased(&path, &state.to_file_bytes()?) {
1526 Ok(()) => {}
1527 Err(WritePhase::PreRename(reason)) => {
1528 return Err(OrgRevocationError::Io {
1529 path: path.display().to_string(),
1530 reason,
1531 })
1532 }
1533 Err(WritePhase::PostRename(reason)) => {
1534 // No core joined yet, but a same-path sibling may hold
1535 // one: gate against ITS pins (Kyra OLB-2B-E3c closure).
1536 // No authority wake here, deliberately (contrast
1537 // `apply_bundle`, E3c blockers §1): this store failed
1538 // to initialize, so it has no subscribers, and the
1539 // interprocess lock is still held — a sibling core's
1540 // routing facts are repaired by the lazy read-time
1541 // epoch comparison, which is the accepted coverage for
1542 // this create-over-a-live-sibling corner.
1543 with_live_poison_gate(&backing_id, || {
1544 let _ = mark_poisoned(&backing_id, &path);
1545 });
1546 return Err(OrgRevocationError::DurabilityUncertain {
1547 path: path.display().to_string(),
1548 reason,
1549 });
1550 }
1551 }
1552 state
1553 }
1554 Err(e) => {
1555 return Err(OrgRevocationError::Io {
1556 path: path.display().to_string(),
1557 reason: e.to_string(),
1558 })
1559 }
1560 };
1561 let (core, raised) = join_or_create_core(backing_id.clone(), &path, state)?;
1562 if was_poisoned {
1563 // Through the CORE, under its poison gate: a sibling handle's
1564 // routing pin may be mid-settlement against `poisoned == true`.
1565 core.clear_poison();
1566 }
1567 drop(lock);
1568 let store = Self { core };
1569 store.core.notify(&raised);
1570 if was_poisoned {
1571 // Recovery raises no floor, so `notify` alone is silent — yet the
1572 // authority just moved from "unusable" back to usable.
1573 store.core.notify_authority_changed();
1574 }
1575 Ok(store)
1576 }
1577
1578 /// Startup entry point: the file MUST exist and parse. Missing
1579 /// or corrupt → loud typed error; protected verification never
1580 /// starts against silently weaker floors.
1581 ///
1582 /// The open ALWAYS serializes behind the interprocess state
1583 /// lock — there is no pre-lock poison fast path (review-9
1584 /// addendum): a writer holding the lock may be mid-rename, so
1585 /// an opener must wait and read the FINAL state, and a poison
1586 /// bit registered while it waited must gate it. If the path is
1587 /// durability-poisoned, the open performs explicit recovery
1588 /// under that lock — a successful parent-directory fsync plus
1589 /// the reread republished through the shared per-path core
1590 /// (every live sibling advances) — BEFORE the poison clears;
1591 /// recovery failure refuses the open. A fresh instance
1592 /// therefore never launders path-wide uncertainty.
1593 pub fn open_existing(path: impl Into<PathBuf>) -> Result<Self, OrgRevocationError> {
1594 let path = normalize_backing_path(&path.into())?;
1595 let lock = lock_state_file(&path)?;
1596 // AV-9: stable `.lock` inode identity (case-aliases collapse).
1597 let backing_id = BackingId::of(&lock, &path)?;
1598 let was_poisoned = is_poisoned(&backing_id, &path);
1599 if was_poisoned {
1600 prove_entry_durable(&path)?;
1601 }
1602 let bytes = match read_regular_nofollow(&path) {
1603 Ok(bytes) => bytes,
1604 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
1605 let err = OrgRevocationError::MissingState {
1606 path: path.display().to_string(),
1607 };
1608 tracing::error!("{err}");
1609 return Err(err);
1610 }
1611 Err(e) => {
1612 return Err(OrgRevocationError::Io {
1613 path: path.display().to_string(),
1614 reason: e.to_string(),
1615 })
1616 }
1617 };
1618 let state = OrgRevocationState::from_file_bytes(&bytes, &path).inspect_err(|err| {
1619 tracing::error!("{err}");
1620 })?;
1621 let (core, raised) = join_or_create_core(backing_id.clone(), &path, state)?;
1622 if was_poisoned {
1623 core.clear_poison();
1624 }
1625 drop(lock);
1626 let store = Self { core };
1627 store.core.notify(&raised);
1628 if was_poisoned {
1629 store.core.notify_authority_changed();
1630 }
1631 Ok(store)
1632 }
1633
1634 /// The backing file path (normalized at construction).
1635 pub fn path(&self) -> &Path {
1636 &self.core.path
1637 }
1638
1639 /// Snapshot of the published live view.
1640 pub fn snapshot(&self) -> Arc<OrgRevocationState> {
1641 self.core.live.read().clone()
1642 }
1643
1644 /// Live floor for `(org, member)`.
1645 pub fn floor_for(&self, org: &OrgId, member: &EntityId) -> u32 {
1646 self.snapshot().floor_for(org, member)
1647 }
1648
1649 /// `true` while this store's BACKING PATH is
1650 /// durability-uncertain (review-9: the poison bit is shared by
1651 /// every instance on the same normalized pathname, not held
1652 /// per object). Cleared only by explicit recovery — a locked
1653 /// reread republished through the shared core plus a
1654 /// successful parent-directory fsync — performed by
1655 /// [`Self::open_existing`], [`Self::init`], or the next
1656 /// [`Self::apply_bundle`].
1657 pub fn is_poisoned(&self) -> bool {
1658 is_poisoned(&self.core.backing_id, &self.core.path)
1659 }
1660
1661 /// Whether the publication generation space is exhausted.
1662 ///
1663 /// OBSERVABILITY ONLY. Never use this to decide currentness: read
1664 /// independently of the generation it qualifies, it races the very
1665 /// publication that exhausts the space. Currentness decisions must take the
1666 /// coherent [`Result`] from [`Self::barriered_generation`] or
1667 /// [`Self::snapshot_with_generation`], which sample both under one barrier
1668 /// (Kyra OLB-2B-E3c).
1669 pub fn generation_exhausted_for_metrics(&self) -> bool {
1670 self.core.generation_exhausted.load(Ordering::Acquire)
1671 }
1672
1673 /// Test-only: drive the publication generation to its ceiling, so a witness
1674 /// can exercise the exhaustion branch without 2^64 real publications.
1675 #[doc(hidden)]
1676 #[cfg(any(test, feature = "fixtures"))]
1677 pub fn saturate_generation_for_test(&self) {
1678 self.core.generation.store(u64::MAX, Ordering::Release);
1679 }
1680
1681 /// Test-only: arm the CONTENDED acknowledgment for `live` — fired only after
1682 /// a publish's `try_write` has failed, so a witness can prove the barrier was
1683 /// actually held rather than that a publisher was merely scheduled.
1684 #[doc(hidden)]
1685 #[cfg(test)]
1686 pub(crate) fn arm_publish_contended_hook(&self, hook: Arc<dyn Fn() + Send + Sync>) {
1687 *self.core.publish_contended_hook.lock() = Some(hook);
1688 }
1689
1690 /// Test-only: force one publication of the current view.
1691 #[doc(hidden)]
1692 #[cfg(any(test, feature = "fixtures"))]
1693 pub fn republish_for_test(&self) {
1694 let current = (*self.core.live.read()).as_ref().clone();
1695 self.core.publish(current);
1696 }
1697
1698 /// Test-only: mark this store's backing path poisoned so
1699 /// [`Self::is_poisoned`] returns true, without forcing a real
1700 /// fsync failure. Lets a witness exercise the
1701 /// durability-uncertain admission-denial branch. `#[doc(hidden)]`
1702 /// (matching the review-9/11 `*_for_test` seams) so integration
1703 /// tests in a separate crate can reach it; never used in
1704 /// production paths.
1705 #[doc(hidden)]
1706 #[cfg(any(test, feature = "fixtures"))]
1707 pub fn mark_poisoned_for_test(&self) {
1708 self.core.mark_poisoned();
1709 }
1710
1711 /// Test-only: arm the pre-`poison_gate` PLACEMENT hook — fired before a
1712 /// poison transition attempts the acquisition, whether or not the gate is
1713 /// held. Fires for marks AND clears, including the ones the production
1714 /// recovery paths perform. A rendezvous point, NOT contention evidence:
1715 /// for that, arm [`Self::arm_poison_contended_hook`] (E3c blockers §3).
1716 #[doc(hidden)]
1717 #[cfg(test)]
1718 pub(crate) fn arm_poison_blocking_hook(&self, hook: Arc<dyn Fn() + Send + Sync>) {
1719 *self.core.poison_blocking_hook.lock() = Some(hook);
1720 }
1721
1722 /// Test-only: arm the poison-gate CONTENTION acknowledgment — fired only
1723 /// when a transition's `try_lock` observed the gate held, immediately
1724 /// before it blocks (E3c blockers §3). This is the ack the gap witnesses
1725 /// sequence their negative assertions after: it proves the exclusion was
1726 /// met, not merely that the contender got scheduled.
1727 #[doc(hidden)]
1728 #[cfg(test)]
1729 pub(crate) fn arm_poison_contended_hook(&self, hook: Arc<dyn Fn() + Send + Sync>) {
1730 *self.core.poison_contended_hook.lock() = Some(hook);
1731 }
1732
1733 /// Test-only, one-shot: force the next `apply_bundle` state write to report
1734 /// a POST-rename durability failure without touching the file (E3c
1735 /// blockers §1). See the [`StoreCore::force_post_rename`] field.
1736 #[doc(hidden)]
1737 #[cfg(test)]
1738 pub(crate) fn arm_forced_post_rename_for_test(&self) {
1739 self.core.force_post_rename.store(true, Ordering::Release);
1740 }
1741
1742 /// `true` iff `other` is backed by the same normalized path —
1743 /// i.e. shares this store's core (live view, publish lock,
1744 /// subscribers).
1745 pub fn shares_core_with(&self, other: &Self) -> bool {
1746 Arc::ptr_eq(&self.core, &other.core)
1747 }
1748
1749 /// The core's publish generation — bumped once per published
1750 /// live view. Lets callers order publications relative to
1751 /// their own critical sections.
1752 ///
1753 /// A BARE atomic load: it does NOT cross the live-view lock, so a
1754 /// publication in progress (view already swapped under
1755 /// `live.write()`, generation not yet bumped) is observed as the
1756 /// OLD generation. Admission stamping must use
1757 /// [`Self::barriered_generation`] / [`Self::snapshot_with_generation`]
1758 /// instead — see their docs (OA2-E1 Kyra review).
1759 ///
1760 /// Demoted from `pub` and marked dead-code-tolerant rather than deleted
1761 /// (review-pass-3 §12): it has no production caller — which is exactly why it
1762 /// was dangerous to export, since the only use a caller could invent for a
1763 /// bare unbarriered `u64` with no exhaustion signal is the currentness
1764 /// decision the paragraph above forbids. It survives as the contrast case the
1765 /// barrier witnesses assert against.
1766 #[allow(dead_code)]
1767 pub(crate) fn publish_generation(&self) -> u64 {
1768 self.core.generation.load(Ordering::Acquire)
1769 }
1770
1771 /// The publish generation read UNDER a `live.read()` barrier
1772 /// (OA2-E1 Kyra review). `StoreCore::publish` swaps the live
1773 /// view and bumps the generation while holding `live.write()`, so
1774 /// acquiring a read guard first guarantees no publication is
1775 /// mid-flight: the returned generation always matches the
1776 /// currently-visible view. Unlike `publish_generation`,
1777 /// this can never return an old generation while a raised floor is
1778 /// already installed — the interleaving that would let a stale
1779 /// admission stamp compare "unchanged" and admit against a floor
1780 /// that has actually risen.
1781 pub fn barriered_generation(&self) -> Result<BarrieredGeneration, GenerationExhausted> {
1782 let _live = self.core.live.read();
1783 Self::sample_generation(&self.core)
1784 }
1785
1786 /// Hold the publication barrier: while this guard lives, NO floor
1787 /// publication can land, because `StoreCore::publish` needs `live.write()`.
1788 ///
1789 /// This is real EXCLUSION, not observation. A consumer whose decision is
1790 /// itself load-bearing — the routing commit pin, whose `Current` causes the
1791 /// supervisor to publish `Healthy` — cannot be made sound by sampling and
1792 /// then acting: a publication landing between the sample and the decision
1793 /// makes the decision false with nothing to detect it (Kyra OLB-2B-E3c).
1794 /// Holding the barrier through the decision closes that gap by construction.
1795 ///
1796 /// Subscribers are invoked OUTSIDE the publish locks, so a blocked publisher
1797 /// cannot deadlock against a callback that takes the routing authority gate.
1798 /// Hold it briefly and never across an await.
1799 pub fn pin_publication(&self) -> PublicationPin<'_> {
1800 // Poison gate BEFORE the view barrier: `apply_bundle` marks poison and
1801 // only then publishes, so the reverse order would deadlock against it.
1802 let poison = self.core.poison_gate.lock();
1803 let live = self.core.live.read();
1804 PublicationPin {
1805 core: &self.core,
1806 _poison: poison,
1807 _live: live,
1808 }
1809 }
1810
1811 /// Sample the generation and its terminal state under ONE `live.read()`.
1812 ///
1813 /// The caller must already hold that guard. Sampling them with two
1814 /// independent calls is itself raceable: the publication that exhausts the
1815 /// space sets the latch and freezes the counter, so a reader can observe the
1816 /// frozen counter with the pre-exhaustion latch and conclude "unchanged"
1817 /// (Kyra OLB-2B-E3c).
1818 fn sample_generation(core: &StoreCore) -> Result<BarrieredGeneration, GenerationExhausted> {
1819 if core.generation_exhausted.load(Ordering::Acquire) {
1820 return Err(GenerationExhausted);
1821 }
1822 Ok(BarrieredGeneration(core.generation.load(Ordering::Acquire)))
1823 }
1824
1825 /// A floor snapshot together with the exact generation it
1826 /// reflects, both read under ONE `live.read()` guard (OA2-E1
1827 /// Kyra review). Publication-barriered like
1828 /// [`Self::barriered_generation`], so the `(snapshot, generation)`
1829 /// pair is always consistent — no seqlock retry needed.
1830 pub fn snapshot_with_generation(
1831 &self,
1832 ) -> Result<(Arc<OrgRevocationState>, BarrieredGeneration), GenerationExhausted> {
1833 let live = self.core.live.read();
1834 let generation = Self::sample_generation(&self.core)?;
1835 Ok((live.clone(), generation))
1836 }
1837
1838 /// Test-only (`#[doc(hidden)]`, mirroring the review-11
1839 /// `*_paused_for_test` seams): arm the one-shot publish pause. The
1840 /// NEXT [`StoreCore::publish`] (e.g. via [`Self::apply_bundle`])
1841 /// will, after swapping the live view and while still holding
1842 /// `live.write()`, signal the returned receiver and then block
1843 /// until the returned sender is used. Lets a witness sit in the
1844 /// "new view, old generation" window to prove the send-path /
1845 /// admission barriered reads never observe it. Not for production.
1846 #[doc(hidden)]
1847 #[cfg(any(test, feature = "fixtures"))]
1848 pub fn arm_publish_pause_for_test(
1849 &self,
1850 ) -> (std::sync::mpsc::Receiver<()>, std::sync::mpsc::Sender<()>) {
1851 let (swapped_tx, swapped_rx) = std::sync::mpsc::channel();
1852 let (resume_tx, resume_rx) = std::sync::mpsc::channel();
1853 *self.core.publish_pause.lock() = Some(PublishPauseHook {
1854 swapped: swapped_tx,
1855 resume: resume_rx,
1856 });
1857 (swapped_rx, resume_tx)
1858 }
1859
1860 /// Pin this store's publish transaction (review-9 addendum):
1861 /// while the returned guard lives, no reload can publish a new
1862 /// live view through this store's core, from any same-path
1863 /// handle. See [`PublishGuard`].
1864 pub(crate) fn publish_guard(&self) -> PublishGuard<'_> {
1865 PublishGuard {
1866 _guards: vec![self.core.reload.lock()],
1867 }
1868 }
1869
1870 /// Number of raise subscribers currently registered on the
1871 /// shared core (test/metric surface). Used by the review-11 P2
1872 /// leak witness to prove a dropped node unsubscribed its
1873 /// callback.
1874 #[doc(hidden)]
1875 pub fn subscriber_count(&self) -> usize {
1876 self.core.subscribers.read().len()
1877 }
1878
1879 /// Test-only (AV-10): snapshot the core's subscriber callbacks
1880 /// EXACTLY as [`StoreCore::notify`] does — clone the callback
1881 /// `Arc`s outside the registry lock. Lets a witness capture a
1882 /// callback BEFORE a node teardown and invoke it afterward to prove
1883 /// the owner-liveness token makes such a late callback inert.
1884 #[doc(hidden)]
1885 pub fn snapshot_subscribers_for_test(&self) -> Vec<FloorsRaisedCallback> {
1886 self.core
1887 .subscribers
1888 .read()
1889 .iter()
1890 .map(|(_, callback)| callback.clone())
1891 .collect()
1892 }
1893
1894 /// Register `callback` as ONE subscriber in the core's raise
1895 /// registry and return an externally-owned [`RaiseSubscription`]
1896 /// RAII guard (review-9 addendum: subscription is a registry, not a
1897 /// single replaceable slot — a second observer must never silently
1898 /// steal the first one's notifications). Subscribers fire after a
1899 /// reload publishes floors above the previously enforced view —
1900 /// including floors learned from OTHER writers via the under-lock
1901 /// reread, and raises published by same-path sibling handles.
1902 ///
1903 /// The callback is wrapped in an exclusion lease (R2-3): its body
1904 /// runs only while registered as in-flight, and a teardown draining
1905 /// the lease blocks until it leaves. Dropping the returned guard
1906 /// retires the subscription — draining any in-flight callback and
1907 /// removing it from the core through a `Weak<StoreCore>` (R2-2), so
1908 /// cleanup never depends on this facade's own drop.
1909 ///
1910 /// The callback may also be invoked with an EMPTY slice, meaning "this
1911 /// path's revocation AUTHORITY moved without raising any floor" — the poison
1912 /// recovery case. A subscriber that only iterates `raised` sees a harmless
1913 /// no-op; one that tracks authority (the routing registry) must treat it as
1914 /// a change (Kyra OLB-2B-E3c closure).
1915 #[must_use = "dropping the returned guard immediately unsubscribes the callback"]
1916 pub fn subscribe_floors_raised(
1917 &self,
1918 callback: impl Fn(&[RaisedFloor]) + Send + Sync + 'static,
1919 ) -> RaiseSubscription {
1920 let lease = SubscriptionLease::new();
1921 let lease_cb = Arc::clone(&lease);
1922 let wrapped: FloorsRaisedCallback = Arc::new(move |raised: &[RaisedFloor]| {
1923 // R2-3: admit under the lease, run the user callback OUTSIDE
1924 // the lease lock (re-entrant `apply_bundle` and long
1925 // retractions must not self-deadlock), then leave — even on
1926 // panic, via the drop guard.
1927 if !lease_cb.enter() {
1928 return;
1929 }
1930 struct LeaveOnDrop<'a>(&'a Arc<SubscriptionLease>);
1931 impl Drop for LeaveOnDrop<'_> {
1932 fn drop(&mut self) {
1933 self.0.leave();
1934 }
1935 }
1936 let _leave = LeaveOnDrop(&lease_cb);
1937 callback(raised);
1938 });
1939 let token = self.core.next_subscriber.fetch_add(1, Ordering::Relaxed);
1940 self.core.subscribers.write().push((token, wrapped));
1941 RaiseSubscription {
1942 core: Arc::downgrade(&self.core),
1943 token,
1944 lease,
1945 }
1946 }
1947
1948 /// Apply an operator bundle under the locked reload order
1949 /// (interprocess-safe, review-8 §5):
1950 ///
1951 /// ```text
1952 /// verify bundle signature
1953 /// → acquire exclusive lock on the stable `.lock` sidecar
1954 /// → REREAD the persisted maxima under the lock (load-bearing:
1955 /// an in-memory snapshot must never be the merge base)
1956 /// → monotone merge
1957 /// → atomically persist iff the disk state changed
1958 /// → publish the merged live view
1959 /// → release the lock, notify raise observers
1960 /// ```
1961 ///
1962 /// Returns the floors raised relative to this store's
1963 /// PREVIOUSLY published view — the supplied bundle's raises
1964 /// plus any floors another writer advanced on disk since the
1965 /// last reload. `Ok(empty)` means nothing rose (a lower bundle
1966 /// never rolls back).
1967 ///
1968 /// On pre-rename errors the persisted last-good state and the
1969 /// live view are both untouched. A POST-rename parent-fsync
1970 /// failure publishes the merged (never-weaker) view through
1971 /// the shared core (every same-path sibling advances with it),
1972 /// poisons the PATH, and returns
1973 /// [`OrgRevocationError::DurabilityUncertain`]; further
1974 /// same-path applies are refused until recovery — a locked
1975 /// reread republished through the core plus a successful
1976 /// parent-directory fsync — clears the uncertainty.
1977 pub fn apply_bundle(
1978 &self,
1979 bundle: &OrgRevocationBundle,
1980 ) -> Result<Vec<RaisedFloor>, OrgRevocationError> {
1981 let path = &self.core.path;
1982
1983 // The locked phase returns its outcome so raise observers
1984 // run AFTER both the file lock and the core's reload guard
1985 // have dropped — a callback that re-enters `apply_bundle`
1986 // on the same store must not deadlock (review-9).
1987 enum LockedOutcome {
1988 /// Raised floors, plus whether this apply RECOVERED the path from
1989 /// poison — which owes an authority wake even when nothing rose.
1990 Applied(Vec<RaisedFloor>, bool),
1991 /// Raised floors, the reason, and whether the mark was the
1992 /// false→true TRANSITION — which owes the same wake the recovery
1993 /// does when nothing rose (E3c blockers §1).
1994 DurabilityUncertain(Vec<RaisedFloor>, String, bool),
1995 }
1996
1997 // 1. Verify the incoming bundle's signature + canonical
1998 // structure BEFORE taking any lock — a corrupt bundle
1999 // keeps last-good, loudly, and touches nothing.
2000 if let Err(e) = bundle.verify() {
2001 let err = OrgRevocationError::InvalidBundle(e);
2002 tracing::error!(
2003 org = %bundle.org_id,
2004 "rejecting revocation bundle, keeping last-good persisted floors: {err}"
2005 );
2006 return Err(err);
2007 }
2008
2009 let outcome = {
2010 // Canonical lock order (review-11 P1): interprocess file
2011 // lock OUTER, core `reload` INNER — the SAME order every
2012 // opener (`join_or_create_core`) uses. Publishing under
2013 // `reload` is what makes [`PublishGuard`] a real barrier:
2014 // no publish can land between a replacement's dominance
2015 // comparison and its swap. A replacement holds only
2016 // `reload` (never the file lock), so no lock cycle forms.
2017 let lock = lock_state_file(path)?;
2018 let _guard = self.core.reload.lock();
2019
2020 // R3-3: the `.lock` sidecar just opened MUST be the SAME
2021 // identity this live core was created on. If the sidecar was
2022 // deleted and recreated (fresh inode) beneath a still-live
2023 // handle — which the `nlink != 1` refusal does NOT catch, since
2024 // the replacement has one link — this transaction would lock
2025 // and publish through a DIFFERENT backing identity than its
2026 // core's, operating outside its original lock / publication
2027 // domain (a new opener is refused by `BackingIdentityConflict`,
2028 // but the existing handle would sail on). Refuse loudly BEFORE
2029 // any reread / merge / write, so disk and the live view are
2030 // both untouched.
2031 let opened_id = BackingId::of(&lock, path)?;
2032 if opened_id != self.core.backing_id {
2033 drop(lock);
2034 return Err(OrgRevocationError::BackingIdentityConflict {
2035 path: path.display().to_string(),
2036 });
2037 }
2038
2039 // 2. Interprocess critical section. A poisoned path
2040 // must first prove its directory entry durable; the
2041 // reread + publish below then republish the ground
2042 // truth through the shared core BEFORE the poison
2043 // bit clears (review-9 addendum: recovery reloads
2044 // live views, it never merely fsyncs).
2045 let was_poisoned = is_poisoned(&self.core.backing_id, path);
2046 if was_poisoned {
2047 prove_entry_durable(path)?;
2048 }
2049
2050 // 3. REREAD the persisted maxima under the lock — the
2051 // reread is load-bearing: merging from this
2052 // instance's live snapshot would let a stale writer
2053 // overwrite floors another writer already persisted.
2054 let disk_bytes = read_regular_nofollow(path).map_err(|e| OrgRevocationError::Io {
2055 path: path.display().to_string(),
2056 reason: e.to_string(),
2057 })?;
2058 let disk = OrgRevocationState::from_file_bytes(&disk_bytes, path)?;
2059
2060 // 4. Monotone merge against the reread disk state.
2061 let mut merged = disk.clone();
2062 let raised_on_disk = merged.merge_bundle(bundle);
2063
2064 // 5. Persist iff the disk state changed; the write must
2065 // complete before anything is published.
2066 let mut durability_uncertain: Option<(String, bool)> = None;
2067 if raised_on_disk > 0 {
2068 // Test seam: report a post-rename durability failure while
2069 // leaving the file at its PRIOR bytes — the exact uncertainty
2070 // PostRename names (the entry may resolve to the old state).
2071 // On Windows the phase is otherwise unreachable in production
2072 // (write-through rename, §13), so the mark path has no other
2073 // witness route there.
2074 #[cfg(test)]
2075 let write = if self.core.force_post_rename.swap(false, Ordering::AcqRel) {
2076 Err(WritePhase::PostRename(
2077 "forced post-rename failure (test seam)".to_string(),
2078 ))
2079 } else {
2080 write_atomic_phased(path, &merged.to_file_bytes()?)
2081 };
2082 #[cfg(not(test))]
2083 let write = write_atomic_phased(path, &merged.to_file_bytes()?);
2084 match write {
2085 Ok(()) => {}
2086 Err(WritePhase::PreRename(reason)) => {
2087 // Old file (rename never happened) and old
2088 // live view both intact — a floor the disk
2089 // could forget is never enforced.
2090 drop(lock);
2091 return Err(OrgRevocationError::Io {
2092 path: path.display().to_string(),
2093 reason,
2094 });
2095 }
2096 Err(WritePhase::PostRename(reason)) => {
2097 // The rename LANDED; only the directory-entry
2098 // durability is uncertain. Still publish the
2099 // merged (never-weaker) view below so
2100 // enforcement doesn't regress under what the
2101 // disk may now hold, but poison the PATH: no
2102 // instance may pretend disk and memory are
2103 // synchronized until recovery proves the
2104 // entry durable.
2105 // Ordered BEFORE the publish below, matching the frozen
2106 // `poison_gate` → `live` order.
2107 let newly_poisoned = self.core.mark_poisoned();
2108 durability_uncertain = Some((reason, newly_poisoned));
2109 }
2110 }
2111 }
2112
2113 // 6. Publish the merged view through the SHARED core —
2114 // every same-path handle's view advances the instant
2115 // this lands (review-9 addendum) — then clear any
2116 // recovered poison and release the lock;
2117 // notification happens outside.
2118 let raised = self.core.publish(merged);
2119 let recovered = was_poisoned && durability_uncertain.is_none();
2120 if recovered {
2121 // Through the CORE: the clear is exactly as load-bearing as the
2122 // mark, and the file lock is still held here — the wake for it
2123 // happens below, outside every guard.
2124 self.core.clear_poison();
2125 }
2126 drop(lock);
2127 match durability_uncertain {
2128 None => LockedOutcome::Applied(raised, recovered),
2129 Some((reason, newly)) => LockedOutcome::DurabilityUncertain(raised, reason, newly),
2130 }
2131 };
2132
2133 match outcome {
2134 LockedOutcome::Applied(raised, recovered) => {
2135 self.core.notify(&raised);
2136 if recovered {
2137 // A recovery that raised nothing still moved authority.
2138 self.core.notify_authority_changed();
2139 }
2140 Ok(raised)
2141 }
2142 LockedOutcome::DurabilityUncertain(raised, reason, newly_poisoned) => {
2143 let err = OrgRevocationError::DurabilityUncertain {
2144 path: path.display().to_string(),
2145 reason,
2146 };
2147 tracing::error!("{err}");
2148 self.core.notify(&raised);
2149 if newly_poisoned && raised.is_empty() {
2150 // The MARK owes the same wake the CLEAR does (E3c blockers
2151 // §1): what this node may serve just went from the real
2152 // material to NOTHING, yet `notify` is silent on an empty
2153 // raise set — which is exactly what a mark produces
2154 // whenever the live view was already ahead of what disk
2155 // can prove (the rollback case PostRename models). Without
2156 // this, the registry stays reconciled to pre-poison facts
2157 // until a reader happens to trip the lazy epoch check.
2158 //
2159 // Guarded on BOTH conditions: a non-empty `raised` already
2160 // woke every subscriber above — authority-tracking
2161 // subscribers treat any invocation as movement, so waking
2162 // again would double-bump the epoch for one transition —
2163 // and a re-mark of an already-poisoned path has no
2164 // transition to report; routing facts are already stamped
2165 // `poisoned == true`.
2166 self.core.notify_authority_changed();
2167 }
2168 Err(err)
2169 }
2170 }
2171 }
2172}
2173
2174impl std::fmt::Debug for OrgRevocationStore {
2175 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2176 f.debug_struct("OrgRevocationStore")
2177 .field("path", &self.core.path)
2178 .field("floors", &self.snapshot().len())
2179 .finish()
2180 }
2181}
2182
2183/// Which phase of the durable write failed. PRE-rename failures
2184/// are recoverable (the target file was never touched; the temp is
2185/// cleaned up). POST-rename failures mean the directory entry may
2186/// already point at the new bytes while its durability is unproven
2187/// — the caller must fail closed (review-8 §13).
2188pub(crate) enum WritePhase {
2189 /// The target file is untouched; nothing published.
2190 PreRename(String),
2191 /// The rename landed; only the parent-directory fsync failed.
2192 PostRename(String),
2193}
2194
2195/// Process-wide durability-uncertainty registry, keyed by the
2196/// NORMALIZED backing path (review-9): the filesystem's uncertainty
2197/// after a landed-rename/failed-dir-fsync belongs to the directory
2198/// entry, not to one `OrgRevocationStore` instance. Every store
2199/// opened on the same pathname shares the poison bit; recovery
2200/// (a locked reread republished through the shared core plus a
2201/// SUCCESSFUL parent-directory fsync) clears it. Separate from the
2202/// core registry because poison must outlive every handle.
2203static PATH_POISON: std::sync::OnceLock<Mutex<PoisonRegistry>> = std::sync::OnceLock::new();
2204
2205/// # §16/§17 — the two limits of this tombstone, stated plainly
2206///
2207/// **It is PROCESS-LOCAL.** `PATH_POISON` is a `OnceLock<Mutex<..>>` in
2208/// process memory, so a RESTART discards every poison record. That matters
2209/// because a restart is the natural operator response to a
2210/// `DurabilityUncertain` error: the new process performs no recovery, reads
2211/// whatever the directory entry now resolves to — possibly the pre-rename
2212/// state — and publishes it as ground truth. A restart is therefore not a
2213/// route to recovery; it is the one action that guarantees the uncertainty is
2214/// discarded unexamined. Documented rather than fixed because a durable
2215/// marker cannot be fsynced into the very directory whose fsync just failed;
2216/// closing it properly needs a marker in a DIFFERENT directory, or an
2217/// operator acknowledgement gate on `open_existing`.
2218///
2219/// **The path index can still be laundered by deleting BOTH files.**
2220/// `poison_path_key` falls back to the non-canonical normalized path when
2221/// `canonicalize` fails (correctly — it must not memoize a guess), but
2222/// `mark_poisoned` recorded the CANONICAL key while the state file still
2223/// existed. Remove the state file and its `.lock`, and a subsequent `init`
2224/// computes the fallback key, misses `by_path`, gets a fresh inode so misses
2225/// `by_id`, and proceeds as unpoisoned. §1's `ProvisioningExpectation` is what
2226/// covers that case now — the caller knows the node was provisioned before
2227/// even when this registry has forgotten.
2228///
2229/// Two poison indexes that must BOTH be consulted (R3-2):
2230///
2231/// - `by_id` — the live `.lock` sidecar identity ([`BackingId`]), which
2232/// is what same-path handles join their core on; and
2233/// - `by_path` — the CANONICAL state-file path mapped to the SET of every
2234/// sidecar identity ever poisoned under it. The path key survives `.lock`
2235/// sidecar replacement: keying poison only on the sidecar identity let a
2236/// durability-uncertain path be laundered by dropping every handle and
2237/// recreating the `.lock` (new inode ⇒ new `BackingId` ⇒ `by_id` miss ⇒
2238/// recovery skipped). The path tombstone closes that — once poisoned, the
2239/// path stays poisoned across sidecar recreation until explicit recovery,
2240/// and case-aliases collapse through the actual filesystem
2241/// (`canonicalize`), not blind case-folding.
2242///
2243/// Tracking the id SET per path (not just the path itself) lets recovery
2244/// retire EVERY stale old `BackingId` for that path in one step (P2
2245/// hygiene): a sidecar that was unlinked and recreated stranded its old id
2246/// in `by_id`, and a later store re-using that recycled inode would
2247/// otherwise trip redundant recovery on the dead id's residue.
2248#[derive(Default)]
2249struct PoisonRegistry {
2250 by_id: std::collections::HashSet<BackingId>,
2251 by_path: std::collections::HashMap<PathBuf, std::collections::HashSet<BackingId>>,
2252}
2253
2254fn poison_registry() -> &'static Mutex<PoisonRegistry> {
2255 PATH_POISON.get_or_init(|| Mutex::new(PoisonRegistry::default()))
2256}
2257
2258/// Memo for [`poison_path_key`]: `normalized_path -> canonical key`.
2259///
2260/// Bounded by the number of distinct authority paths this process has
2261/// touched — the same bound `PATH_POISON` already carries — so it needs no
2262/// eviction.
2263static POISON_KEY_MEMO: std::sync::OnceLock<Mutex<std::collections::HashMap<PathBuf, PathBuf>>> =
2264 std::sync::OnceLock::new();
2265
2266fn poison_key_memo() -> &'static Mutex<std::collections::HashMap<PathBuf, PathBuf>> {
2267 POISON_KEY_MEMO.get_or_init(|| Mutex::new(std::collections::HashMap::new()))
2268}
2269
2270/// The case-normalized poison-tombstone key for `normalized_path` (R3-2).
2271/// `canonicalize` collapses case-aliases through the ACTUAL filesystem
2272/// identity — not blind ASCII case-folding, which would over-poison two
2273/// genuinely distinct files on a case-sensitive filesystem. Falls back to
2274/// the normalized path when the state file does not yet exist (a fresh
2275/// `init` before creation) or `canonicalize` otherwise fails.
2276///
2277/// MEMOIZED (§13). The state being guarded is a process-local `HashSet` /
2278/// `HashMap`, but reaching it used to cost a full path resolution on EVERY
2279/// call: on Linux an `lstat`/`readlink` per component, on Windows a
2280/// `CreateFileW` + `GetFinalPathNameByHandleW` — a real file open. That was
2281/// paid three times per protected unary RPC (`org_admission_gate`) and twice
2282/// per inbound scoped-announcement envelope that clears the relay gate.
2283///
2284/// The sharp case was not throughput: `apply_bundle` calls `is_poisoned`
2285/// while holding BOTH the interprocess `.lock` sidecar and `core.reload`. On
2286/// an NFS/SMB or stalled-disk authority directory that `canonicalize` blocks
2287/// for the filesystem timeout with the cross-process revocation lock held,
2288/// stalling every other process's `apply_bundle` and every `StoreCore::publish`
2289/// on that core — hence every `barriered_generation()` / `snapshot_with_generation()`
2290/// reader in `verify_provider_authority`. The same pattern under `_pin` in
2291/// `install_org_revocation_store_locked` freezes publishes on two cores at once.
2292///
2293/// Only SUCCESSFUL canonicalizations are memoized. Caching the fallback would
2294/// pin the non-canonical path as the key forever — including after the state
2295/// file is created — and a later case-alias would then miss its tombstone.
2296///
2297/// A memoized key can only go stale if the path is later repointed (e.g. a
2298/// symlink swung elsewhere). That direction is fail-CLOSED: the tombstone
2299/// keeps applying to the original identity rather than silently following the
2300/// path to a new one.
2301fn poison_path_key(normalized_path: &Path) -> PathBuf {
2302 if let Some(hit) = poison_key_memo().lock().get(normalized_path) {
2303 return hit.clone();
2304 }
2305 // Resolve OUTSIDE the memo lock: this is the call that can block for a
2306 // filesystem timeout, and holding the memo lock across it would just move
2307 // the stall rather than remove it.
2308 match std::fs::canonicalize(normalized_path) {
2309 Ok(canonical) => {
2310 poison_key_memo()
2311 .lock()
2312 .insert(normalized_path.to_path_buf(), canonical.clone());
2313 canonical
2314 }
2315 // Not yet created (fresh `init`) or otherwise unresolvable — fall back
2316 // WITHOUT memoizing, so the real canonical key is picked up once the
2317 // file exists.
2318 Err(_) => normalized_path.to_path_buf(),
2319 }
2320}
2321
2322/// Poison `normalized_path` under BOTH indexes (R3-2), recording `id` in the
2323/// path's id set so recovery can retire every id ever poisoned here.
2324/// Returns whether the path was NEWLY poisoned — neither index held it before.
2325/// The transition signal `apply_bundle` uses to decide whether a mark that
2326/// raised no floor still owes an authority wake (E3c blockers §1). Computed
2327/// under the registry lock, so it cannot race a concurrent mark or clear.
2328fn mark_poisoned(id: &BackingId, normalized_path: &Path) -> bool {
2329 let key = poison_path_key(normalized_path);
2330 let mut reg = poison_registry().lock();
2331 let newly = !reg.by_id.contains(id) && !reg.by_path.contains_key(&key);
2332 reg.by_id.insert(id.clone());
2333 reg.by_path.entry(key).or_default().insert(id.clone());
2334 newly
2335}
2336
2337/// Poisoned iff EITHER the live sidecar identity OR the canonical state
2338/// path is tombstoned — so a recreated sidecar (new `BackingId`, same
2339/// path) is still caught (R3-2).
2340fn is_poisoned(id: &BackingId, normalized_path: &Path) -> bool {
2341 let key = poison_path_key(normalized_path);
2342 let reg = poison_registry().lock();
2343 reg.by_id.contains(id) || reg.by_path.contains_key(&key)
2344}
2345
2346/// Normalize a backing pathname ONCE at store construction
2347/// (review-9 addendum): the CANONICAL parent joined with the
2348/// literal final component, with NO verbatim fallback. Aliases of
2349/// one file (bare vs `./`, relative vs absolute, `..` hops,
2350/// symlinked parents) land on ONE core and ONE poison entry, so a
2351/// single backing file never gets independent security views. A
2352/// path with no final component, or whose parent cannot resolve,
2353/// is refused.
2354///
2355/// The final component is validated for symlink/non-regular
2356/// ATOMICALLY (review-11 P2): the previous form did
2357/// `symlink_metadata` then `canonicalize` as two syscalls, and the
2358/// final component could be swapped to a symlink in between —
2359/// `canonicalize` would then follow it and key the store to the
2360/// link's target, which the later no-follow opens could not detect.
2361/// A no-follow open of the joined path IS the check: it refuses a
2362/// symlink (`ELOOP`) or non-regular final in one syscall, or
2363/// reports the file simply does not exist yet (a fresh `init`).
2364///
2365/// The parent is canonicalized (resolving parent symlinks and
2366/// case), so parent-side aliases still collapse; the FINAL
2367/// component is taken literally rather than canonicalized. In
2368/// practice the final component is a fixed constant
2369/// (`revocation-state.json`, `owner-audience.key`), so
2370/// final-component case aliasing on case-insensitive filesystems
2371/// is not a real call shape — trading it away removes the TOCTOU.
2372pub(crate) fn normalize_backing_path(path: &Path) -> Result<PathBuf, OrgRevocationError> {
2373 let io = |reason: String| OrgRevocationError::Io {
2374 path: path.display().to_string(),
2375 reason,
2376 };
2377 let Some(file_name) = path.file_name() else {
2378 return Err(io("backing path has no final component".to_string()));
2379 };
2380 let parent = match path.parent() {
2381 Some(p) if !p.as_os_str().is_empty() => p,
2382 _ => Path::new("."),
2383 };
2384 let canon_parent = parent
2385 .canonicalize()
2386 .map_err(|e| io(format!("cannot canonicalize parent directory: {e}")))?;
2387 let joined = canon_parent.join(file_name);
2388 // Atomic final-component validation: the no-follow open refuses
2389 // a symlink/FIFO/non-regular final in one syscall (no
2390 // stat→canonicalize gap). NotFound is fine — a fresh store
2391 // creates the file under exactly this name.
2392 match open_regular_nofollow(&joined) {
2393 Ok(_) => Ok(joined),
2394 Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(joined),
2395 Err(e) => Err(io(format!(
2396 "refusing non-regular backing path (symlink/FIFO/other): {e}"
2397 ))),
2398 }
2399}
2400
2401/// First half of durability recovery, called with the interprocess
2402/// lock HELD: prove the directory entry durable with a
2403/// parent-directory fsync. Failure refuses with
2404/// [`OrgRevocationError::Poisoned`] — no same-path operation may
2405/// proceed while uncertainty remains. On success the caller MUST
2406/// reread the state file and republish it through the shared core
2407/// (so every live sibling advances to ground truth) BEFORE calling
2408/// [`clear_poison`] — recovery reloads live views; it never merely
2409/// fsyncs (review-9 addendum).
2410fn prove_entry_durable(path: &Path) -> Result<(), OrgRevocationError> {
2411 fsync_parent_dir(path).map_err(|e| {
2412 tracing::error!(
2413 path = %path.display(),
2414 error = %e,
2415 "revocation-state durability recovery failed; path remains poisoned"
2416 );
2417 OrgRevocationError::Poisoned {
2418 path: path.display().to_string(),
2419 }
2420 })
2421}
2422
2423/// Second half of durability recovery: clear the path-wide bit
2424/// after the entry was proven durable AND the reread state was
2425/// republished through the shared core.
2426fn clear_poison(id: &BackingId, path: &Path) {
2427 let key = poison_path_key(path);
2428 {
2429 let mut reg = poison_registry().lock();
2430 // Retire EVERY sidecar identity ever poisoned under this canonical
2431 // path, not just the recovering one (P2 hygiene): a prior sidecar
2432 // that was unlinked and recreated left its old `BackingId` stranded
2433 // in `by_id`, and a later store re-using that recycled inode would
2434 // otherwise trip redundant recovery on the dead residue.
2435 if let Some(ids) = reg.by_path.remove(&key) {
2436 for stale in ids {
2437 reg.by_id.remove(&stale);
2438 }
2439 }
2440 reg.by_id.remove(id);
2441 }
2442 tracing::warn!(
2443 path = %path.display(),
2444 "revocation-state durability uncertainty recovered \
2445 (locked reread republished; parent directory fsynced)"
2446 );
2447}
2448
2449/// Open `path` as a REGULAR file without following symlinks
2450/// (review-9): authority/state data and the stable lock inode must
2451/// never be attacker-steerable through a planted link, and the
2452/// permission/type checks must run on the OPENED handle so there is
2453/// no check-to-use window.
2454///
2455/// Unix uses `O_NOFOLLOW` (a symlink final component fails to
2456/// open); other platforms fall back to a `symlink_metadata`
2457/// pre-check plus a handle-metadata type check.
2458pub(crate) fn open_regular_nofollow(path: &Path) -> std::io::Result<std::fs::File> {
2459 let mut opts = std::fs::OpenOptions::new();
2460 opts.read(true);
2461 #[cfg(unix)]
2462 {
2463 use std::os::unix::fs::OpenOptionsExt;
2464 opts.custom_flags(libc::O_NOFOLLOW);
2465 }
2466 #[cfg(not(unix))]
2467 {
2468 let meta = std::fs::symlink_metadata(path)?;
2469 if meta.file_type().is_symlink() {
2470 return Err(std::io::Error::new(
2471 std::io::ErrorKind::InvalidInput,
2472 "refusing symlink: authority files must be regular files",
2473 ));
2474 }
2475 }
2476 let opened = opts.open(path);
2477 // Map the Unix `O_NOFOLLOW` symlink rejection (`ELOOP`) to a clear typed
2478 // error. Unix-only: `#[cfg(not(unix))]` has no `O_NOFOLLOW`, and mapping
2479 // there would be an identity map (`|e| e`).
2480 #[cfg(unix)]
2481 let opened = opened.map_err(|e| {
2482 if e.raw_os_error() == Some(libc::ELOOP) {
2483 std::io::Error::new(
2484 std::io::ErrorKind::InvalidInput,
2485 "refusing symlink: authority files must be regular files",
2486 )
2487 } else {
2488 e
2489 }
2490 });
2491 let file = opened?;
2492 // Type check on the opened descriptor — immune to a swap
2493 // between check and use.
2494 let meta = file.metadata()?;
2495 if !meta.is_file() {
2496 return Err(std::io::Error::new(
2497 std::io::ErrorKind::InvalidInput,
2498 "refusing non-regular file: authority files must be regular files",
2499 ));
2500 }
2501 Ok(file)
2502}
2503
2504/// Read a whole regular file through a no-follow handle.
2505pub(crate) fn read_regular_nofollow(path: &Path) -> std::io::Result<Vec<u8>> {
2506 use std::io::Read;
2507 let mut file = open_regular_nofollow(path)?;
2508 let mut bytes = Vec::new();
2509 file.read_to_end(&mut bytes)?;
2510 Ok(bytes)
2511}
2512
2513/// Acquire the exclusive interprocess lock guarding `path` via its
2514/// stable `.lock` sidecar (the state file itself is replaced by
2515/// rename, so it cannot carry the lock). Blocking; released when
2516/// the returned handle drops. std advisory file locking — same
2517/// semantics as the sdk revocation store's fs2 sidecar.
2518///
2519/// `pub(crate)`: the adoption ceremony's final phase holds this
2520/// lock across its floor re-verification and membership write.
2521pub(crate) fn lock_state_file(path: &Path) -> Result<std::fs::File, OrgRevocationError> {
2522 let io = |e: std::io::Error| OrgRevocationError::Io {
2523 path: path.display().to_string(),
2524 reason: format!("state lock: {e}"),
2525 };
2526 let mut lock_path = path.as_os_str().to_os_string();
2527 lock_path.push(".lock");
2528 let lock = open_lock_file(&PathBuf::from(lock_path)).map_err(io)?;
2529 // R2-4: a legitimately-created sidecar has exactly ONE hard link. A
2530 // link count above one means someone hard-linked this sidecar's inode
2531 // to a SECOND name — the attack that would otherwise collapse two
2532 // distinct state paths onto one [`BackingId`] (and thus one core /
2533 // poison entry). Refuse fail-closed on every platform. `std` exposes the
2534 // link count on Unix (`nlink`) and on Windows only via the stable Win32
2535 // `GetFileInformationByHandle` (`nNumberOfLinks`), read here directly.
2536 #[cfg(unix)]
2537 let nlink = {
2538 use std::os::unix::fs::MetadataExt;
2539 // `MetadataExt::nlink()` is already `u64` on every Unix — no conversion.
2540 lock.metadata().map_err(io)?.nlink()
2541 };
2542 #[cfg(windows)]
2543 let nlink = {
2544 let (_volume, _index, links) = windows_file_identity(&lock).map_err(io)?;
2545 u64::from(links)
2546 };
2547 #[cfg(any(unix, windows))]
2548 if nlink != 1 {
2549 return Err(OrgRevocationError::Io {
2550 path: path.display().to_string(),
2551 reason: format!(
2552 "state lock: refusing .lock sidecar with {nlink} hard links \
2553 (expected 1) — a hard-linked sidecar would alias two backing paths"
2554 ),
2555 });
2556 }
2557 Ok(lock)
2558}
2559
2560/// Open-and-lock a lock inode (`.lock` sidecar, ceremony lock)
2561/// under the full regular-file policy (review-9): no-follow (a
2562/// planted symlink cannot redirect the lock inode), `O_NONBLOCK`
2563/// (a planted FIFO fails or returns instead of blocking the open
2564/// forever), and a type check on the OPENED descriptor — advisory
2565/// locking a non-regular inode is not a lock on anything this
2566/// module owns. `O_NONBLOCK` is inert for regular files and does
2567/// not affect the (deliberately blocking) advisory lock call.
2568///
2569/// `pub(crate)`: the adoption ceremony lock
2570/// (`org_authority::lock_ceremony`) applies the same policy.
2571pub(crate) fn open_lock_file(lock_path: &Path) -> std::io::Result<std::fs::File> {
2572 let mut opts = std::fs::OpenOptions::new();
2573 opts.create(true).write(true).truncate(false);
2574 #[cfg(unix)]
2575 {
2576 use std::os::unix::fs::OpenOptionsExt;
2577 opts.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
2578 opts.mode(0o600);
2579 }
2580 #[cfg(not(unix))]
2581 {
2582 // Non-Unix has no O_NOFOLLOW: same symlink precheck as
2583 // `open_regular_nofollow` (plus the opened-handle type
2584 // check below).
2585 if let Ok(meta) = std::fs::symlink_metadata(lock_path) {
2586 if meta.file_type().is_symlink() {
2587 return Err(std::io::Error::new(
2588 std::io::ErrorKind::InvalidInput,
2589 "refusing symlink: lock files must be regular files",
2590 ));
2591 }
2592 }
2593 }
2594 let f = opts.open(lock_path)?;
2595 // Type check on the opened descriptor — immune to a swap
2596 // between check and use.
2597 if !f.metadata()?.is_file() {
2598 return Err(std::io::Error::new(
2599 std::io::ErrorKind::InvalidInput,
2600 "refusing non-regular file: lock files must be regular files",
2601 ));
2602 }
2603 f.lock()?;
2604 Ok(f)
2605}
2606
2607/// Flush the parent directory of `path`, so the RENAME that published the file
2608/// is durable and not just the file's contents.
2609///
2610/// Split out so the durability-recovery path (review-9) can prove the
2611/// directory entry durable without rewriting the file.
2612///
2613/// # §13 — why this is Unix-only, correctly
2614///
2615/// The original comment justified the non-Unix no-op with "the rename
2616/// primitive carries the metadata guarantee", which is too vague to check and
2617/// reads like a hand-wave. It is, however, the right ANSWER for the wrong
2618/// reason, and the fix is not where it looks.
2619///
2620/// Windows has no directory fsync. `FlushFileBuffers` on a directory handle
2621/// returns `ERROR_ACCESS_DENIED` — it is not a supported operation, whatever
2622/// the symmetry with POSIX suggests. (Verified here: an implementation using
2623/// `CreateFileW` + `FILE_FLAG_BACKUP_SEMANTICS` + `FlushFileBuffers` failed
2624/// every store test with os error 5.)
2625///
2626/// The documented Win32 mechanism is on the RENAME instead:
2627/// `MoveFileExW(..., MOVEFILE_WRITE_THROUGH)` "guarantees that the move is
2628/// flushed to disk before the function returns". [`write_atomic_phased`] now
2629/// uses it, so on Windows durability is achieved INLINE with the publish
2630/// rather than in a second phase.
2631///
2632/// One consequence is worth stating rather than leaving to be rediscovered:
2633/// `WritePhase::PostRename` remains unreachable on Windows, so the poison
2634/// machinery still does not arm there. That is now correct rather than a gap —
2635/// with write-through there is no "renamed, but perhaps not durable" window to
2636/// be uncertain ABOUT. The rename either committed durably or returned an
2637/// error, which the pre-rename phase already handles.
2638fn fsync_parent_dir(path: &Path) -> std::io::Result<()> {
2639 #[cfg(unix)]
2640 {
2641 let dir = match path.parent() {
2642 Some(p) if !p.as_os_str().is_empty() => p,
2643 _ => Path::new("."),
2644 };
2645 std::fs::File::open(dir)?.sync_all()?;
2646 }
2647 #[cfg(not(unix))]
2648 {
2649 // Durability is carried by MOVEFILE_WRITE_THROUGH at rename time; see
2650 // the §13 note above. Deliberately not an error: there is nothing left
2651 // to prove at this point.
2652 let _ = path;
2653 }
2654 Ok(())
2655}
2656
2657/// Atomically replace `dest` with `src`, DURABLY (§13, Windows).
2658///
2659/// `std::fs::rename` is `MoveFileExW` with `MOVEFILE_REPLACE_EXISTING` and no
2660/// write-through, so the directory entry lands in the volume metadata cache
2661/// and a power loss seconds after `apply_bundle` returned `Ok` could lose it —
2662/// while the node had already published the raised floor and its subscribers
2663/// had retracted ownership. That is precisely the rollback this module exists
2664/// to prevent, and NTFS journalling does not close it: the journal guarantees
2665/// metadata CONSISTENCY after a crash, not that a completed rename was flushed
2666/// before the call returned.
2667///
2668/// `MOVEFILE_WRITE_THROUGH` is the documented fix and makes the publish
2669/// durable inline.
2670#[cfg(windows)]
2671#[allow(clippy::multiple_unsafe_ops_per_block)]
2672fn rename_write_through(src: &Path, dest: &Path) -> std::io::Result<()> {
2673 use std::os::windows::ffi::OsStrExt;
2674 extern "system" {
2675 fn MoveFileExW(existing: *const u16, new: *const u16, flags: u32) -> i32;
2676 }
2677 const MOVEFILE_REPLACE_EXISTING: u32 = 0x0000_0001;
2678 const MOVEFILE_WRITE_THROUGH: u32 = 0x0000_0008;
2679
2680 let mut from: Vec<u16> = src.as_os_str().encode_wide().collect();
2681 from.push(0);
2682 let mut to: Vec<u16> = dest.as_os_str().encode_wide().collect();
2683 to.push(0);
2684 // SAFETY: both buffers are NUL-terminated and outlive the call; the return
2685 // value is checked and no pointer escapes this scope.
2686 let ok = unsafe {
2687 MoveFileExW(
2688 from.as_ptr(),
2689 to.as_ptr(),
2690 MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
2691 )
2692 };
2693 if ok == 0 {
2694 return Err(std::io::Error::last_os_error());
2695 }
2696 Ok(())
2697}
2698/// Monotone counter qualifying temp names so two writers in one
2699/// process (or a reused PID) can never collide on a temp inode.
2700static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
2701
2702/// A fresh, unpredictable same-directory temp path:
2703/// `<file>.tmp.<pid>.<seq>.<rand16hex>`. Appended to the FULL file
2704/// name (the previous `with_extension` form replaced `.json`,
2705/// making the name predictable — review-8 §10: a pre-created
2706/// permissive temp would survive `create(true).truncate(true)`
2707/// with its original mode).
2708///
2709/// Entropy failure is an ERROR, not a silent all-zero suffix
2710/// (review-9): pid + a process-local sequence do not survive PID
2711/// reuse, so the random suffix is load-bearing for the
2712/// unpredictability claim. `create_new` keeps even that failure
2713/// mode fail-loud, but we don't rely on it.
2714fn fresh_temp_path(path: &Path) -> Result<PathBuf, WritePhase> {
2715 let mut rand = [0u8; 8];
2716 getrandom::fill(&mut rand)
2717 .map_err(|e| WritePhase::PreRename(format!("temp-name entropy unavailable: {e:?}")))?;
2718 let mut s = path.as_os_str().to_os_string();
2719 s.push(format!(
2720 ".tmp.{}.{}.{}",
2721 std::process::id(),
2722 TEMP_SEQ.fetch_add(1, Ordering::Relaxed),
2723 hex::encode(rand)
2724 ));
2725 Ok(PathBuf::from(s))
2726}
2727
2728/// Durable atomic write with phase-typed failures: fresh
2729/// `create_new` temp (owner-only mode applied at creation — never
2730/// a reused inode) → write → flush → fsync temp → atomic rename →
2731/// fsync parent directory. The temp file is removed on every
2732/// pre-rename failure. Unlike the sdk's `RevocationStore`, the
2733/// parent-dir fsync is a hard requirement here (plan §1.5 locked
2734/// order).
2735pub(crate) fn write_atomic_phased(path: &Path, bytes: &[u8]) -> Result<(), WritePhase> {
2736 let pre = |e: std::io::Error| WritePhase::PreRename(e.to_string());
2737
2738 if let Some(parent) = path.parent() {
2739 if !parent.as_os_str().is_empty() {
2740 std::fs::create_dir_all(parent).map_err(pre)?;
2741 }
2742 }
2743
2744 // `create_new` + creation-time 0600: an attacker cannot
2745 // pre-create the (unpredictable) name, and even a collision
2746 // with a crash-left temp fails loudly instead of truncating a
2747 // permissive inode. A handful of retries covers the
2748 // astronomically unlikely name collision.
2749 let mut tmp = fresh_temp_path(path)?;
2750 let mut file = None;
2751 for _ in 0..4 {
2752 let mut opts = std::fs::OpenOptions::new();
2753 opts.write(true).create_new(true);
2754 #[cfg(unix)]
2755 {
2756 use std::os::unix::fs::OpenOptionsExt;
2757 opts.mode(0o600);
2758 }
2759 match opts.open(&tmp) {
2760 Ok(f) => {
2761 file = Some(f);
2762 break;
2763 }
2764 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2765 tmp = fresh_temp_path(path)?;
2766 }
2767 Err(e) => return Err(pre(e)),
2768 }
2769 }
2770 let Some(mut f) = file else {
2771 return Err(WritePhase::PreRename(
2772 "could not create a fresh temp file after 4 attempts".to_string(),
2773 ));
2774 };
2775
2776 // Any failure before the rename removes the temp so no stale
2777 // inode accumulates for later reuse.
2778 let write_result = (|| -> std::io::Result<()> {
2779 use std::io::Write;
2780 f.write_all(bytes)?;
2781 f.flush()?;
2782 f.sync_all()?;
2783 Ok(())
2784 })();
2785 if let Err(e) = write_result {
2786 drop(f);
2787 let _ = std::fs::remove_file(&tmp);
2788 return Err(pre(e));
2789 }
2790 drop(f);
2791
2792 // Atomic replacement. On Unix, rename(2) atomically replaces
2793 // an existing destination. On Windows, std::fs::rename is
2794 // DOCUMENTED to replace an existing destination file (see the
2795 // std platform-specific behavior notes), so no separate
2796 // ReplaceFileW path is required for replacement semantics.
2797 // Crash DURABILITY is a distinct boundary, and `rename` alone does
2798 // not carry it on ANY platform: the directory entry lives in the
2799 // metadata cache until the parent directory is flushed. The
2800 // parent-dir flush below is implemented on Unix (`fsync`) and on
2801 // Windows (`FlushFileBuffers` on a backup-semantics directory
2802 // handle), so the fail-closed poison machinery arms on both.
2803 //
2804 // It was Unix-only, which made `WritePhase::PostRename`
2805 // unreachable on Windows and every poison/recovery path there
2806 // vacuous — see `fsync_parent_dir` (§13).
2807 // §13 — on Windows, publish through MOVEFILE_WRITE_THROUGH so the
2808 // directory entry is durable when this returns; `std::fs::rename` omits
2809 // the flag and leaves it in the volume metadata cache. On Unix the
2810 // parent fsync below carries it.
2811 #[cfg(windows)]
2812 let renamed = rename_write_through(&tmp, path);
2813 #[cfg(not(windows))]
2814 let renamed = std::fs::rename(&tmp, path);
2815 if let Err(e) = renamed {
2816 let _ = std::fs::remove_file(&tmp);
2817 return Err(pre(e));
2818 }
2819
2820 // rename() updates the directory entry in cache only; a crash
2821 // before the directory is flushed can revert to the old file
2822 // (BUG #93 lineage, mirrors redex/disk.rs). Required, not
2823 // best-effort — and a failure HERE is post-rename: the caller
2824 // must treat disk state as unproven (review-8 §13). True on
2825 // Windows as well as POSIX, which is what §13 corrected.
2826 if let Err(e) = fsync_parent_dir(path) {
2827 return Err(WritePhase::PostRename(e.to_string()));
2828 }
2829 Ok(())
2830}
2831
2832/// Phase-flattened wrapper for callers whose files carry no
2833/// published live view (the org-authority config writes): any
2834/// failure — pre- or post-rename — is an error to surface.
2835///
2836/// `pub(crate)`: the org-authority scaffolding (`org_authority.rs`)
2837/// writes its sibling config files with the same discipline.
2838pub(crate) fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), OrgRevocationError> {
2839 write_atomic_phased(path, bytes).map_err(|phase| match phase {
2840 WritePhase::PreRename(reason) => OrgRevocationError::Io {
2841 path: path.display().to_string(),
2842 reason,
2843 },
2844 WritePhase::PostRename(reason) => OrgRevocationError::DurabilityUncertain {
2845 path: path.display().to_string(),
2846 reason,
2847 },
2848 })
2849}
2850
2851#[cfg(test)]
2852mod tests {
2853 use super::*;
2854 use crate::adapter::net::behavior::org::OrgKeypair;
2855 use std::sync::atomic::{AtomicUsize, Ordering};
2856
2857 static TEST_DIR_SEQ: AtomicUsize = AtomicUsize::new(0);
2858
2859 /// §20 — the zero-floor rule is enforced at PARSE too, not only in
2860 /// `merge_bundle`.
2861 ///
2862 /// §14 removed zero rows at the merge entry point and left two others
2863 /// open: `from_file_bytes` accepted them from disk, and `publish`'s
2864 /// `or_insert(0)` could materialize one. A state file that already
2865 /// contained zero rows — hand-edited, or written by a build predating §14
2866 /// — therefore carried them forward through `merged = disk.clone()` on
2867 /// every subsequent write, re-opening the install-sweep stall §14
2868 /// describes.
2869 ///
2870 /// Dropped rather than rejected: a zero row is semantically identical to
2871 /// absence, so refusing the file would turn a no-op into an outage.
2872 #[test]
2873 fn parsed_state_drops_zero_floor_rows() {
2874 let scratch = Scratch::new();
2875 let path = scratch.state_path();
2876 let org_id = org().org_id();
2877 let live = member();
2878 let null = EntityId::from_bytes([0xEE; 32]);
2879
2880 // Hand-write a state file carrying BOTH a real floor and a zero row.
2881 let json = format!(
2882 r#"{{"version":{ORG_REVOCATION_STATE_VERSION},"floors":[
2883 {{"org":"{org_hex}","member":"{live_hex}","floor":7}},
2884 {{"org":"{org_hex}","member":"{null_hex}","floor":0}}
2885 ]}}"#,
2886 org_hex = hex::encode(org_id.as_bytes()),
2887 live_hex = hex::encode(live.as_bytes()),
2888 null_hex = hex::encode(null.as_bytes()),
2889 );
2890 std::fs::write(&path, json).expect("write hand-made state");
2891
2892 let state =
2893 OrgRevocationState::from_file_bytes(&std::fs::read(&path).expect("read back"), &path)
2894 .expect("a zero row must not make the file unloadable");
2895
2896 assert_eq!(
2897 state.floor_for(&org_id, &live),
2898 7,
2899 "the real floor survives"
2900 );
2901 assert_eq!(
2902 state.floor_for(&org_id, &null),
2903 0,
2904 "a zero floor reads as the implicit default either way",
2905 );
2906 assert_eq!(
2907 state.iter().count(),
2908 1,
2909 "the zero row must not be MATERIALIZED — it is what accumulates \
2910 and makes every authority install take an exclusive fold lock \
2911 per entry",
2912 );
2913 }
2914 /// §14 — a zero floor is the implicit default, so it must not be
2915 /// materialized into the persisted state.
2916 ///
2917 /// `floor_for` returns 0 for an absent key, so a stored `floor: 0` row
2918 /// says exactly nothing — and `floors` is never pruned, so those rows
2919 /// accumulate permanently. They are not just disk noise: the authority
2920 /// install sweep walks the whole snapshot and calls
2921 /// `retract_floored_ownership` per entry, which takes an EXCLUSIVE fold
2922 /// write lock. An org that has named N members over its lifetime made
2923 /// every install pay N sequential exclusive acquisitions, nearly all of
2924 /// which can retract nothing (`generation < 0` is unsatisfiable for u32).
2925 ///
2926 /// Red-witness: restoring the unconditional `or_insert(0)` puts the
2927 /// zero-floor member in the map and fails the length assertion.
2928 #[test]
2929 fn a_zero_floor_is_not_persisted() {
2930 let org = org();
2931 let zero_member = crate::adapter::net::identity::EntityKeypair::generate()
2932 .entity_id()
2933 .clone();
2934 let real_member = crate::adapter::net::identity::EntityKeypair::generate()
2935 .entity_id()
2936 .clone();
2937
2938 let mut map = BTreeMap::new();
2939 map.insert(zero_member.clone(), 0u32);
2940 map.insert(real_member.clone(), 3u32);
2941 let bundle = OrgRevocationBundle::try_issue(&org, &map).expect("bundle");
2942
2943 let mut state = OrgRevocationState::empty();
2944 let raised = state.merge_bundle(&bundle);
2945
2946 assert_eq!(raised, 1, "only the nonzero floor counts as a raise");
2947 assert_eq!(
2948 state.floors.len(),
2949 1,
2950 "the zero floor must not materialize a row; got {:?}",
2951 state.floors,
2952 );
2953 // Semantics are unchanged either way — that is the point.
2954 assert_eq!(state.floor_for(&org.org_id(), &zero_member), 0);
2955 assert_eq!(state.floor_for(&org.org_id(), &real_member), 3);
2956
2957 // And a later REAL floor for that member still lands.
2958 let mut map = BTreeMap::new();
2959 map.insert(zero_member.clone(), 5u32);
2960 let bundle = OrgRevocationBundle::try_issue(&org, &map).expect("bundle");
2961 assert_eq!(state.merge_bundle(&bundle), 1);
2962 assert_eq!(state.floor_for(&org.org_id(), &zero_member), 5);
2963 }
2964
2965 /// §13 — the poison key memo must not cache the PRE-CREATION fallback.
2966 ///
2967 /// `poison_path_key` falls back to the normalized path when
2968 /// `canonicalize` fails, which is the normal state during a fresh `init`
2969 /// before the state file exists. Memoizing that fallback would pin the
2970 /// non-canonical path as the key forever, so once the file appeared a
2971 /// case-alias (or any other path spelling resolving to the same file)
2972 /// would look up a DIFFERENT key and miss its tombstone — silently
2973 /// un-poisoning a store that must stay fail-closed.
2974 ///
2975 /// Red-witness: memoizing the `Err` branch makes the post-creation key
2976 /// equal the pre-creation fallback, failing the final assertion on any
2977 /// platform where `canonicalize` rewrites the path (it prefixes `\?\`
2978 /// on Windows and resolves `..`/symlinks everywhere).
2979 #[test]
2980 fn poison_key_memo_does_not_cache_the_pre_creation_fallback() {
2981 let scratch = Scratch::new();
2982 // A spelling `canonicalize` will REWRITE: descend then come back up,
2983 // which it resolves away. This makes the pre- and post-creation keys
2984 // observably different on every platform.
2985 let indirect = scratch.0.join("sub").join("..").join("state.json");
2986 std::fs::create_dir_all(scratch.0.join("sub")).expect("mkdir");
2987
2988 // Before creation: canonicalize fails, so we get the fallback.
2989 let before = poison_path_key(&indirect);
2990 assert_eq!(
2991 before,
2992 indirect.to_path_buf(),
2993 "a missing file falls back to the normalized path",
2994 );
2995
2996 // Create it, then ask again. The memo must NOT have pinned the
2997 // fallback — the real canonical key has to win now.
2998 std::fs::write(&indirect, b"{}").expect("write state");
2999 let after = poison_path_key(&indirect);
3000 let expected = std::fs::canonicalize(&indirect).expect("canonicalize");
3001 assert_eq!(
3002 after, expected,
3003 "once the file exists the canonical key must be used",
3004 );
3005 assert_ne!(
3006 after, before,
3007 "the pre-creation fallback must not have been memoized",
3008 );
3009
3010 // And the successful resolution IS memoized — a second call agrees.
3011 assert_eq!(poison_path_key(&indirect), expected, "memo is stable");
3012 }
3013
3014 /// Unique per-test scratch dir (house pattern — no tempfile dev-dep).
3015 /// A scratch directory for one test's state file and `.lock` sidecar.
3016 ///
3017 /// **It deliberately does not delete itself.** The path is already unique
3018 /// per test through `TEST_DIR_SEQ`, so cleanup would only reclaim disk —
3019 /// and on unix it would do something far worse than save a few bytes.
3020 ///
3021 /// [`BackingId::FileId`] keys the PROCESS-GLOBAL core registry by the
3022 /// sidecar's `(device, inode)`, so that two path aliases of one sidecar
3023 /// share one live view (AV-9). Deleting a finished test's directory frees
3024 /// that inode, Linux reuses a freed inode immediately, and the NEXT store
3025 /// opened anywhere in this test binary can land on it — deriving the same
3026 /// `BackingId` and having [`join_or_create_core`] join the finished test's
3027 /// still-live core by design, floors, poison bit, generation and
3028 /// exhaustion latch included.
3029 ///
3030 /// The victims are whichever tests happen to be scheduled next, so it
3031 /// presents as unrelated witnesses failing in varying combinations rather
3032 /// than as one deterministic break. It did exactly that to four
3033 /// `org_routing_wiring_tests` witnesses on Linux CI, which share this test
3034 /// binary; the fixture there carries the same note and the same fix.
3035 struct Scratch(PathBuf);
3036 impl Scratch {
3037 fn new() -> Self {
3038 let dir = std::env::temp_dir().join(format!(
3039 "net-org-revocation-{}-{}",
3040 std::process::id(),
3041 TEST_DIR_SEQ.fetch_add(1, Ordering::Relaxed)
3042 ));
3043 std::fs::create_dir_all(&dir).expect("create scratch dir");
3044 Self(dir)
3045 }
3046 fn state_path(&self) -> PathBuf {
3047 self.0.join("revocation-state.json")
3048 }
3049 }
3050
3051 fn org() -> OrgKeypair {
3052 OrgKeypair::from_bytes([0x42u8; 32])
3053 }
3054
3055 fn member() -> EntityId {
3056 EntityId::from_bytes([0x24u8; 32])
3057 }
3058
3059 fn bundle_with_floor(generation: u32) -> OrgRevocationBundle {
3060 let mut floors = BTreeMap::new();
3061 floors.insert(member(), generation);
3062 OrgRevocationBundle::try_issue(&org(), &floors).expect("issue")
3063 }
3064
3065 /// AV-9 item 9: on a case-INSENSITIVE filesystem, two
3066 /// differently-cased aliases of one backing file resolve to the
3067 /// SAME `.lock` inode, so they must collapse to ONE core (shared
3068 /// live view + publish lock + poison) — not split as the pre-AV-9
3069 /// literal-cased path key did. On a case-SENSITIVE filesystem the
3070 /// two names ARE distinct files, so the assertions are skipped (the
3071 /// fix is correctly a no-op there).
3072 ///
3073 /// Red-witness (on a case-insensitive FS): reverting the CORES /
3074 /// PATH_POISON key from [`BackingId`] to the normalized path makes
3075 /// the two aliases distinct keys — `shares_core_with` is then false
3076 /// and this fails.
3077 #[test]
3078 fn case_aliased_paths_share_one_core_on_case_insensitive_fs() {
3079 let scratch = Scratch::new();
3080 let lower = scratch.0.join("revocation-state.json");
3081 let upper = scratch.0.join("REVOCATION-STATE.JSON");
3082
3083 let a = OrgRevocationStore::init(&lower, ProvisioningExpectation::MayBeFresh)
3084 .expect("init lower alias");
3085
3086 // Probe the filesystem: does the upper-cased alias resolve to
3087 // the file just created? If not (case-sensitive FS), there is
3088 // no alias to unify and the fix is a no-op.
3089 if !upper.exists() {
3090 return;
3091 }
3092
3093 let b = OrgRevocationStore::open_existing(&upper).expect("open upper alias");
3094 assert!(
3095 a.shares_core_with(&b),
3096 "case-aliases on a case-insensitive FS must share ONE core (same .lock inode)",
3097 );
3098
3099 // A floor published through one alias is visible through the
3100 // other immediately (shared live view).
3101 a.apply_bundle(&bundle_with_floor(5))
3102 .expect("apply floor via lower alias");
3103 assert_eq!(
3104 b.floor_for(&org().org_id(), &member()),
3105 5,
3106 "a floor published through one alias must be visible through the other",
3107 );
3108
3109 // Poison registered through one alias is visible through the
3110 // other (shared poison entry).
3111 a.mark_poisoned_for_test();
3112 assert!(
3113 b.is_poisoned(),
3114 "poison under one alias must be visible through the other",
3115 );
3116 }
3117
3118 #[test]
3119 fn init_creates_empty_state_and_open_existing_loads_it() {
3120 let scratch = Scratch::new();
3121 let path = scratch.state_path();
3122
3123 let store =
3124 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
3125 assert!(store.snapshot().is_empty());
3126 assert!(path.exists());
3127
3128 let reopened = OrgRevocationStore::open_existing(&path).expect("open");
3129 assert!(reopened.snapshot().is_empty());
3130 }
3131
3132 #[test]
3133 fn open_existing_refuses_missing_state() {
3134 let scratch = Scratch::new();
3135 let err = OrgRevocationStore::open_existing(scratch.state_path())
3136 .expect_err("missing file must be loud");
3137 assert!(matches!(err, OrgRevocationError::MissingState { .. }));
3138 }
3139
3140 #[test]
3141 fn apply_bundle_raises_persists_and_publishes() {
3142 let scratch = Scratch::new();
3143 let store =
3144 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
3145 .expect("init");
3146
3147 let raised = store.apply_bundle(&bundle_with_floor(5)).expect("apply");
3148 assert_eq!(raised, vec![(org().org_id(), member(), 5)]);
3149 assert_eq!(store.floor_for(&org().org_id(), &member()), 5);
3150
3151 // Persisted: a fresh open (simulated restart) sees floor 5.
3152 drop(store);
3153 let reopened = OrgRevocationStore::open_existing(scratch.state_path()).expect("open");
3154 assert_eq!(reopened.floor_for(&org().org_id(), &member()), 5);
3155 }
3156
3157 /// The OA-1 exit-gate restart witness, verbatim:
3158 ///
3159 /// ```text
3160 /// load floor generation 5 → persist
3161 /// replace operator bundle with VALID generation 3
3162 /// restart
3163 /// → generation 5 remains authoritative
3164 /// ```
3165 #[test]
3166 fn restart_witness_lower_valid_bundle_never_rolls_back() {
3167 let scratch = Scratch::new();
3168 let path = scratch.state_path();
3169
3170 // Load floor generation 5 → persist.
3171 let store =
3172 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
3173 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
3174 drop(store);
3175
3176 // "Replace the operator bundle with VALID generation 3" +
3177 // restart: the persisted maxima, not the bundle file, is
3178 // what survives.
3179 let store = OrgRevocationStore::open_existing(&path).expect("restart");
3180 let before = std::fs::read(&path).expect("read state");
3181 let raised = store
3182 .apply_bundle(&bundle_with_floor(3))
3183 .expect("valid lower bundle is not an error");
3184 assert!(raised.is_empty(), "lower floor must not merge");
3185 assert_eq!(store.floor_for(&org().org_id(), &member()), 5);
3186 // No-op reload leaves the persisted file byte-identical.
3187 assert_eq!(std::fs::read(&path).expect("read state"), before);
3188
3189 // Second restart: generation 5 still authoritative.
3190 drop(store);
3191 let store = OrgRevocationStore::open_existing(&path).expect("restart 2");
3192 assert_eq!(store.floor_for(&org().org_id(), &member()), 5);
3193 }
3194
3195 #[test]
3196 fn corrupt_incoming_bundle_keeps_last_good() {
3197 let scratch = Scratch::new();
3198 let store =
3199 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
3200 .expect("init");
3201 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
3202
3203 // Tamper the signature of a higher-generation bundle.
3204 let mut evil = bundle_with_floor(9);
3205 evil.signature[0] ^= 1;
3206 let before = std::fs::read(store.path()).expect("read state");
3207 let err = store
3208 .apply_bundle(&evil)
3209 .expect_err("tampered bundle rejected");
3210 assert!(matches!(err, OrgRevocationError::InvalidBundle(_)));
3211 // Live view AND persisted file untouched.
3212 assert_eq!(store.floor_for(&org().org_id(), &member()), 5);
3213 assert_eq!(std::fs::read(store.path()).expect("read state"), before);
3214 }
3215
3216 #[test]
3217 fn corrupt_persisted_state_is_loud_at_startup() {
3218 let scratch = Scratch::new();
3219 let path = scratch.state_path();
3220 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
3221
3222 std::fs::write(&path, b"{ not json").expect("corrupt");
3223 let err = OrgRevocationStore::open_existing(&path).expect_err("corrupt is loud");
3224 assert!(matches!(err, OrgRevocationError::CorruptState { .. }));
3225
3226 // Unsupported version is equally loud.
3227 std::fs::write(&path, br#"{"version":99,"floors":[]}"#).expect("write");
3228 let err = OrgRevocationStore::open_existing(&path).expect_err("version is loud");
3229 assert!(matches!(
3230 err,
3231 OrgRevocationError::UnsupportedVersion { found: 99, .. }
3232 ));
3233
3234 // Duplicate (org, member) keys are corruption.
3235 let org_hex = hex::encode(org().org_id().as_bytes());
3236 let member_hex = hex::encode(member().as_bytes());
3237 let dup = format!(
3238 r#"{{"version":1,"floors":[
3239 {{"org":"{org_hex}","member":"{member_hex}","floor":1}},
3240 {{"org":"{org_hex}","member":"{member_hex}","floor":2}}
3241 ]}}"#
3242 );
3243 std::fs::write(&path, dup).expect("write");
3244 let err = OrgRevocationStore::open_existing(&path).expect_err("dup is loud");
3245 assert!(matches!(err, OrgRevocationError::CorruptState { .. }));
3246 }
3247
3248 #[test]
3249 fn init_preserves_existing_maxima_on_readopt() {
3250 let scratch = Scratch::new();
3251 let path = scratch.state_path();
3252 let store =
3253 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
3254 store.apply_bundle(&bundle_with_floor(5)).expect("apply");
3255 drop(store);
3256
3257 // Re-running adopt must NOT reset floors to empty.
3258 let readopted =
3259 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("re-init");
3260 assert_eq!(readopted.floor_for(&org().org_id(), &member()), 5);
3261 }
3262
3263 /// The sibling of the test above, for the case it does NOT cover: the file
3264 /// is not merely stale, it is GONE.
3265 ///
3266 /// `init` used to read that as a first adopt and durably write an empty
3267 /// state — un-revoking every certificate the org had retired — while
3268 /// `open_existing` read the identical situation as `MissingState` and
3269 /// refused. The permissive reading was on `net node adopt`, the path an
3270 /// operator reaches for when something already looks wrong.
3271 ///
3272 /// The `.lock` sidecar outlives the state file and is the evidence that
3273 /// this store was provisioned before. Note the caller still says
3274 /// `MayBeFresh` here: this asserts the store defends itself even when the
3275 /// caller believes a fresh adopt is plausible.
3276 /// §13 — the durable publish goes through `MOVEFILE_WRITE_THROUGH` on
3277 /// Windows, and it must behave exactly like the plain rename it replaces.
3278 ///
3279 /// A unit test cannot prove crash durability. What it CAN pin is that the
3280 /// write-through path is the one actually taken, that it REPLACES an
3281 /// existing destination (the flag combination is easy to get wrong —
3282 /// omitting `MOVEFILE_REPLACE_EXISTING` would fail on every republish),
3283 /// and that a genuine failure still surfaces as an error rather than
3284 /// being swallowed.
3285 ///
3286 /// Recorded because the first attempt at this fix was WRONG: it used
3287 /// `FlushFileBuffers` on a `FILE_FLAG_BACKUP_SEMANTICS` directory handle,
3288 /// by analogy with the POSIX parent fsync. That is not a supported
3289 /// operation on Windows — it returns `ERROR_ACCESS_DENIED` and failed
3290 /// every store test here. There is no directory fsync; the durability
3291 /// primitive is on the rename.
3292 #[cfg(windows)]
3293 #[test]
3294 fn write_through_rename_replaces_and_reports_failure() {
3295 let scratch = Scratch::new();
3296 let dest = scratch.state_path();
3297 let src = scratch.0.join("staged.json");
3298
3299 std::fs::write(&dest, b"old").expect("seed destination");
3300 std::fs::write(&src, b"new").expect("seed source");
3301
3302 rename_write_through(&src, &dest).expect("write-through rename must succeed");
3303 assert_eq!(
3304 std::fs::read(&dest).expect("read dest"),
3305 b"new",
3306 "the rename must REPLACE an existing destination — without \
3307 MOVEFILE_REPLACE_EXISTING every republish would fail",
3308 );
3309 assert!(!src.exists(), "the source must be consumed by the move");
3310
3311 // A missing source is an error, not a silent success.
3312 let ghost = scratch.0.join("does-not-exist.json");
3313 assert!(
3314 rename_write_through(&ghost, &dest).is_err(),
3315 "a failed move must surface as an error; swallowing it would make \
3316 a lost publish look durable",
3317 );
3318 }
3319 /// The sibling of the test above, for the case it does NOT cover: the file
3320 /// is not merely stale, it is GONE.
3321 ///
3322 /// `init` used to read that as a first adopt and durably write an empty
3323 /// state — un-revoking every certificate the org had retired — while
3324 /// `open_existing` read the identical situation as `MissingState` and
3325 /// refused. The permissive reading was on `net node adopt`, the path an
3326 /// operator reaches for when something already looks wrong.
3327 ///
3328 /// The `.lock` sidecar outlives the state file and is the evidence that
3329 /// this store was provisioned before. Note the caller still says
3330 /// `MayBeFresh` here: this asserts the store defends itself even when the
3331 /// caller believes a fresh adopt is plausible.
3332 #[test]
3333 fn init_refuses_to_recreate_a_state_file_that_was_deleted() {
3334 let scratch = Scratch::new();
3335 let path = scratch.state_path();
3336 let store = OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh)
3337 .expect("first adopt");
3338 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
3339 drop(store);
3340
3341 // Config management / a restore / a selective delete removes the state
3342 // file. The sidecar stays.
3343 std::fs::remove_file(&path).expect("remove state file");
3344 let mut sidecar = path.as_os_str().to_os_string();
3345 sidecar.push(".lock");
3346 assert!(
3347 std::fs::symlink_metadata(PathBuf::from(sidecar)).is_ok(),
3348 "precondition: the sidecar must outlive the state file, otherwise \
3349 this test proves nothing about the signal under test",
3350 );
3351
3352 let err = OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh)
3353 .expect_err("a deleted state file must not be re-created as empty");
3354 assert!(
3355 matches!(err, OrgRevocationError::MissingState { .. }),
3356 "expected MissingState, got {err:?}",
3357 );
3358
3359 // And the refusal is fail-closed: nothing was written in its place, so
3360 // a later repair still sees an absent file rather than an empty one
3361 // that silently reads as "no floors".
3362 assert!(
3363 !path.exists(),
3364 "the refusal wrote an empty state anyway — the floors are gone",
3365 );
3366 }
3367
3368 /// The residual the sidecar signal cannot cover: BOTH files removed. Only
3369 /// the caller can tell, because only the caller sees the rest of the
3370 /// authority directory — which is why the expectation is a parameter and
3371 /// not merely an internal check.
3372 #[test]
3373 fn init_honours_the_callers_must_exist_expectation() {
3374 let scratch = Scratch::new();
3375 let path = scratch.state_path();
3376 let store = OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh)
3377 .expect("first adopt");
3378 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
3379 drop(store);
3380
3381 std::fs::remove_file(&path).expect("remove state file");
3382 let mut sidecar = path.as_os_str().to_os_string();
3383 sidecar.push(".lock");
3384 let _ = std::fs::remove_file(PathBuf::from(sidecar));
3385
3386 let err = OrgRevocationStore::init(&path, ProvisioningExpectation::MustExist)
3387 .expect_err("MustExist must refuse an absent state file");
3388 assert!(
3389 matches!(err, OrgRevocationError::MissingState { .. }),
3390 "expected MissingState, got {err:?}",
3391 );
3392 }
3393
3394 /// Positive control for both refusals: a genuinely fresh path still adopts.
3395 /// Without this, a regression that refused unconditionally would pass the
3396 /// two tests above.
3397 #[test]
3398 fn init_still_creates_a_genuinely_fresh_store() {
3399 let scratch = Scratch::new();
3400 let path = scratch.state_path();
3401 let store = OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh)
3402 .expect("a first adopt on a clean path must succeed");
3403 assert_eq!(store.floor_for(&org().org_id(), &member()), 0);
3404 assert!(path.exists(), "a fresh adopt must durably create the state");
3405 }
3406
3407 #[test]
3408 fn merge_is_per_key_monotone_across_orgs_and_members() {
3409 let org_a = OrgKeypair::from_bytes([1u8; 32]);
3410 let org_b = OrgKeypair::from_bytes([2u8; 32]);
3411 let m1 = EntityId::from_bytes([11u8; 32]);
3412 let m2 = EntityId::from_bytes([22u8; 32]);
3413
3414 let mut state = OrgRevocationState::empty();
3415
3416 let mut floors = BTreeMap::new();
3417 floors.insert(m1.clone(), 5);
3418 floors.insert(m2.clone(), 2);
3419 let a1 = OrgRevocationBundle::try_issue(&org_a, &floors).expect("issue");
3420 assert_eq!(state.merge_bundle(&a1), 2);
3421
3422 // Same members under a DIFFERENT org are independent keys.
3423 let b1 = OrgRevocationBundle::try_issue(&org_b, &floors).expect("issue");
3424 assert_eq!(state.merge_bundle(&b1), 2);
3425 assert_eq!(state.floor_for(&org_a.org_id(), &m1), 5);
3426 assert_eq!(state.floor_for(&org_b.org_id(), &m1), 5);
3427
3428 // Mixed raise/no-op within one bundle: m1 lower (no-op),
3429 // m2 higher (raises).
3430 let mut floors = BTreeMap::new();
3431 floors.insert(m1.clone(), 3);
3432 floors.insert(m2.clone(), 7);
3433 let a2 = OrgRevocationBundle::try_issue(&org_a, &floors).expect("issue");
3434 assert_eq!(state.merge_bundle(&a2), 1);
3435 assert_eq!(state.floor_for(&org_a.org_id(), &m1), 5);
3436 assert_eq!(state.floor_for(&org_a.org_id(), &m2), 7);
3437 // Unknown keys floor at 0.
3438 assert_eq!(
3439 state.floor_for(&org_a.org_id(), &EntityId::from_bytes([99u8; 32])),
3440 0
3441 );
3442 }
3443
3444 #[cfg(unix)]
3445 #[test]
3446 fn persist_failure_never_publishes_the_live_view() {
3447 let scratch = Scratch::new();
3448 let path = scratch.state_path();
3449 let store =
3450 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
3451 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
3452
3453 // Force the atomic rename to fail: replace the state file
3454 // with a non-empty DIRECTORY at the same path.
3455 std::fs::remove_file(&path).expect("remove");
3456 std::fs::create_dir(&path).expect("dir at path");
3457 std::fs::write(path.join("occupied"), b"x").expect("occupy");
3458
3459 let err = store
3460 .apply_bundle(&bundle_with_floor(9))
3461 .expect_err("rename onto non-empty dir must fail");
3462 assert!(matches!(err, OrgRevocationError::Io { .. }));
3463 // The live view still serves the last DURABLE floor — the
3464 // undurable 9 is never enforced. Pre-rename failure does
3465 // NOT poison: nothing on disk changed.
3466 assert_eq!(store.floor_for(&org().org_id(), &member()), 5);
3467 assert!(!store.is_poisoned());
3468
3469 // No temp files left behind by the failed write.
3470 let leftovers: Vec<_> = std::fs::read_dir(&scratch.0)
3471 .expect("read scratch")
3472 .filter_map(|e| e.ok())
3473 .filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
3474 .collect();
3475 assert!(leftovers.is_empty(), "stale temps: {leftovers:?}");
3476 }
3477
3478 fn bundle_for(member: EntityId, generation: u32) -> OrgRevocationBundle {
3479 let mut floors = BTreeMap::new();
3480 floors.insert(member, generation);
3481 OrgRevocationBundle::try_issue(&org(), &floors).expect("issue")
3482 }
3483
3484 /// Review-8 §5 + review-9 addendum witness: two store handles
3485 /// on one file share ONE core — a sibling observes a raise the
3486 /// instant it publishes (never a stale independent view) — and
3487 /// the under-lock REREAD keeps every maximum in the persisted
3488 /// file (the reread still guards CROSS-PROCESS writers, which
3489 /// cannot share a core).
3490 #[test]
3491 fn same_path_handles_share_one_live_view_and_preserve_all_maxima() {
3492 let scratch = Scratch::new();
3493 let path = scratch.state_path();
3494 let member_x = EntityId::from_bytes([0xAAu8; 32]);
3495 let member_y = EntityId::from_bytes([0xBBu8; 32]);
3496
3497 let store_a =
3498 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init A");
3499 let store_b = OrgRevocationStore::open_existing(&path).expect("open B");
3500 assert!(
3501 store_a.shares_core_with(&store_b),
3502 "same normalized path must join one core"
3503 );
3504
3505 // A raises member_x to 5; B's view advances IMMEDIATELY —
3506 // one backing file is never two security views (review-9
3507 // addendum).
3508 store_a
3509 .apply_bundle(&bundle_for(member_x.clone(), 5))
3510 .expect("A applies x=5");
3511 assert_eq!(store_b.floor_for(&org().org_id(), &member_x), 5);
3512
3513 // B raises member_y to 7; the shared view means only y
3514 // newly rises, and the persisted file carries BOTH maxima.
3515 let raised = store_b
3516 .apply_bundle(&bundle_for(member_y.clone(), 7))
3517 .expect("B applies y=7");
3518 assert_eq!(raised, vec![(org().org_id(), member_y.clone(), 7)]);
3519
3520 let reopened = OrgRevocationStore::open_existing(&path).expect("reopen");
3521 assert_eq!(reopened.floor_for(&org().org_id(), &member_x), 5);
3522 assert_eq!(reopened.floor_for(&org().org_id(), &member_y), 7);
3523 assert!(reopened.shares_core_with(&store_a));
3524 }
3525
3526 /// Review-9 addendum: `open_existing` has NO pre-lock poison
3527 /// fast path — an opener serializes behind the state lock, and
3528 /// a poison bit registered while it waited gates it. Recovery
3529 /// rereads the FINAL persisted state and returns it, never a
3530 /// stale pre-write view.
3531 #[test]
3532 fn fresh_open_serializes_behind_the_state_lock_and_recovers_poison() {
3533 let scratch = Scratch::new();
3534 let path = scratch.state_path();
3535 drop(OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init"));
3536 let norm = normalize_backing_path(&path).expect("normalize");
3537
3538 // Writer holds the interprocess lock…
3539 let lock = lock_state_file(&norm).expect("lock");
3540
3541 let opener_path = path.clone();
3542 let (started_tx, started_rx) = std::sync::mpsc::channel();
3543 let (done_tx, done_rx) = std::sync::mpsc::channel();
3544 let opener = std::thread::spawn(move || {
3545 started_tx.send(()).expect("send started");
3546 let result = OrgRevocationStore::open_existing(&opener_path);
3547 done_tx.send(()).expect("send done");
3548 result
3549 });
3550 started_rx
3551 .recv_timeout(std::time::Duration::from_secs(5))
3552 .expect("opener started");
3553 // …so the opener must NOT complete while the lock is held.
3554 assert!(
3555 done_rx
3556 .recv_timeout(std::time::Duration::from_millis(300))
3557 .is_err(),
3558 "open_existing must serialize behind the state lock"
3559 );
3560
3561 // Still under the lock: the writer lands a stronger state
3562 // and (simulating a failed post-rename parent fsync)
3563 // registers the path-wide poison.
3564 let mut stronger = OrgRevocationState::empty();
3565 stronger.merge_bundle(&bundle_with_floor(9));
3566 write_atomic(&norm, &stronger.to_file_bytes().expect("bytes")).expect("write");
3567 mark_poisoned(&BackingId::of(&lock, &norm).expect("backing id"), &norm);
3568
3569 // Lock releases → the opener proceeds: it must observe the
3570 // poison, recover (reread + successful parent fsync), and
3571 // return the FINAL floor — never the pre-write view.
3572 drop(lock);
3573 let opened = opener
3574 .join()
3575 .expect("join opener")
3576 .expect("open recovers and succeeds");
3577 assert_eq!(opened.floor_for(&org().org_id(), &member()), 9);
3578 assert!(
3579 !opened.is_poisoned(),
3580 "successful recovery clears the path-wide bit"
3581 );
3582 }
3583
3584 /// Review-11 P1: an opener publishing through an EXISTING core
3585 /// obeys the same `PublishGuard` a replacement holds. While the
3586 /// guard is held, a same-path opener cannot publish its
3587 /// (stronger) floor — it blocks until the guard drops, so a
3588 /// replacement's dominance comparison and swap see a frozen
3589 /// live view. This is the store-level root of the review-10 red
3590 /// (opener published floor 10 inside a held guard).
3591 #[test]
3592 fn opener_cannot_publish_through_a_held_publish_guard() {
3593 let scratch = Scratch::new();
3594 let path = scratch.state_path();
3595 // store_a creates and keeps the core alive at floor 0.
3596 let store_a =
3597 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
3598
3599 // A stronger state is already durable on disk (an operator
3600 // bundle another writer persisted); an opener would read and
3601 // publish floor 10 through the shared core.
3602 let norm = normalize_backing_path(&path).expect("normalize");
3603 let mut stronger = OrgRevocationState::empty();
3604 stronger.merge_bundle(&bundle_with_floor(10));
3605 {
3606 let _lk = lock_state_file(&norm).expect("lock");
3607 write_atomic(&norm, &stronger.to_file_bytes().expect("bytes")).expect("write");
3608 }
3609
3610 // Hold the publish guard (what a replacement holds across
3611 // dominance→swap).
3612 let guard = store_a.publish_guard();
3613
3614 let opener_path = path.clone();
3615 let (done_tx, done_rx) = std::sync::mpsc::channel();
3616 let opener = std::thread::spawn(move || {
3617 let s = OrgRevocationStore::open_existing(&opener_path).expect("open");
3618 done_tx.send(()).expect("done");
3619 s
3620 });
3621 // The opener must NOT publish while the guard is held: the
3622 // shared live view stays at floor 0.
3623 assert!(
3624 done_rx
3625 .recv_timeout(std::time::Duration::from_millis(300))
3626 .is_err(),
3627 "opener published inside a held PublishGuard"
3628 );
3629 assert_eq!(
3630 store_a.floor_for(&org().org_id(), &member()),
3631 0,
3632 "the guarded live view must not move under an opener"
3633 );
3634
3635 // Releasing the guard lets the opener publish; the shared
3636 // view then advances to 10.
3637 drop(guard);
3638 let opened = opener.join().expect("join");
3639 assert_eq!(opened.floor_for(&org().org_id(), &member()), 10);
3640 assert_eq!(store_a.floor_for(&org().org_id(), &member()), 10);
3641 }
3642
3643 /// Review-9 addendum: the raise-observer registry supports
3644 /// multiple subscribers — registering a second observer never
3645 /// steals the first one's notifications, same-path handles'
3646 /// callbacks all fire, and a token unsubscribes only its own
3647 /// registration.
3648 #[test]
3649 fn multiple_subscribers_on_one_path_all_observe_raises() {
3650 let scratch = Scratch::new();
3651 let path = scratch.state_path();
3652 let store_a =
3653 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init A");
3654 let store_b = OrgRevocationStore::open_existing(&path).expect("open B");
3655
3656 let seen_a: Arc<Mutex<Vec<u32>>> = Arc::new(Mutex::new(Vec::new()));
3657 let seen_b: Arc<Mutex<Vec<u32>>> = Arc::new(Mutex::new(Vec::new()));
3658 let seen_tok: Arc<Mutex<Vec<u32>>> = Arc::new(Mutex::new(Vec::new()));
3659 let sink = seen_a.clone();
3660 let _sub_a = store_a.subscribe_floors_raised(move |raised| {
3661 sink.lock().extend(raised.iter().map(|(_, _, f)| *f));
3662 });
3663 let sink = seen_b.clone();
3664 let _sub_b = store_b.subscribe_floors_raised(move |raised| {
3665 sink.lock().extend(raised.iter().map(|(_, _, f)| *f));
3666 });
3667 let sink = seen_tok.clone();
3668 let subscription = store_a.subscribe_floors_raised(move |raised| {
3669 sink.lock().extend(raised.iter().map(|(_, _, f)| *f));
3670 });
3671
3672 // One raise through A notifies EVERY registration —
3673 // including B's, which observes the raise through the
3674 // shared core (previously the review-9 addendum red: only
3675 // the final `set_on_floors_raised` caller was notified).
3676 store_a
3677 .apply_bundle(&bundle_with_floor(5))
3678 .expect("apply 5");
3679 assert_eq!(*seen_a.lock(), vec![5]);
3680 assert_eq!(*seen_b.lock(), vec![5]);
3681 assert_eq!(*seen_tok.lock(), vec![5]);
3682
3683 // Dropping the RAII guard removes ONLY that registration.
3684 drop(subscription);
3685 store_b
3686 .apply_bundle(&bundle_with_floor(7))
3687 .expect("apply 7");
3688 assert_eq!(*seen_a.lock(), vec![5, 7]);
3689 assert_eq!(*seen_b.lock(), vec![5, 7]);
3690 assert_eq!(*seen_tok.lock(), vec![5], "unsubscribed token is silent");
3691 }
3692
3693 /// R2-2: `subscribe_floors_raised` hands back an externally-owned
3694 /// RAII guard; dropping the GUARD retires the subscription even while
3695 /// the owning store facade is still very much alive — removal goes
3696 /// through the guard's `Weak<StoreCore>`, not the facade's `Drop`
3697 /// (which a `core → callback → Arc<store> → core` capture cycle could
3698 /// keep from ever running).
3699 ///
3700 /// Red-witness: making `RaiseSubscription::drop` skip
3701 /// `core.remove_subscriber` leaves the count at 1.
3702 #[test]
3703 fn dropping_the_subscription_guard_unsubscribes_while_the_store_lives() {
3704 let scratch = Scratch::new();
3705 let store =
3706 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
3707 .expect("init");
3708 assert_eq!(store.subscriber_count(), 0);
3709 let subscription = store.subscribe_floors_raised(|_raised| {});
3710 assert_eq!(
3711 store.subscriber_count(),
3712 1,
3713 "subscribe registers one callback"
3714 );
3715 drop(subscription);
3716 assert_eq!(
3717 store.subscriber_count(),
3718 0,
3719 "dropping the guard unsubscribed while the store handle is still alive",
3720 );
3721 }
3722
3723 /// R2-3: teardown EXCLUDES an in-flight callback. A callback that has
3724 /// passed the liveness check and is mid-body keeps the subscription's
3725 /// Drop BLOCKED (draining the exclusion lease) until it leaves — so a
3726 /// retraction can never be torn in half by a concurrent teardown, and
3727 /// no new callback starts once teardown has begun.
3728 ///
3729 /// Deterministic barrier: the callback signals `entered` (now counted
3730 /// in-flight) and blocks; a teardown thread drops the guard and must
3731 /// park in `kill_and_drain`; while parked it provably cannot signal
3732 /// completion (asserted via a bounded `recv_timeout` that MUST expire);
3733 /// releasing the callback lets it leave, the drain completes, and only
3734 /// then does teardown finish.
3735 ///
3736 /// Red-witness: dropping the `while in_flight > 0` drain loop in
3737 /// `kill_and_drain` lets teardown complete while the callback is still
3738 /// in-flight, so the "must block" `recv_timeout` receives early and the
3739 /// assertion fails.
3740 #[test]
3741 fn teardown_blocks_until_an_in_flight_callback_leaves() {
3742 use std::sync::mpsc;
3743 use std::time::Duration;
3744
3745 let scratch = Scratch::new();
3746 let store = Arc::new(
3747 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
3748 .expect("init"),
3749 );
3750
3751 let (entered_tx, entered_rx) = mpsc::channel::<()>();
3752 let (release_tx, release_rx) = mpsc::channel::<()>();
3753 // The callback must be `Fn + Send + Sync`; the mpsc endpoints are
3754 // `!Sync`, so guard them.
3755 let entered_tx = Mutex::new(entered_tx);
3756 let release_rx = Mutex::new(release_rx);
3757 let ran = Arc::new(AtomicUsize::new(0));
3758 let ran_cb = Arc::clone(&ran);
3759 let subscription = store.subscribe_floors_raised(move |_raised| {
3760 ran_cb.fetch_add(1, Ordering::SeqCst);
3761 entered_tx.lock().send(()).expect("signal entered");
3762 // Block INSIDE the callback body: the exclusion lease counts
3763 // this run as in-flight for the whole duration.
3764 release_rx.lock().recv().expect("await release");
3765 });
3766
3767 // Fire a raise on a worker thread so the callback blocks there.
3768 let store_fire = Arc::clone(&store);
3769 let fire = std::thread::spawn(move || {
3770 store_fire
3771 .apply_bundle(&bundle_with_floor(5))
3772 .expect("apply 5");
3773 });
3774
3775 // The callback is now in-flight (blocked on release).
3776 entered_rx
3777 .recv_timeout(Duration::from_secs(2))
3778 .expect("callback entered");
3779
3780 // Tear down on another thread; it must PARK in kill_and_drain.
3781 let (teardown_done_tx, teardown_done_rx) = mpsc::channel::<()>();
3782 let teardown = std::thread::spawn(move || {
3783 drop(subscription);
3784 teardown_done_tx.send(()).expect("signal teardown done");
3785 });
3786
3787 // Block proof: while the callback is in-flight, teardown cannot
3788 // complete — this `recv_timeout` MUST expire.
3789 assert!(
3790 teardown_done_rx
3791 .recv_timeout(Duration::from_millis(300))
3792 .is_err(),
3793 "teardown must block while a callback is in-flight",
3794 );
3795
3796 // Release the callback → it leaves → the drain wakes → teardown
3797 // completes.
3798 release_tx.send(()).expect("release callback");
3799 teardown_done_rx
3800 .recv_timeout(Duration::from_secs(2))
3801 .expect("teardown completes after the callback drains");
3802
3803 fire.join().expect("fire thread");
3804 teardown.join().expect("teardown thread");
3805 assert_eq!(ran.load(Ordering::SeqCst), 1, "callback ran exactly once");
3806 assert_eq!(
3807 store.subscriber_count(),
3808 0,
3809 "the drained guard removed the subscriber",
3810 );
3811 }
3812
3813 /// R3-4: dropping the externally-owned guard BREAKS the
3814 /// `core → subscribers → callback → Arc<store> → Arc<core> → core`
3815 /// capture cycle, so a callback that captures `Arc<store>` no longer
3816 /// leaks the store. (The removed `set_on_floors_raised` stored its
3817 /// guard inside the facade, which that same cycle kept alive forever,
3818 /// so its drop never ran.)
3819 ///
3820 /// Red-witness: making `RaiseSubscription::drop` skip
3821 /// `remove_subscriber` leaves the capturing callback in the core, so
3822 /// the store never frees and `weak.upgrade()` stays `Some`.
3823 #[test]
3824 fn dropping_the_external_guard_breaks_a_store_capturing_cycle() {
3825 let scratch = Scratch::new();
3826 let store = Arc::new(
3827 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
3828 .expect("init"),
3829 );
3830 let weak = Arc::downgrade(&store);
3831 // The callback CAPTURES the store Arc — the exact cycle.
3832 let captured = Arc::clone(&store);
3833 let sub = store.subscribe_floors_raised(move |_raised| {
3834 let _keep = &captured;
3835 });
3836 // Dropping the external guard removes the callback, releasing its
3837 // captured `Arc<store>`; then the last external handle drops.
3838 drop(sub);
3839 drop(store);
3840 assert!(
3841 weak.upgrade().is_none(),
3842 "dropping the external guard must break the callback→store cycle so the store frees",
3843 );
3844 }
3845
3846 /// R3-4: a callback that drops its OWN guard from inside the callback
3847 /// must not deadlock. `kill_and_drain` excludes this thread's own
3848 /// in-flight frame (via the thread-local lease tracking), so it does
3849 /// not wait for the frame that is dropping it; that frame's
3850 /// `LeaveOnDrop` performs the final retirement.
3851 ///
3852 /// Red-witness: reverting `kill_and_drain` to wait for `in_flight == 0`
3853 /// unconditionally deadlocks the self-dropping callback, so the worker
3854 /// never signals and the bounded `recv_timeout` expires.
3855 #[test]
3856 fn a_callback_can_drop_its_own_guard_without_deadlock() {
3857 use std::sync::mpsc;
3858 use std::time::Duration;
3859
3860 let scratch = Scratch::new();
3861 let store = Arc::new(
3862 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
3863 .expect("init"),
3864 );
3865 // The guard lives in a slot the callback takes + drops from inside.
3866 let slot: Arc<Mutex<Option<RaiseSubscription>>> = Arc::new(Mutex::new(None));
3867 let slot_cb = Arc::clone(&slot);
3868 let sub = store.subscribe_floors_raised(move |_raised| {
3869 // Self-unsubscribe: drop this subscription's own guard.
3870 let _dropped = slot_cb.lock().take();
3871 });
3872 *slot.lock() = Some(sub);
3873
3874 let (done_tx, done_rx) = mpsc::channel::<()>();
3875 let store_t = Arc::clone(&store);
3876 let worker = std::thread::spawn(move || {
3877 store_t
3878 .apply_bundle(&bundle_with_floor(5))
3879 .expect("apply 5");
3880 done_tx.send(()).expect("signal done");
3881 });
3882 assert!(
3883 done_rx.recv_timeout(Duration::from_secs(5)).is_ok(),
3884 "a callback dropping its own guard must not deadlock",
3885 );
3886 worker.join().expect("worker joined");
3887 assert_eq!(
3888 store.subscriber_count(),
3889 0,
3890 "the self-drop removed the subscription",
3891 );
3892 }
3893
3894 /// R3-4 (cross-thread): a callback that self-unsubscribes while ANOTHER
3895 /// thread is inside a callback of the SAME subscription must not wait for
3896 /// that foreign frame — even (especially) when the foreign frame is
3897 /// blocked on a user lock the self-unsubscribing callback holds.
3898 ///
3899 /// Timeline: A enters and takes a user lock; B enters and blocks needing
3900 /// that lock; A self-unsubscribes (dropping its own guard) WHILE holding
3901 /// the lock and while B is in-flight, then releases the lock so B can
3902 /// finish.
3903 ///
3904 /// Red-witness: the pre-fix `while in_flight > own_frames` wait blocks A
3905 /// (in_flight == 2, own_frames == 1) on B; B is blocked on the user lock A
3906 /// holds; A cannot release it until the wait returns → cross-thread
3907 /// deadlock, and the bounded `recv_timeout`s below expire. `leave`'s
3908 /// notify-only-at-zero made even relaxing the threshold insufficient; the
3909 /// fix is to not wait at all when `own_frames > 0`.
3910 #[test]
3911 fn self_unsubscribe_does_not_wait_for_a_concurrent_foreign_callback() {
3912 use std::sync::mpsc;
3913 use std::time::Duration;
3914
3915 let scratch = Scratch::new();
3916 let store = Arc::new(
3917 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
3918 .expect("init"),
3919 );
3920
3921 // A user lock A holds across its self-unsubscribe and B needs.
3922 let user_lock = Arc::new(Mutex::new(()));
3923 // First entrant is role A (self-unsubscriber), second is role B.
3924 let role = Arc::new(AtomicUsize::new(0));
3925 // A's own guard, taken + dropped from inside A's callback.
3926 let slot: Arc<Mutex<Option<RaiseSubscription>>> = Arc::new(Mutex::new(None));
3927
3928 let (a_holds_tx, a_holds_rx) = mpsc::channel::<()>();
3929 let (b_entered_tx, b_entered_rx) = mpsc::channel::<()>();
3930 let (proceed_a_tx, proceed_a_rx) = mpsc::channel::<()>();
3931 // The callback is `Fn + Send + Sync`; the mpsc endpoints are `!Sync`.
3932 let a_holds_tx = Mutex::new(a_holds_tx);
3933 let b_entered_tx = Mutex::new(b_entered_tx);
3934 let proceed_a_rx = Mutex::new(proceed_a_rx);
3935
3936 let user_lock_cb = Arc::clone(&user_lock);
3937 let role_cb = Arc::clone(&role);
3938 let slot_cb = Arc::clone(&slot);
3939 let sub = store.subscribe_floors_raised(move |_raised| {
3940 if role_cb.fetch_add(1, Ordering::SeqCst) == 0 {
3941 // Role A: hold the user lock, announce, await the go-ahead,
3942 // then self-unsubscribe WHILE holding the lock and while B is
3943 // in-flight, and only then release the lock.
3944 let held = user_lock_cb.lock();
3945 a_holds_tx
3946 .lock()
3947 .send(())
3948 .expect("A announces it holds the lock");
3949 proceed_a_rx.lock().recv().expect("A awaits go-ahead");
3950 drop(slot_cb.lock().take()); // self-unsubscribe (must not block)
3951 drop(held); // release → B can proceed
3952 } else {
3953 // Role B: needs the user lock A holds.
3954 b_entered_tx.lock().send(()).expect("B announces entry");
3955 let _held = user_lock_cb.lock();
3956 }
3957 });
3958 *slot.lock() = Some(sub);
3959
3960 let (done_tx, done_rx) = mpsc::channel::<()>();
3961
3962 // Fire A on thread 1; wait until it holds the user lock (role 0 taken).
3963 let store1 = Arc::clone(&store);
3964 let done1 = done_tx.clone();
3965 let t1 = std::thread::spawn(move || {
3966 store1.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
3967 done1.send(()).expect("t1 done");
3968 });
3969 a_holds_rx
3970 .recv_timeout(Duration::from_secs(5))
3971 .expect("A entered and holds the user lock");
3972
3973 // Fire B on thread 2; wait until it has entered (in_flight == 2).
3974 let store2 = Arc::clone(&store);
3975 let done2 = done_tx.clone();
3976 let t2 = std::thread::spawn(move || {
3977 store2.apply_bundle(&bundle_with_floor(6)).expect("apply 6");
3978 done2.send(()).expect("t2 done");
3979 });
3980 b_entered_rx
3981 .recv_timeout(Duration::from_secs(5))
3982 .expect("B entered the callback");
3983
3984 // Release A: self-unsubscribe must return without waiting for B, so A
3985 // releases the user lock and BOTH workers finish within the bound.
3986 proceed_a_tx.send(()).expect("release A");
3987 for _ in 0..2 {
3988 done_rx
3989 .recv_timeout(Duration::from_secs(5))
3990 .expect("a worker deadlocked in self-unsubscribe");
3991 }
3992 t1.join().expect("thread 1 joined");
3993 t2.join().expect("thread 2 joined");
3994
3995 assert_eq!(
3996 role.load(Ordering::SeqCst),
3997 2,
3998 "exactly A and B ran (each once)",
3999 );
4000 assert_eq!(
4001 store.subscriber_count(),
4002 0,
4003 "the self-drop removed the subscription",
4004 );
4005 // No future callback enters: the subscriber is gone, so a later raise
4006 // fires nothing and the role counter stays at 2.
4007 store.apply_bundle(&bundle_with_floor(7)).expect("apply 7");
4008 assert_eq!(
4009 role.load(Ordering::SeqCst),
4010 2,
4011 "no callback runs after self-unsubscription removed the subscriber",
4012 );
4013 }
4014
4015 /// Review-9 addendum: aliased pathnames — `..` hops, `./`
4016 /// prefixes, symlinked parents — normalize onto ONE core and
4017 /// ONE poison key; no verbatim fallback survives.
4018 #[test]
4019 fn aliased_paths_share_one_core() {
4020 let scratch = Scratch::new();
4021 let sub = scratch.0.join("sub");
4022 std::fs::create_dir_all(&sub).expect("mkdir sub");
4023 let direct = sub.join("revocation-state.json");
4024 let dotted = scratch.0.join("sub/../sub/revocation-state.json");
4025
4026 let store_a = OrgRevocationStore::init(&direct, ProvisioningExpectation::MayBeFresh)
4027 .expect("init direct");
4028 let store_b = OrgRevocationStore::open_existing(&dotted).expect("open dotted alias");
4029 assert!(
4030 store_a.shares_core_with(&store_b),
4031 "`..` alias joins the core"
4032 );
4033 store_a.apply_bundle(&bundle_with_floor(5)).expect("apply");
4034 assert_eq!(store_b.floor_for(&org().org_id(), &member()), 5);
4035
4036 #[cfg(unix)]
4037 {
4038 let link = scratch.0.join("linked-sub");
4039 std::os::unix::fs::symlink(&sub, &link).expect("symlink dir");
4040 let via_link = OrgRevocationStore::open_existing(link.join("revocation-state.json"))
4041 .expect("open through symlinked parent");
4042 assert!(
4043 store_a.shares_core_with(&via_link),
4044 "symlinked-parent alias joins the core"
4045 );
4046 }
4047
4048 // Normalization invariants: bare and `./`-prefixed names
4049 // resolve absolute (no verbatim fallback)…
4050 let bare = normalize_backing_path(Path::new("bare-floors.json")).expect("bare");
4051 let dot = normalize_backing_path(Path::new("./bare-floors.json")).expect("dot");
4052 assert!(bare.is_absolute());
4053 assert_eq!(bare, dot);
4054 // …and a path that cannot normalize is refused, never keyed
4055 // verbatim.
4056 assert!(
4057 normalize_backing_path(&scratch.0.join("no-such-dir/state.json")).is_err(),
4058 "unresolvable parent must refuse"
4059 );
4060 assert!(
4061 normalize_backing_path(Path::new("..")).is_err(),
4062 "no final component must refuse"
4063 );
4064 }
4065
4066 /// Review-11 P2: the final component is validated ATOMICALLY
4067 /// (no-follow open), so a symlink final is refused in one
4068 /// syscall — no `symlink_metadata`→`canonicalize` TOCTOU. The
4069 /// parent is still canonicalized, so parent-side aliasing
4070 /// collapses; final-component case aliasing is deliberately NOT
4071 /// folded (the filename is a fixed constant in every real call
4072 /// site — trading it away removes the race).
4073 #[cfg(unix)]
4074 #[test]
4075 fn final_component_symlink_is_refused_atomically() {
4076 let scratch = Scratch::new();
4077 let path = scratch.state_path();
4078 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
4079 drop(OrgRevocationStore::open_existing(&path).expect("regular final opens"));
4080
4081 // Swap the final component for a symlink to a real file: the
4082 // no-follow validation refuses it — canonicalize never gets
4083 // the chance to follow it and re-key the store.
4084 let real = scratch.0.join("real-state.json");
4085 std::fs::rename(&path, &real).expect("move real");
4086 std::os::unix::fs::symlink(&real, &path).expect("plant final symlink");
4087 assert!(
4088 normalize_backing_path(&path).is_err(),
4089 "a symlink final component must refuse atomically"
4090 );
4091 assert!(
4092 OrgRevocationStore::open_existing(&path).is_err(),
4093 "open must refuse a symlink final"
4094 );
4095 }
4096
4097 /// Review-9: lock inodes are held to the full regular-file
4098 /// policy — a planted FIFO is refused (and, thanks to
4099 /// `O_NONBLOCK`, cannot park the open forever waiting for a
4100 /// reader), it does not carry the lock.
4101 #[cfg(unix)]
4102 #[test]
4103 fn non_regular_lock_sidecar_is_refused() {
4104 let scratch = Scratch::new();
4105 let path = scratch.state_path();
4106 let mut lock_path = path.as_os_str().to_os_string();
4107 lock_path.push(".lock");
4108 let status = std::process::Command::new("mkfifo")
4109 .arg(&lock_path)
4110 .status()
4111 .expect("run mkfifo");
4112 assert!(status.success(), "mkfifo failed");
4113
4114 let err = OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh)
4115 .expect_err("FIFO lock must refuse");
4116 assert!(matches!(err, OrgRevocationError::Io { .. }), "got: {err}");
4117 }
4118
4119 /// R2-4 (#3): a `.lock` sidecar with more than one hard link is
4120 /// refused fail-closed. Hard-linking a sidecar's inode to a second
4121 /// name is the attack that would otherwise collapse two DISTINCT
4122 /// backing paths onto one [`BackingId`] (one core + one poison
4123 /// entry).
4124 ///
4125 /// Red-witness: dropping the link-count check in `lock_state_file` lets
4126 /// the reopen succeed. Runs on both Unix (`nlink`) and Windows
4127 /// (`GetFileInformationByHandle`'s `nNumberOfLinks`) — a hard-linked
4128 /// sidecar must be refused identically on each.
4129 #[cfg(any(unix, windows))]
4130 #[test]
4131 fn hard_linked_lock_sidecar_is_refused() {
4132 let scratch = Scratch::new();
4133 let path = scratch.state_path();
4134 // First open creates the sidecar (nlink == 1), then drops so its
4135 // advisory lock and core are released.
4136 drop(OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init"));
4137 let mut lock_path = path.as_os_str().to_os_string();
4138 lock_path.push(".lock");
4139 let lock_path = PathBuf::from(lock_path);
4140 let alias = scratch.0.join("alias.lock");
4141 std::fs::hard_link(&lock_path, &alias).expect("hard-link the sidecar");
4142 let err = OrgRevocationStore::open_existing(&path)
4143 .expect_err("a hard-linked sidecar must be refused");
4144 assert!(
4145 matches!(&err, OrgRevocationError::Io { reason, .. } if reason.contains("hard links")),
4146 "got: {err}",
4147 );
4148 }
4149
4150 /// R2-4 (#2): the file-identity fallback key carries the COMPLETE
4151 /// normalized path, never a 64-bit hash — so two distinct paths can
4152 /// never collide onto one core/poison entry (the pre-R2-4
4153 /// `DefaultHasher` fallback could, at the ~2^32 birthday bound).
4154 ///
4155 /// This pins the fallback's TYPE (a `PathBuf`, not a hashed `u64`);
4156 /// the fstat-failure branch that selects it cannot be provoked from a
4157 /// unit test.
4158 #[test]
4159 fn path_fallback_backing_id_retains_the_full_path() {
4160 let a = BackingId::Path(PathBuf::from("/x/alpha/revocation-state.json"));
4161 let a2 = BackingId::Path(PathBuf::from("/x/alpha/revocation-state.json"));
4162 let b = BackingId::Path(PathBuf::from("/x/beta/revocation-state.json"));
4163 assert_eq!(a, a2, "the same normalized path is the same identity");
4164 assert_ne!(a, b, "distinct paths never share a fallback identity");
4165 assert_ne!(
4166 BackingId::FileId {
4167 device: 0,
4168 inode: 0
4169 },
4170 BackingId::Path(PathBuf::new()),
4171 "file-identity and path-fallback are distinct identity spaces",
4172 );
4173 }
4174
4175 /// R2-4 (#4): a backing path whose `.lock` sidecar is REPLACED under
4176 /// a still-live core is refused loudly, rather than silently forking
4177 /// the path into a second independent core (two security views of one
4178 /// path).
4179 ///
4180 /// Red-witness: removing the binding check in `join_or_create_core`
4181 /// lets the second open create a fresh core for the new sidecar
4182 /// identity, so it succeeds instead of failing.
4183 #[cfg(unix)]
4184 #[test]
4185 fn recreated_sidecar_under_a_live_core_is_refused() {
4186 let scratch = Scratch::new();
4187 let path = scratch.state_path();
4188 // Keep the first store ALIVE: its core (and the path→identity
4189 // binding) survive the whole test.
4190 let live =
4191 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
4192 let mut lock_path = path.as_os_str().to_os_string();
4193 lock_path.push(".lock");
4194 let lock_path = PathBuf::from(lock_path);
4195 // Pin the original sidecar inode across the replace. The store's
4196 // interprocess lock is transient (acquired per transaction, then
4197 // dropped — `StoreCore` holds no sidecar fd), so nothing keeps the
4198 // original inode allocated on its own. On an inode-recycling
4199 // filesystem (tmpfs, as under Kyra's Linux `/tmp`) an unlinked inode
4200 // with no open fd is reused immediately, so the recreation below
4201 // would collide back onto the SAME `(dev, inode)` and there would be
4202 // no replacement to detect. An explicit held fd guarantees the
4203 // recreated sidecar gets a DISTINCT inode on every filesystem.
4204 let pin = std::fs::File::open(&lock_path).expect("pin the original sidecar inode");
4205 // Replace the sidecar: unlink its directory entry (the original inode
4206 // persists under `pin`) and let the next open create a FRESH inode
4207 // under the same name.
4208 std::fs::remove_file(&lock_path).expect("unlink the old sidecar");
4209 let err = OrgRevocationStore::open_existing(&path)
4210 .expect_err("a recreated sidecar under a live core must be refused");
4211 assert!(
4212 matches!(err, OrgRevocationError::BackingIdentityConflict { .. }),
4213 "got: {err}",
4214 );
4215 drop(pin);
4216 drop(live);
4217 }
4218
4219 /// Review-8 §9 plumbing: the raise callback fires with exactly
4220 /// the raised floors, and never for a no-op (lower) bundle.
4221 #[test]
4222 fn raise_callback_fires_only_on_raises() {
4223 let scratch = Scratch::new();
4224 let store =
4225 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
4226 .expect("init");
4227
4228 let seen: Arc<Mutex<Vec<RaisedFloor>>> = Arc::new(Mutex::new(Vec::new()));
4229 let sink = seen.clone();
4230 let _sub = store.subscribe_floors_raised(move |raised| {
4231 sink.lock().extend_from_slice(raised);
4232 });
4233
4234 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
4235 assert_eq!(*seen.lock(), vec![(org().org_id(), member(), 5)]);
4236
4237 seen.lock().clear();
4238 store.apply_bundle(&bundle_with_floor(3)).expect("apply 3");
4239 assert!(seen.lock().is_empty(), "lower bundle must not notify");
4240 }
4241
4242 /// Review-8 §13 + review-9 witness: a POST-rename parent-fsync
4243 /// failure publishes the merged (never-weaker) view and poisons
4244 /// the BACKING PATH — every same-path instance refuses until an
4245 /// explicit recovery (locked reread + successful parent-dir
4246 /// fsync) proves the directory entry durable.
4247 #[cfg(unix)]
4248 #[test]
4249 fn post_rename_fsync_failure_poisons_the_path_until_recovery() {
4250 use std::os::unix::fs::PermissionsExt;
4251 let scratch = Scratch::new();
4252 let path = scratch.state_path();
4253 let store =
4254 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
4255 // A second instance on the SAME path, opened before the
4256 // failure — path-wide poison must gate it too (review-9).
4257 let sibling = OrgRevocationStore::open_existing(&path).expect("sibling");
4258 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
4259
4260 // Write+execute but NO read on the parent: lookups, file
4261 // reads, temp creation, and the rename all still work, but
4262 // opening the directory for fsync needs read — the exact
4263 // post-rename failure.
4264 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o300))
4265 .expect("chmod 0300");
4266 let err = store
4267 .apply_bundle(&bundle_with_floor(9))
4268 .expect_err("dir fsync must fail");
4269 assert!(
4270 matches!(err, OrgRevocationError::DurabilityUncertain { .. }),
4271 "got: {err}"
4272 );
4273
4274 // Fail-closed in the never-weaker direction: the merged
4275 // floor IS enforced (the rename landed; the disk may hold
4276 // it), but no same-path instance may pretend disk and
4277 // memory are synchronized while recovery is impossible.
4278 assert_eq!(store.floor_for(&org().org_id(), &member()), 9);
4279 assert!(store.is_poisoned());
4280 assert!(
4281 sibling.is_poisoned(),
4282 "poison is path-wide, not per instance"
4283 );
4284 let err = store
4285 .apply_bundle(&bundle_with_floor(11))
4286 .expect_err("originating store refuses while recovery fails");
4287 assert!(matches!(err, OrgRevocationError::Poisoned { .. }));
4288 // The SIBLING's no-op-shaped lower apply must equally refuse
4289 // — this is the review-9 red (it previously returned Ok).
4290 let err = sibling
4291 .apply_bundle(&bundle_with_floor(3))
4292 .expect_err("sibling refuses while the path is uncertain");
4293 assert!(matches!(err, OrgRevocationError::Poisoned { .. }));
4294 // A NEWLY OPENED instance cannot launder the uncertainty
4295 // either: its open attempts recovery, which still fails.
4296 assert!(
4297 OrgRevocationStore::open_existing(&path).is_err(),
4298 "fresh open must not bypass path poison while recovery fails"
4299 );
4300
4301 // Once the environment is repaired, the next operation
4302 // performs explicit recovery (locked reread + successful
4303 // parent fsync) and clears the uncertainty.
4304 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o700))
4305 .expect("chmod back");
4306 let raised = sibling
4307 .apply_bundle(&bundle_with_floor(11))
4308 .expect("recovered apply succeeds");
4309 assert!(raised.contains(&(org().org_id(), member(), 11)));
4310 assert!(!store.is_poisoned(), "recovery clears the path-wide bit");
4311 let reopened = OrgRevocationStore::open_existing(&path).expect("reopen");
4312 assert_eq!(reopened.floor_for(&org().org_id(), &member()), 11);
4313 }
4314
4315 /// R3-2: durability poison SURVIVES dropping every handle and
4316 /// recreating the `.lock` sidecar. Poison keyed only on the live
4317 /// sidecar [`BackingId`] would be laundered — a recreated `.lock` is a
4318 /// fresh, unpoisoned inode — so the canonical PATH tombstone keeps the
4319 /// path poisoned until explicit recovery, exactly once.
4320 ///
4321 /// Red-witness: dropping the `by_path` arm of `is_poisoned` lets the
4322 /// recreated-sidecar reopen skip recovery, so the "recovery still
4323 /// mandatory" `is_err()` assertion fails.
4324 #[cfg(unix)]
4325 #[test]
4326 fn poison_survives_dead_core_sidecar_recreation() {
4327 use std::os::unix::fs::PermissionsExt;
4328 let scratch = Scratch::new();
4329 let path = scratch.state_path();
4330 let mut lock_path = path.as_os_str().to_os_string();
4331 lock_path.push(".lock");
4332 let lock_path = PathBuf::from(lock_path);
4333
4334 // 1. Create + poison via a real post-rename parent-fsync failure.
4335 let store =
4336 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
4337 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
4338 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o300))
4339 .expect("chmod 0300");
4340 let err = store
4341 .apply_bundle(&bundle_with_floor(9))
4342 .expect_err("dir fsync must fail");
4343 assert!(
4344 matches!(err, OrgRevocationError::DurabilityUncertain { .. }),
4345 "got: {err}"
4346 );
4347 assert!(store.is_poisoned());
4348
4349 // 2. Drop every handle: the core dies and its path binding is
4350 // GC'd, so ONLY the poison registry remembers the uncertainty.
4351 drop(store);
4352
4353 // 3. Replace the `.lock` sidecar with a fresh inode (new BackingId).
4354 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o700))
4355 .expect("chmod 0700");
4356 std::fs::remove_file(&lock_path).expect("unlink old sidecar");
4357
4358 // 4. Reopen with recovery still BLOCKED — the path poison survived
4359 // the sidecar swap, so the fsync-refused reopen is refused.
4360 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o300))
4361 .expect("chmod 0300 again");
4362 assert!(
4363 OrgRevocationStore::open_existing(&path).is_err(),
4364 "recovery must still be mandatory after sidecar recreation — poison survived",
4365 );
4366
4367 // 5. Repair: the reopen recovers and clears the poison exactly once.
4368 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o700))
4369 .expect("chmod back");
4370 let recovered = OrgRevocationStore::open_existing(&path).expect("recovered reopen");
4371 assert!(
4372 !recovered.is_poisoned(),
4373 "successful recovery clears poison"
4374 );
4375 assert_eq!(recovered.floor_for(&org().org_id(), &member()), 9);
4376 let reopened = OrgRevocationStore::open_existing(&path).expect("clean reopen");
4377 assert!(
4378 !reopened.is_poisoned(),
4379 "poison stays cleared (cleared exactly once)"
4380 );
4381 }
4382
4383 /// R3-2: the same survival holds when the recreated path is reopened
4384 /// through a DIFFERENTLY-CASED alias on a case-insensitive filesystem —
4385 /// the canonical tombstone collapses the alias through the actual
4386 /// filesystem identity, so it is caught even after the sidecar swap.
4387 #[cfg(unix)]
4388 #[test]
4389 fn poison_survives_sidecar_recreation_across_a_cased_alias() {
4390 use std::os::unix::fs::PermissionsExt;
4391 let scratch = Scratch::new();
4392 let lower = scratch.0.join("revocation-state.json");
4393 let upper = scratch.0.join("REVOCATION-STATE.JSON");
4394
4395 let store = OrgRevocationStore::init(&lower, ProvisioningExpectation::MayBeFresh)
4396 .expect("init lower");
4397 // Case-insensitivity probe — a no-op on a case-sensitive FS.
4398 if !upper.exists() {
4399 return;
4400 }
4401 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
4402 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o300))
4403 .expect("chmod 0300");
4404 let err = store
4405 .apply_bundle(&bundle_with_floor(9))
4406 .expect_err("dir fsync must fail");
4407 assert!(matches!(
4408 err,
4409 OrgRevocationError::DurabilityUncertain { .. }
4410 ));
4411 drop(store);
4412
4413 // Recreate the `.lock` sidecar (new inode).
4414 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o700))
4415 .expect("chmod 0700");
4416 let mut lock_path = lower.as_os_str().to_os_string();
4417 lock_path.push(".lock");
4418 std::fs::remove_file(PathBuf::from(lock_path)).expect("unlink sidecar");
4419
4420 // Reopen through the UPPER-cased alias with recovery still blocked:
4421 // the tombstone must survive BOTH the sidecar swap and the alias.
4422 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o300))
4423 .expect("chmod 0300 again");
4424 assert!(
4425 OrgRevocationStore::open_existing(&upper).is_err(),
4426 "poison must survive a sidecar swap AND a cased-alias reopen",
4427 );
4428 std::fs::set_permissions(&scratch.0, std::fs::Permissions::from_mode(0o700))
4429 .expect("chmod back");
4430 let recovered = OrgRevocationStore::open_existing(&upper).expect("recovered via alias");
4431 assert!(!recovered.is_poisoned());
4432 }
4433
4434 /// P2 hygiene: recovering a canonical path retires EVERY sidecar identity
4435 /// ever poisoned under it — not just the recovering one — so a stale old
4436 /// `BackingId` (left behind when a sidecar was unlinked and recreated)
4437 /// does not linger in `by_id` to trip redundant recovery after inode
4438 /// reuse. Directly exercises the poison registry (no filesystem poison
4439 /// needed), so it runs on every platform.
4440 ///
4441 /// Red-witness: reverting `clear_poison` to remove only the passed `id`
4442 /// leaves `old_id` poisoned, so the final `is_poisoned(&old_id, ..)` holds.
4443 #[test]
4444 fn clear_poison_retires_all_stale_ids_for_the_path() {
4445 let scratch = Scratch::new();
4446 let path = scratch.state_path();
4447 let old_id = BackingId::FileId {
4448 device: 0x5005,
4449 inode: 0xF00D_0001,
4450 };
4451 let new_id = BackingId::FileId {
4452 device: 0x5005,
4453 inode: 0xF00D_0002,
4454 };
4455 // Two sidecar identities poisoned under the SAME path — the
4456 // unlink+recreate that strands the old id in `by_id`.
4457 mark_poisoned(&old_id, &path);
4458 mark_poisoned(&new_id, &path);
4459 assert!(is_poisoned(&old_id, &path));
4460 assert!(is_poisoned(&new_id, &path));
4461
4462 // Recovery through the CURRENT (new) id clears the path tombstone AND
4463 // retires the stale old id in lockstep — no dead residue survives.
4464 clear_poison(&new_id, &path);
4465 assert!(!is_poisoned(&new_id, &path), "recovered id cleared");
4466 assert!(
4467 !is_poisoned(&old_id, &path),
4468 "stale old id retired with the path recovery — no dead residue",
4469 );
4470 }
4471
4472 /// Gate-1: the generic, path-agnostic store API must NOT chmod a supplied
4473 /// parent directory — only the dedicated authority scaffold
4474 /// (`org_authority::ensure_secure_authority_dir`) creates/tightens the
4475 /// owner-only authority root. Create a parent with a known loose mode,
4476 /// init a store under it, and assert the parent's mode is untouched (so
4477 /// a legitimate shared application directory is never mutated).
4478 #[cfg(unix)]
4479 #[test]
4480 fn generic_store_init_does_not_chmod_the_parent() {
4481 use std::os::unix::fs::PermissionsExt;
4482 let scratch = Scratch::new();
4483 let parent = scratch.0.join("shared-app-dir");
4484 std::fs::create_dir_all(&parent).expect("mkdir parent");
4485 std::fs::set_permissions(&parent, std::fs::Permissions::from_mode(0o755))
4486 .expect("chmod 0755");
4487 let store = OrgRevocationStore::init(
4488 parent.join("revocation-state.json"),
4489 ProvisioningExpectation::MayBeFresh,
4490 )
4491 .expect("init");
4492 drop(store);
4493 let mode = std::fs::metadata(&parent)
4494 .expect("metadata")
4495 .permissions()
4496 .mode();
4497 assert_eq!(
4498 mode & 0o777,
4499 0o755,
4500 "generic store init must not chmod the parent (mode {mode:o})",
4501 );
4502 }
4503
4504 /// R3-3: an existing live handle's `apply_bundle` verifies the opened
4505 /// `.lock` sidecar identity against its core's BEFORE reread/merge/
4506 /// write. If the sidecar was replaced under the live handle (fresh
4507 /// inode — which the `nlink` refusal does not catch), the transaction
4508 /// is refused loudly with `BackingIdentityConflict` and neither the
4509 /// live view nor the disk floors advance.
4510 ///
4511 /// Red-witness: dropping the `opened_id != core.backing_id` check lets
4512 /// the existing handle lock and publish through the replaced sidecar,
4513 /// so `apply_bundle` returns `Ok` and the floor advances to 9.
4514 #[cfg(unix)]
4515 #[test]
4516 fn existing_handle_refuses_a_replaced_sidecar() {
4517 let scratch = Scratch::new();
4518 let path = scratch.state_path();
4519 let store =
4520 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
4521 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
4522
4523 // Replace the `.lock` sidecar under the LIVE store. The store's
4524 // interprocess lock is transient (per transaction — `StoreCore`
4525 // holds no sidecar fd), so the original inode is NOT kept allocated
4526 // on its own. Pin it with an explicit fd so the recreation is
4527 // guaranteed a DISTINCT inode even on an inode-recycling filesystem
4528 // (tmpfs, as under Kyra's Linux `/tmp`, reuses an unlinked inode with
4529 // no open fd immediately — which would collide the recreated sidecar
4530 // back onto the original identity, leaving nothing to detect).
4531 let mut lock_path = path.as_os_str().to_os_string();
4532 lock_path.push(".lock");
4533 let lock_path = PathBuf::from(lock_path);
4534 let pin = std::fs::File::open(&lock_path).expect("pin the original sidecar inode");
4535 // Unlink the entry (the old inode persists via `pin`); the next lock
4536 // open recreates it with a fresh inode.
4537 std::fs::remove_file(&lock_path).expect("unlink sidecar");
4538
4539 let err = store
4540 .apply_bundle(&bundle_with_floor(9))
4541 .expect_err("existing handle must refuse a replaced sidecar");
4542 assert!(
4543 matches!(err, OrgRevocationError::BackingIdentityConflict { .. }),
4544 "got: {err}"
4545 );
4546 // The live view never advanced past 5.
4547 assert_eq!(store.floor_for(&org().org_id(), &member()), 5);
4548
4549 drop(pin);
4550 // Nor did the disk: a fresh handle (after the live core drops so
4551 // its stale path binding is released) reads 5, never 9.
4552 drop(store);
4553 let reopened = OrgRevocationStore::open_existing(&path).expect("reopen after drop");
4554 assert_eq!(
4555 reopened.floor_for(&org().org_id(), &member()),
4556 5,
4557 "the refused transaction must not have written floor 9 to disk",
4558 );
4559 }
4560
4561 /// Review-9: raise callbacks run OUTSIDE both the file lock and
4562 /// the instance reload guard — a callback that synchronously
4563 /// re-enters `apply_bundle` on the same store must not
4564 /// deadlock.
4565 #[test]
4566 fn reentrant_callback_does_not_deadlock() {
4567 let scratch = Scratch::new();
4568 let store = Arc::new(
4569 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
4570 .expect("init"),
4571 );
4572
4573 let reentered = Arc::new(Mutex::new(false));
4574 let store_for_callback = Arc::downgrade(&store);
4575 let flag = reentered.clone();
4576 let _sub = store.subscribe_floors_raised(move |raised| {
4577 // Re-enter once, from the first raise only.
4578 if raised.iter().any(|(_, _, floor)| *floor == 5) {
4579 if let Some(store) = store_for_callback.upgrade() {
4580 store
4581 .apply_bundle(&bundle_with_floor(7))
4582 .expect("re-entrant apply must not deadlock");
4583 *flag.lock() = true;
4584 }
4585 }
4586 });
4587
4588 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
4589 assert!(*reentered.lock(), "callback re-entered apply_bundle");
4590 assert_eq!(store.floor_for(&org().org_id(), &member()), 7);
4591 }
4592
4593 /// Review-9 filesystem policy: state files and the lock sidecar
4594 /// are opened no-follow — a planted symlink is refused, not
4595 /// followed.
4596 #[cfg(unix)]
4597 #[test]
4598 fn symlinked_state_and_lock_files_are_refused() {
4599 let scratch = Scratch::new();
4600 let path = scratch.state_path();
4601 let store =
4602 OrgRevocationStore::init(&path, ProvisioningExpectation::MayBeFresh).expect("init");
4603 store.apply_bundle(&bundle_with_floor(5)).expect("apply 5");
4604 drop(store);
4605
4606 // Symlinked STATE file: reads refuse.
4607 let real = scratch.0.join("elsewhere.json");
4608 std::fs::rename(&path, &real).expect("move state");
4609 std::os::unix::fs::symlink(&real, &path).expect("plant symlink");
4610 assert!(
4611 OrgRevocationStore::open_existing(&path).is_err(),
4612 "symlinked state file must refuse"
4613 );
4614 std::fs::remove_file(&path).expect("remove link");
4615 std::fs::rename(&real, &path).expect("restore state");
4616 OrgRevocationStore::open_existing(&path).expect("regular file opens");
4617
4618 // Symlinked LOCK sidecar: locking refuses rather than
4619 // following the link to a foreign inode — at open time
4620 // (every open serializes behind the lock, review-9
4621 // addendum) and on a reload through a previously-opened
4622 // handle alike.
4623 let store = OrgRevocationStore::open_existing(&path).expect("open before planting");
4624 let mut lock_path = path.as_os_str().to_os_string();
4625 lock_path.push(".lock");
4626 let lock_path = PathBuf::from(lock_path);
4627 let _ = std::fs::remove_file(&lock_path);
4628 let foreign = scratch.0.join("foreign.lock");
4629 std::fs::write(&foreign, b"").expect("foreign lock");
4630 std::os::unix::fs::symlink(&foreign, &lock_path).expect("plant lock symlink");
4631 assert!(
4632 OrgRevocationStore::open_existing(&path).is_err(),
4633 "symlinked lock sidecar must refuse the open"
4634 );
4635 assert!(
4636 store.apply_bundle(&bundle_with_floor(9)).is_err(),
4637 "symlinked lock sidecar must refuse a reload"
4638 );
4639 }
4640
4641 /// OA2-E1 (Kyra review) — the publication barrier. A barriered
4642 /// generation read issued while a floor publish is paused between
4643 /// the live-view swap and the generation bump must NOT observe the
4644 /// stale (pre-bump) generation the bare `publish_generation()`
4645 /// still returns; it blocks on `live.read()` and, once released,
4646 /// returns the NEW generation. Deterministic: the publisher is
4647 /// pinned in the exact "new view installed, old generation
4648 /// present" window by the one-shot pause hook.
4649 #[test]
4650 fn barriered_generation_never_observes_an_in_progress_publish() {
4651 use std::sync::mpsc;
4652
4653 let scratch = Scratch::new();
4654 let store = Arc::new(
4655 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
4656 .expect("init"),
4657 );
4658 let g0 = store.barriered_generation().expect("not exhausted").get();
4659
4660 // Arm the one-shot pause, then raise a floor on another thread:
4661 // it swaps the live view and blocks BEFORE bumping the
4662 // generation, holding `live.write()` throughout.
4663 let (swapped_rx, resume_tx) = store.arm_publish_pause_for_test();
4664 let publisher = {
4665 let store = store.clone();
4666 std::thread::spawn(move || {
4667 store.apply_bundle(&bundle_with_floor(9)).expect("apply");
4668 })
4669 };
4670 swapped_rx.recv().expect("publisher reached the pause");
4671
4672 // Window open: new view installed, generation NOT yet bumped,
4673 // write lock held. A BARE read sees the stale generation — the
4674 // hazard the barrier closes.
4675 assert_eq!(
4676 store.publish_generation(),
4677 g0,
4678 "bare read observes the pre-bump generation while the new floor is already swapped in",
4679 );
4680
4681 // A BARRIERED read issued now must block on `live.read()` — no
4682 // result until the publisher releases the write lock.
4683 let (reader_tx, reader_rx) = mpsc::channel();
4684 let reader = {
4685 let store = store.clone();
4686 std::thread::spawn(move || {
4687 let g = store.barriered_generation().expect("not exhausted");
4688 let _ = reader_tx.send(g.get());
4689 })
4690 };
4691 std::thread::sleep(std::time::Duration::from_millis(50));
4692 assert!(
4693 reader_rx.try_recv().is_err(),
4694 "barriered read must block while the publish holds live.write() mid-swap",
4695 );
4696
4697 // Release: the publisher bumps the generation and drops the
4698 // write lock; the barriered reader unblocks.
4699 resume_tx.send(()).expect("resume");
4700 publisher.join().expect("publisher join");
4701 reader.join().expect("reader join");
4702
4703 let observed = reader_rx.recv().expect("barriered read result");
4704 assert_eq!(
4705 observed,
4706 g0 + 1,
4707 "the barriered read returned the NEW generation, never the stale one",
4708 );
4709 assert_eq!(
4710 store.barriered_generation().expect("not exhausted").get(),
4711 g0 + 1
4712 );
4713 assert!(store.floor_for(&org().org_id(), &member()) >= 9);
4714 }
4715
4716 /// The publication generation NEVER wraps: at the ceiling it freezes and
4717 /// latches.
4718 ///
4719 /// Wrapping is not a bounded-counter inconvenience, it is an aliasing bug: a
4720 /// consumer using the generation as a currentness discriminator would accept
4721 /// evidence built against the OLD view as current against the NEW one (Kyra
4722 /// OLB-2B-E3c).
4723 #[test]
4724 fn an_exhausted_publication_generation_freezes_rather_than_wrapping() {
4725 let scratch = Scratch::new();
4726 let store =
4727 OrgRevocationStore::init(scratch.state_path(), ProvisioningExpectation::MayBeFresh)
4728 .expect("init");
4729
4730 assert!(store.barriered_generation().is_ok());
4731 store.saturate_generation_for_test();
4732 assert_eq!(
4733 store
4734 .barriered_generation()
4735 .expect("not yet exhausted")
4736 .get(),
4737 u64::MAX
4738 );
4739
4740 store.republish_for_test();
4741
4742 assert_eq!(
4743 store.barriered_generation(),
4744 Err(GenerationExhausted),
4745 "the exhausted space must be reported as an ERROR the caller cannot ignore, not as a frozen integer that reads as unchanged"
4746 );
4747 assert_eq!(
4748 store.snapshot_with_generation().err(),
4749 Some(GenerationExhausted),
4750 "the coherent snapshot sampler fails closed the same way"
4751 );
4752 assert!(store.generation_exhausted_for_metrics());
4753
4754 // Terminal: a further publication does not clear it.
4755 store.republish_for_test();
4756 assert_eq!(store.barriered_generation(), Err(GenerationExhausted));
4757 }
4758}