diff --git a/CHANGELOG.md b/CHANGELOG.md index a4ae3e4..8f303c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,17 @@ Deprecated, Fixed or Security. ### Added +- **Provisional:** runtime peer records support immutable `expires_at` deadlines + within the next 600 seconds. Existing atomic batches can install a prepared + next NodeID's exact grants and schedule the old owner record's retirement. + Timed policy generations close expired access independently of blocked disk + writers, preserving static/other-owner grants; automatic removal reuses + admission/session cancellation, connection retirement and durable completion + receipts. Expiry reserves conflict-checked receipt IDs and capacity, fences + stale sweep generations, survives restart, and uses a re-anchored progressing + monotonic floor across forward wall steps and rollback. Runtime state that + has used expiry cannot be + downgraded to strict readers lacking these fields. - **Provisional:** opt-in pairing permit issue/query/cancel control requests, with private durable state, immutable exact policy with a versioned digest, bounded lifetime and reserved terminal receipt capacity (1,024 globally, diff --git a/docs/runtime-peer-api.md b/docs/runtime-peer-api.md index c56d562..8f82fde 100644 --- a/docs/runtime-peer-api.md +++ b/docs/runtime-peer-api.md @@ -269,11 +269,14 @@ The `operation` field selects one of: | `list` | Optional `cursor` from the preceding page. | | `add_peer` | `owner`, `request_id`, `expected_revision`, `node_id`; optional `name`, `direction`, `addr`. | | `set_grants` | `owner`, `request_id`, `expected_revision`, `node_id`, `grants`. | +| `set_expiry` | `owner`, `request_id`, `expected_revision`, `node_id`, `expires_at`. | | `remove_peer` | `owner`, `request_id`, `expected_revision`, `node_id`. | | `batch` | `owner`, `request_id`, `expected_revision`, `operations`. | | `cancellation` | `request_id` of a retained mutation receipt. | An owner and request ID each contain 1–128 bytes without control characters. +The `expiry:` request-ID prefix is system-reserved: owner mutations cannot use +it, including before a deadline is scheduled or after its receipt is retained. A new record starts with no grants and normal direction. An omitted name, direction, or address preserves that owner's existing value. `set_grants` replaces the entire owner's grant set; `[]` withdraws all services while @@ -312,6 +315,69 @@ an intermediate identity or unintended dual access: Static and other-owner records on the old NodeID remain visible and effective; the caller must inspect the result before claiming complete retirement. +### Bounded overlap expiry + +To publish a prepared replacement over the still-working old identity, use one +batch with `add_peer` for the next NodeID, `set_grants` for its **exact** grants, +and `set_expiry` for the old owner's record. For example, replace the immediate +removal in the example above with: + +```json +{"op":"set_expiry","node_id":"","expires_at":1792887000} +``` + +The integer is illustrative: use an absolute UTC Unix-seconds deadline strictly +after the daemon's current observation and no more than 600 seconds ahead. +The signed 64-bit wire/storage representation rejects overflow; zero, negative, +past and over-limit deadlines return `invalid_argument` without committing any +part of the batch. The old and next NodeIDs must use distinct labels during +overlap. This grants overlap on the receiving peer only, not dual-identity +serving by the rotating daemon. + +The owner's record starts with no expiry. Once set, `expires_at` cannot be +cleared, shortened or extended, including by updating grants, re-adding the +same existing record, or replaying an old request. An identical setter is a +no-op. Explicit removal and re-addition creates a new record; it is not an +implicit renewal. Static and other-owner access is never modified. + +Runtime entries in `trust_sources`, `direction_sources`, and service provenance +include `expires_at` and `expiry_request_id` when set. Save `expiry_request_id` +from the scheduling reply/list before expiry; query it with `cancellation` to +observe automatic removal, residual access, selected work and completion. The +reserved ID belongs to that scheduling transaction, so a separately re-created +record cannot replay an earlier expiry receipt. Until cleanup commits, its +cancellation query returns an unknown receipt, never a premature completion. +Reservation conflicts with retained receipt IDs return `request_id_conflict` +before any part of the scheduling batch commits. + +Authorization and attach select a prevalidated effective generation at the +deadline, without disk I/O or waiting for the mutation sequencer. A separate +50 ms deadline sweep uses the same admission/session cancellation and connection +retirement paths as removal. The serialized 100 ms cleanup sweep writes the +owner-scoped removal and advances the policy revision; idle ticks do not fsync. +Expiry withdrawals are not blocked by stale sync selectors. A blocked writer +cannot extend expired authorization or live work; residual grants preserve +still-permitted work. Cancellation monitoring retains its two-second deadline +and missed-bound evidence even when the automatic receipt is persisted later. +Each automatic receipt names only cancelled work for its own expiring NodeID; +other records expiring in the same sweep retain their own cancellation evidence. +Sweep selection and signaling share an in-memory publication-generation guard; +they cannot revoke against a stale snapshot after newer residual grants publish. +This guard never covers filesystem work. + +Running deadline observations combine wall and monotonic time and never move +backward, including reload reconciliation. Each forward wall observation anchors +a progressing monotonic floor, so rollback after scheduling cannot extend the +600-second elapsed bound. Restart removes elapsed records durably before +accepting traffic. A clock below the persisted last expiry +observation conservatively expires outstanding records. Arbitrary rollback +across power loss is not assumed detectable. Cleanup write failures cannot +restore expired access; uncertain commits use ordinary policy quarantine. + +Homes without expiry keep the old representation. Once expiry is used, its +record fields and retained clock anchor make runtime state incompatible with +older strict readers lacking this feature. + ## Revisions, persistence, and lost replies Shutdown allows up to three seconds for in-flight policy/config writes to drain, @@ -354,9 +420,11 @@ request; do not invent a new request ID for an uncertain outcome. The store permits at most 1,024 owner/NodeID records in total. Receipts are retained for at least 24 hours in a pool of 1,024 slots. Ordinary mutations may use 960 slots; the remaining 64 are reserved for batches containing only -`remove_peer` or `set_grants` that remove services without adding any. Pure -no-ops consume no slot, even when the pool is full. Expired receipts are pruned -by a subsequent changed mutation. When the applicable capacity is full, that +`remove_peer` or `set_grants` that remove services without adding any. +Scheduling with `set_expiry` uses ordinary capacity and also reserves a future automatic receipt slot; all +mutations account for those reservations before committing. Pure no-ops consume +no slot, even when the pool is full. Expired receipts are pruned by a subsequent +changed mutation. When the applicable capacity is full, that mutation is refused before commit. After retention, reconcile current state and revision before deciding on a new operation. diff --git a/docs/runtime-peers.md b/docs/runtime-peers.md index 6581f6a..c397a1a 100644 --- a/docs/runtime-peers.md +++ b/docs/runtime-peers.md @@ -2,9 +2,9 @@ Status: approved design. Runtime mode, direct peer/grant transactions, revisions, and revocation receipts are implemented as a provisional API; see the -[runtime peer API](runtime-peer-api.md). Rotation expiry, policy watches, -pairing and the typed client are follow-up slices whose request names and -limits below remain proposals. The surface remains provisional through at +[runtime peer API](runtime-peer-api.md). Immutable rotation expiry and pairing +are implemented; policy watches and the typed client remain follow-up slices. +The surface remains provisional through at least one minor release of real use, with an explicit review before promotion. This note covers [#312](https://github.com/compoundingtech/fabric/issues/312) and its typed control client, [#310](https://github.com/compoundingtech/fabric/issues/310). @@ -112,7 +112,7 @@ static or other-owner access to that NodeID is reported by the same rule. ## Transactions, replacement, and expiry -The proposed typed operations are `add_peer`, `set_grants`, `remove_peer`, +The typed operations are `add_peer`, `set_grants`, `set_expiry`, `remove_peer`, `list`, and `batch`. `set_grants` replaces that owner's entire set; `[]` withdraws it. `add_peer` for a new ID starts with zero grants. Renaming an existing ID preserves its grants. A batch has one owner, a fresh request ID, @@ -163,28 +163,55 @@ reinstalling access; the client must reconcile instead of inventing a new ID for an uncertain request. No-op mutations return the current revision without storing a receipt. -Rotation defaults to immediate replacement. Explicit overlap uses a batch that -installs the new record and sets an immutable expiry on the old owner's record, -with a maximum overlap of 10 minutes. Overlap cannot silently extend on retry; -a further change requires a new explicit transaction. The old ID retains a -distinct label during overlap. Admission and session attach enforce the expiry -even if the cleanup timer or disk writer is late. The timer removes the expired -record and advances policy revision. Admissions and sessions that rely on an -expiring record also carry its deadline: they recheck the remaining effective -grants and signal cancellation independently of the persistence queue. A stalled -cleanup writer cannot extend live work that lost its last grant. The same -2-second completion target below is measured from the expiry deadline for -automatic withdrawal; a missed bound is reported. Persistence failure cannot -keep expired access live. Restart filters expired records before -admitting traffic and durably records their removal before mutation success. - -Expiry stores an absolute UTC deadline and uses a monotonic deadline while -running, taking the earlier limit so a backward wall-clock step cannot extend -access. Recovery needs a trustworthy clock; detected rollback or uncertain -time closes expiring access and unconfirmed permits instead of renewing them. -This bound does not assume arbitrary clock rollback can be measured across a -power loss. Static or other-owner grants have their own provenance and are -never removed by an overlap expiry. +Rotation defaults to immediate replacement. Explicit overlap uses the existing +atomic batch to add the new NodeID, set its exact grants, and `set_expiry` on +the old owner's record. This changes peer policy only: the rotating machine +prepares its key, publishes that batch over its still-working old identity, and +then activates its new identity. Fabric never needs to serve both identities. +The old ID retains a distinct label during overlap. + +`expires_at` is an absolute UTC Unix-seconds signed 64-bit integer, strictly +in the future and at most 600 seconds away when first set. Negative, zero, +out-of-range and later deadlines are refused atomically. The deadline is +immutable for that owner/NodeID record, including through `add_peer` and +`set_grants`; an identical setter is a no-op and a historical batch retry +cannot renew or reinstall access. Removing and explicitly re-adding a record +creates a new record. Runtime source provenance reports `expires_at` and the +reserved `expiry_request_id` used to query automatic cancellation. + +Policy generations precompute the effective union at each distinct deadline. +Their shared wall/monotonic observation never moves backward during one daemon +lifetime, including reconciliation. Every forward observation re-anchors a +progressing monotonic floor: a later wall rollback cannot freeze the clock or +extend a newly installed overlap beyond 600 elapsed seconds. +Admission, service authorization, outbound +policy lookup and session attach use the applicable generation without waiting +for a timer, mutation sequencer or disk writer. The deadline sweep independently +signals the existing admission/session revocation path, including connection +retirement, while the serialized writer records owner-scoped removal and +advances policy revision. Still-permitted static and other-owner work survives. +Selection and signaling hold a separate in-memory generation read guard; +publication fences it only after durability, never while waiting on disk. +A sweep cannot cancel work against a superseded policy lacking newer grants. +The two-second cancellation target is measured from expiry, not a delayed +cleanup commit; timeout evidence survives through the automatic receipt. + +Setting expiry reserves an automatic receipt slot in the existing bounded pool. +Generated receipt reservations are checked against retained receipts and the +final candidate before committing any scheduling operation; collision refuses +the complete batch rather than poisoning automatic cleanup or restart. +An idle sweep performs no disk writes. Failed cleanup persistence cannot +reopen expired access; indeterminate outcomes use ordinary quarantine and reload +reconciliation. Cleanup withdrawals cannot be blocked by stale sync selectors. +Restart filters and durably removes elapsed records before admitting traffic, +with complete empty cancellation receipts because no work from the new daemon +was admitted. The persisted expiry clock detects rollback below its last +observation and conservatively expires outstanding records. This does not claim +that arbitrary clock rollback across a power loss can be measured. + +Expiry fields and the clock anchor are omitted until expiry is used. Existing +state without expiry remains readable by older strict readers; state that has +used expiry is not downgrade-compatible with releases lacking this slice. ## Revision and watch contract diff --git a/src/admission.rs b/src/admission.rs index 151a784..11296f3 100644 --- a/src/admission.rs +++ b/src/admission.rs @@ -120,6 +120,7 @@ impl Admissions { }) .map(|(id, entry)| Revocation { id: *id, + peer: entry.peer, revoked: entry.revoked.clone(), done: entry.done.clone(), children: entry.children.clone(), @@ -178,6 +179,7 @@ impl Requirements { #[derive(Debug, Clone)] pub(crate) struct Revocation { pub id: u64, + pub peer: EndpointId, revoked: CancellationToken, done: CancellationToken, children: fabric_service_api::child_tracking::Children, diff --git a/src/config.rs b/src/config.rs index c9bfa6e..0b00ea6 100644 --- a/src/config.rs +++ b/src/config.rs @@ -276,6 +276,58 @@ const ROLLBACK_MIRROR_COMMENT: &str = "\ pub struct PeerBook { peers: Vec, git_remotes: Vec, + expiry: Option>, +} + +#[derive(Debug)] +struct TimedPeers { + clock: std::sync::Arc, + generations: Vec<(u64, Vec)>, +} + +/// A forward observation establishes a new progressing monotonic floor, not a +/// frozen high-water mark. Shared by authorization, cleanup and reconciliation. +#[derive(Debug)] +pub(crate) struct ExpiryClock { + anchor: std::sync::Mutex<(std::time::Duration, std::time::Instant)>, +} + +impl ExpiryClock { + pub(crate) fn new(anchor: (u64, std::time::Instant)) -> Self { + Self { + anchor: std::sync::Mutex::new((std::time::Duration::from_secs(anchor.0), anchor.1)), + } + } + + fn observe_at(&self, wall: std::time::Duration, instant: std::time::Instant) -> std::time::Duration { + let mut anchor = self.anchor.lock().unwrap_or_else(|poison| poison.into_inner()); + let floor = anchor.0.saturating_add(instant.saturating_duration_since(anchor.1)); + if wall > floor { + *anchor = (wall, instant); + wall + } else { + floor + } + } + + fn observed(&self, floor: u64) -> std::time::Duration { + let wall = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .max(std::time::Duration::from_secs(floor)); + self.observe_at(wall, std::time::Instant::now()) + } + + pub(crate) fn timestamp(&self, floor: u64) -> u64 { + self.observed(floor).as_secs() + } + + #[cfg(test)] + pub(crate) fn elapse_for_test(&self, elapsed: std::time::Duration) { + let timestamp = self.observed(0).saturating_add(elapsed); + *self.anchor.lock().unwrap_or_else(|poison| poison.into_inner()) = + (timestamp, std::time::Instant::now()); + } } #[derive(Debug, Default, Clone, Deserialize)] @@ -317,6 +369,7 @@ impl Default for PeerBook { Self { peers: Vec::new(), git_remotes: Vec::new(), + expiry: None, } } } @@ -358,6 +411,7 @@ impl PeerBook { let book = Self { peers, git_remotes: self.git_remotes.clone(), + expiry: None, }; book.validate()?; Ok(book) @@ -465,6 +519,7 @@ impl PeerBook { Self { peers: stored.peers, git_remotes: stored.git_remotes, + expiry: None, }, migrated, removed_grants, @@ -571,6 +626,7 @@ impl PeerBook { let before_book = Self { peers: before.peers.clone(), git_remotes: before.git_remotes.clone(), + expiry: None, }; before_book.validate()?; let mut document: DocumentMut = existing @@ -632,12 +688,61 @@ impl PeerBook { config.save(home) } + pub(crate) fn into_effective_peers(self) -> Vec { + self.peers + } + + /// Prevalidated immutable generations keep expiry checks allocation-free + /// and independent of the policy writer and filesystem. + pub(crate) fn with_expiry_generations( + mut self, + clock: std::sync::Arc, + generations: Vec<(u64, Vec)>, + ) -> Self { + if !generations.is_empty() { + self.expiry = Some(std::sync::Arc::new(TimedPeers { + clock, + generations, + })); + } + self + } + + pub(crate) fn expiry_lateness(&self, deadline: u64) -> std::time::Duration { + self.expiry.as_ref().map_or(std::time::Duration::ZERO, |expiry| { + expiry.clock.observed(0).saturating_sub(std::time::Duration::from_secs(deadline)) + }) + } + + fn expiry_index(&self) -> usize { + let Some(expiry) = &self.expiry else { return 0; }; + let timestamp = expiry.clock.timestamp(0); + expiry.generations.partition_point(|(deadline, _)| *deadline <= timestamp) + } + + pub(crate) fn expired_generation(&self) -> Option { + let index = self.expiry_index(); + (index > 0).then(|| self.expiry.as_ref().unwrap().generations[index - 1].0) + } + + pub(crate) fn same_expiry_policy(&self, other: &Self) -> bool { + match (&self.expiry, &other.expiry) { + (Some(left), Some(right)) => std::sync::Arc::ptr_eq(left, right), + (None, None) => true, + _ => false, + } + } + pub fn peers(&self) -> &[Peer] { + let index = self.expiry_index(); + if index > 0 { + return &self.expiry.as_ref().unwrap().generations[index - 1].1; + } &self.peers } pub fn grants_service(&self, service: &str) -> bool { - self.peers + self.peers() .iter() .any(|peer| peer.allow.iter().any(|grant| grant == service)) } @@ -746,7 +851,7 @@ impl PeerBook { /// /// This peer grant is the complete policy for the service. pub fn may(&self, id: &EndpointId, service: &str) -> Result<(), Denied> { - let Some(peer) = self.peers.iter().find(|peer| peer.id == *id) else { + let Some(peer) = self.peers().iter().find(|peer| peer.id == *id) else { return Err(Denied::NotTrusted); }; // Before the allow list is read, so nothing written there matters. @@ -769,7 +874,7 @@ impl PeerBook { /// The peers let in as inbound connections: every peer but a dial-only one, /// which this machine dials and which never gets to dial it. pub fn trusted_ids(&self) -> HashSet { - self.peers + self.peers() .iter() .filter(|peer| !peer.dial_only) .map(|peer| peer.id) @@ -778,14 +883,14 @@ impl PeerBook { /// Is `id` a peer that may connect to this machine at all? pub fn accepts_inbound(&self, id: &EndpointId) -> bool { - self.peers + self.peers() .iter() .any(|peer| peer.id == *id && !peer.dial_only) } /// Is `id` a peer this machine dials and never lets in? pub fn is_dial_only(&self, id: &EndpointId) -> bool { - self.peers + self.peers() .iter() .any(|peer| peer.id == *id && peer.dial_only) } @@ -793,7 +898,7 @@ impl PeerBook { /// Does this peer list grants that a dial-only flag overrides? A file in /// that state is safe and contradicts itself; `fabric doctor` says so. pub fn ignored_grants(&self, id: &EndpointId) -> bool { - self.peers + self.peers() .iter() .any(|peer| peer.id == *id && peer.dial_only && !peer.allow.is_empty()) } @@ -846,7 +951,7 @@ impl PeerBook { /// Is `id` a peer that only connects here, and that this machine never dials? pub fn is_inbound_only(&self, id: &EndpointId) -> bool { - self.peers + self.peers() .iter() .any(|peer| peer.id == *id && peer.inbound_only) } @@ -854,7 +959,7 @@ impl PeerBook { /// The peers this machine may dial, probe and announce to: every peer but /// an inbound-only one. pub fn dialable_peers(&self) -> impl Iterator { - self.peers.iter().filter(|peer| !peer.inbound_only) + self.peers().iter().filter(|peer| !peer.inbound_only) } /// Refuse to mix the two directions. `peer` is a name or a NodeID; an @@ -962,7 +1067,7 @@ impl PeerBook { } let matches: Vec<&Peer> = self - .peers + .peers() .iter() .filter(|entry| entry.name.as_deref() == Some(peer)) .collect(); @@ -999,7 +1104,7 @@ impl PeerBook { } fn addr_for_id(&self, id: EndpointId) -> EndpointAddr { - self.peers + self.peers() .iter() .find(|entry| entry.id == id) .and_then(|entry| entry.addr.clone()) @@ -1709,6 +1814,7 @@ impl FabricConfig { PeerBook { peers: self.peers.clone(), git_remotes: Vec::new(), + expiry: None, } .validate()?; @@ -1873,6 +1979,25 @@ fn invalid_argument(message: String) -> anyhow::Error { #[cfg(test)] mod tests { + #[test] + fn expiry_clock_reanchors_forward_steps_before_wall_rollback() { + use std::time::{Duration, Instant}; + let origin = Instant::now(); + let clock = super::ExpiryClock::new((1_000, origin)); + assert_eq!( + clock.observe_at(Duration::from_secs(4_610), origin + Duration::from_secs(10)), + Duration::from_secs(4_610), + ); + assert_eq!( + clock.observe_at(Duration::from_secs(1_610), origin + Duration::from_secs(610)), + Duration::from_secs(5_210), + ); + assert_eq!( + clock.observe_at(Duration::from_secs(1_611), origin + Duration::from_secs(611)), + Duration::from_secs(5_211), + ); + } + #[test] fn dial_queue_timeout_is_additive_and_rejects_invalid_deadlines() -> anyhow::Result<()> { let legacy: super::PersistedDial = toml::from_str("peer = 'example'\nprotocol = 'web'\n")?; diff --git a/src/control.rs b/src/control.rs index 7f8ff6e..c14155f 100644 --- a/src/control.rs +++ b/src/control.rs @@ -639,6 +639,8 @@ mod runtime_peer_schema_tests { }, Source::Runtime { owner: "controller-b".into(), + expires_at: None, + expiry_request_id: None, }, ]; let peer = EffectivePeer { diff --git a/src/daemon.rs b/src/daemon.rs index 632a546..ae08c3e 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -466,6 +466,12 @@ pub struct DaemonState { endpoint_recycle: Mutex<()>, last_endpoint_recycle: Mutex>, peer_book: Arc>, + expiry_policy: RwLock, + #[cfg(test)] + expiry_sweep_pause: std::sync::Mutex, + tokio::sync::oneshot::Receiver<()>, + )>>, policy_update: Arc>, runtime_peers: Option>>, pairing_gate: Arc, @@ -1288,6 +1294,7 @@ impl DaemonState { peer_connections.use_runtime_policy(peer_book.clone()); } + let expiry_policy = RwLock::new(peer_book.read().await.clone()); Ok(Arc::new(Self { config_write: Arc::new(std::sync::Mutex::new(())), home, @@ -1295,6 +1302,9 @@ impl DaemonState { endpoint_recycle: Mutex::new(()), last_endpoint_recycle: Mutex::new(None), peer_book, + expiry_policy, + #[cfg(test)] + expiry_sweep_pause: std::sync::Mutex::new(None), policy_update: Arc::new(Mutex::new(())), runtime_peers, pairing_gate, @@ -3556,6 +3566,7 @@ async fn serve(state: Arc) -> Result<()> { // Cancellation stops future sweeps but never drops an in-flight durable commit. // The existing policy drain owns its shutdown barrier. let _pairing_expiry = tokio::spawn(pairing::expire_loop(state.clone())); + let _runtime_peer_expiry = tokio::spawn(runtime_control::runtime_peer_expiry_loop(state.clone())); let result = tokio::select! { result = run_control_socket(control_listener, state.clone()) => result, diff --git a/src/daemon/runtime_control.rs b/src/daemon/runtime_control.rs index ef2ccb4..d6e9e94 100644 --- a/src/daemon/runtime_control.rs +++ b/src/daemon/runtime_control.rs @@ -14,10 +14,86 @@ async fn with_runtime( .await? } +/// Deadline enforcement never waits for the mutation sequencer or runtime-file +/// mutex. The cached policy has allocation-free, monotonic expiry generations. +pub(super) async fn runtime_peer_expiry_loop(state: Arc) { + let Some(runtime) = state.runtime_peers.clone() else { + return; + }; + let pending = Arc::new(std::sync::Mutex::new(Revocations { + admissions: vec![], + sessions: vec![], + deadline: tokio::time::Instant::now() + Duration::from_secs(2), + timed_out: Arc::new(std::sync::atomic::AtomicBool::new(false)), + })); + let writer_state = state.clone(); + let writer_pending = pending.clone(); + let writer = tokio::spawn(async move { + let mut ticks = tokio::time::interval(Duration::from_millis(100)); + ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = writer_state.cancel.cancelled() => return, + _ = ticks.tick() => {} + } + if writer_state.expiry_policy.read().await.expired_generation().is_none() { + continue; + } + let _serial = writer_state.policy_update.lock().await; + if writer_state.ensure_policy_writable().await.is_err() { + continue; + } + loop { + let transaction = with_runtime(runtime.clone(), |runtime| { + Ok(runtime.expired_transactions()?.into_iter().next()) + }).await; + let transaction = match transaction { + Ok(Some(transaction)) => transaction, + Ok(None) => break, + Err(error) => { + warn!(%error, "runtime peer expiry inspection failed"); + break; + } + }; + let node_id = transaction.operations[0].node_id(); + let result = writer_state.commit_runtime_transaction_with_revocations( + runtime.clone(), transaction, Some(writer_pending.clone()), + ).await; + match result { + Ok(receipt) => { + let mut pending = writer_pending.lock().unwrap(); + pending.admissions.retain(|entry| entry.peer != node_id || !receipt.cancellation.admissions.contains(&entry.id)); + pending.sessions.retain(|entry| entry.peer() != node_id || !receipt.cancellation.sessions.contains(&entry.id)); + } + Err(error) => { + warn!(%error, "runtime peer expiry persistence failed; expired access remains closed"); + break; + } + } + } + } + }); + let mut book = state.expiry_policy.read().await.clone(); + let mut observed_generation = None; + let mut ticks = tokio::time::interval(Duration::from_millis(50)); + ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = state.cancel.cancelled() => break, + _ = ticks.tick() => {} + } + state.enforce_runtime_expiry(&mut book, &mut observed_generation, &pending).await; + } + // Do not abort an in-flight durable commit on shutdown. + let _ = writer.await; +} + #[derive(Debug, Clone)] pub(super) struct Revocations { admissions: Vec, sessions: Vec, + deadline: tokio::time::Instant, + timed_out: Arc, } impl Revocations { fn pending(&self) -> Cancellation { @@ -61,6 +137,64 @@ impl Revocations { } } +impl DaemonState { + /// This secondary policy lock never covers disk I/O. Holding its read + /// generation through selection and signaling fences newer publications, + /// without waiting for a writer parked behind the main admission barrier. + async fn enforce_runtime_expiry( + &self, + book: &mut PeerBook, + observed_generation: &mut Option, + pending: &std::sync::Mutex, + ) { + let current = self.expiry_policy.read().await; + if !book.same_expiry_policy(¤t) { + *book = current.clone(); + *observed_generation = None; + } + let Some(generation) = book.expired_generation() else { return; }; + if *observed_generation == Some(generation) { + return; + } + *observed_generation = Some(generation); + #[cfg(test)] + { + let pause = self.expiry_sweep_pause.lock().unwrap().take(); + if let Some((selected, resume)) = pause { + let _ = selected.send(()); + let _ = resume.await; + } + } + let mut selected = self.select_runtime_revocations(book).await; + selected.deadline = tokio::time::Instant::now() + + Duration::from_secs(2).saturating_sub(book.expiry_lateness(generation)); + if selected.admissions.is_empty() && selected.sessions.is_empty() { + return; + } + { + let mut pending = pending.lock().unwrap(); + if pending.admissions.is_empty() && pending.sessions.is_empty() { + pending.deadline = selected.deadline; + pending.timed_out = selected.timed_out.clone(); + } + pending.admissions.extend(selected.admissions.iter().cloned()); + pending.sessions.extend(selected.sessions.iter().cloned()); + pending.admissions.sort_by_key(|entry| entry.id); + pending.admissions.dedup_by_key(|entry| entry.id); + pending.sessions.sort_by(|a, b| a.id.cmp(&b.id)); + pending.sessions.dedup_by(|a, b| a.id == b.id); + } + self.signal_runtime_revocations(&selected).await; + self.monitor_runtime_revocations(self.runtime_peers.as_ref().unwrap().clone(), None, selected); + } + + async fn publish_runtime_policy(&self, published: &mut PeerBook, book: PeerBook) { + let mut expiry = self.expiry_policy.write().await; + *published = book; + *expiry = published.clone(); + } +} + pub(super) enum ConfigUpdate { UpsertExpose(PersistedExpose), RemoveExpose(String), @@ -122,6 +256,8 @@ impl DaemonState { Revocations { admissions: self.admissions.select_withdrawn(book), sessions: self.tunnel_sessions.select_withdrawn(book).await, + deadline: tokio::time::Instant::now() + Duration::from_secs(2), + timed_out: Arc::new(std::sync::atomic::AtomicBool::new(false)), } } async fn signal_runtime_revocations(&self, selected: &Revocations) { @@ -146,12 +282,12 @@ impl DaemonState { let selected = Arc::new(selected); let pending = self.pending_policy_revocations.clone(); pending.lock().unwrap().push(selected.clone()); - let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let deadline = selected.deadline; let cancel = self.cancel.clone(); let policy = self.policy_update.clone(); let quarantine = self.policy_quarantine.clone(); tokio::spawn(async move { - let mut reported_timeout = false; + let mut reported_timeout = selected.timed_out.load(std::sync::atomic::Ordering::Relaxed); loop { let mut status = selected .status(tokio::time::Instant::now() >= deadline) @@ -161,6 +297,7 @@ impl DaemonState { status.state == CancellationState::TimedOut && !reported_timeout; if first_timeout { reported_timeout = true; + selected.timed_out.store(true, std::sync::atomic::Ordering::Relaxed); warn!(event = "policy_cancellation_timeout", code = "cancellation_timeout", ?request_id, admissions = ?status.remaining_admissions, sessions = ?status.remaining_sessions, "revocation did not finish within two seconds"); } // Keep the missed bound visible even after the remaining @@ -227,6 +364,8 @@ impl DaemonState { Revocations { admissions: vec![], sessions: vec![], + deadline: tokio::time::Instant::now() + Duration::from_secs(2), + timed_out: Arc::new(std::sync::atomic::AtomicBool::new(false)), } }; for previous in self.pending_policy_revocations.lock().unwrap().iter() { @@ -400,7 +539,7 @@ impl DaemonState { }); match reply { Ok(reply) => { - *published = book; + self.publish_runtime_policy(&mut published, book).await; self.peer_connections.presence.retry_now(); Ok(reply) } @@ -569,6 +708,7 @@ impl DaemonState { .collect::>(), Request::AddPeer { node_id, .. } | Request::SetGrants { node_id, .. } + | Request::SetExpiry { node_id, .. } | Request::RemovePeer { node_id, .. } => vec![*node_id], _ => vec![], }; @@ -643,18 +783,32 @@ impl DaemonState { &self, runtime: Arc>, transaction: runtime_peers::Transaction, + ) -> Result { + self.commit_runtime_transaction_with_revocations(runtime, transaction, None).await + } + + async fn commit_runtime_transaction_with_revocations( + &self, + runtime: Arc>, + transaction: runtime_peers::Transaction, + prior: Option>>, ) -> Result { self.ensure_policy_writable().await?; + let automatic_expiry = prior.is_some(); let prepared = { let runtime = runtime.clone(); let transaction = transaction.clone(); let home = self.home.clone(); tokio::task::spawn_blocking(move || { - let prepared = runtime + let runtime = runtime .lock() - .unwrap_or_else(|poison| poison.into_inner()) - .prepare(&transaction)?; - if !prepared.replay && prepared.no_op.is_none() { + .unwrap_or_else(|poison| poison.into_inner()); + let prepared = if automatic_expiry { + runtime.prepare_expiry(&transaction)? + } else { + runtime.prepare(&transaction)? + }; + if !automatic_expiry && !prepared.replay && prepared.no_op.is_none() { SyncBook::load(&home)?.validate_against(&prepared.book)?; } Ok::<_, anyhow::Error>(prepared) @@ -674,8 +828,37 @@ impl DaemonState { // writers are serialized; commit also rechecks the prepared revision. // Admissions see one durable, published generation. let mut published = self.peer_book.write().await; - let selected = self.select_runtime_revocations(&prepared.book).await; - let cancellation = selected.pending(); + let mut selected = self.select_runtime_revocations(&prepared.book).await; + if automatic_expiry { + let node_id = transaction.operations[0].node_id(); + selected.admissions.retain(|entry| entry.peer == node_id); + selected.sessions.retain(|entry| entry.peer() == node_id); + let policy = self.expiry_policy.read().await; + if let Some(generation) = policy.expired_generation() { + selected.deadline = tokio::time::Instant::now() + + Duration::from_secs(2).saturating_sub(policy.expiry_lateness(generation)); + } + } + if let Some(prior) = prior { + let prior = prior.lock().unwrap(); + let node_id = transaction.operations[0].node_id(); + if prior.admissions.iter().any(|entry| entry.peer == node_id) + || prior.sessions.iter().any(|entry| entry.peer() == node_id) + { + selected.deadline = selected.deadline.min(prior.deadline); + selected.timed_out = prior.timed_out.clone(); + } + selected.admissions.extend(prior.admissions.iter().filter(|entry| entry.peer == node_id).cloned()); + selected.sessions.extend(prior.sessions.iter().filter(|entry| entry.peer() == node_id).cloned()); + selected.admissions.sort_by_key(|entry| entry.id); + selected.admissions.dedup_by_key(|entry| entry.id); + selected.sessions.sort_by(|a, b| a.id.cmp(&b.id)); + selected.sessions.dedup_by(|a, b| a.id == b.id); + } + let mut cancellation = selected.status(tokio::time::Instant::now() >= selected.deadline).await; + if selected.timed_out.load(std::sync::atomic::Ordering::Relaxed) { + cancellation.state = CancellationState::TimedOut; + } let book = prepared.book.clone(); let receipt = { let runtime = runtime.clone(); @@ -705,7 +888,7 @@ impl DaemonState { return Err(error); } }; - *published = book; + self.publish_runtime_policy(&mut published, book).await; self.signal_runtime_revocations(&selected).await; drop(published); self.monitor_runtime_revocations(runtime, Some(transaction.request_id), selected); @@ -742,7 +925,7 @@ impl DaemonState { let book = result?; let mut published = self.peer_book.write().await; let selected = self.select_runtime_revocations(&book).await; - *published = book; + self.publish_runtime_policy(&mut published, book).await; self.signal_runtime_revocations(&selected).await; *self.policy_quarantine.lock().await = None; drop(published); @@ -809,7 +992,9 @@ impl DaemonState { .into()) }); match result { - Ok(book) => *published = book, + Ok(book) => { + self.publish_runtime_policy(&mut published, book).await; + } Err(error) => { let indeterminate = crate::error::code_of(&error) == ErrorCode::RuntimeCommitIndeterminate; @@ -978,6 +1163,324 @@ mod tests { .await? } + async fn wait_expiry_receipt(home: &FabricHome, request_id: &str) -> Result { + tokio::time::timeout(Duration::from_secs(4), async { + loop { + let reply = request(home, Request::Cancellation { request_id: request_id.into() }).await?; + if let Some(receipt) = reply.mutation { + if receipt.cancellation.state != CancellationState::Pending { + return Ok::<_, anyhow::Error>(receipt); + } + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await? + } + + #[tokio::test] + async fn expiry_revokes_mux_stream_and_retains_automatic_cancellation_receipt() -> Result<()> { + let _serial = SERIAL.lock().await; + let server_dir = tempfile::tempdir()?; + let client_dir = tempfile::tempdir()?; + let home = FabricHome::new(server_dir.path()); + let client_home = FabricHome::new(client_dir.path()); + let server = FabricNode::start_with_daemon_options(home.clone(), options()).await?; + let client = FabricNode::start(client_home.clone()).await?; + trust_server(&client_home, &client, &server).await?; + let (echo, task) = echo_server().await?; + server.expose_tcp("work/0", echo).await?; + let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 600; + let deadline_signed = i64::try_from(deadline)?; + mutate(&home, "expiring", 0, vec![ + add(&client, "client"), grant(&client, &["work/0"]), + Operation::SetExpiry { node_id: client.id(), expires_at: deadline_signed }, + ]).await?; + let addr = client.dial_tcp("server", "work/0", "127.0.0.1:0".into()).await?; + let mut stream = tokio::net::TcpStream::connect(&addr).await?; + round_trip(&mut stream, b"live-before-expiry").await?; + with_runtime(server.state().runtime_peers.as_ref().unwrap().clone(), move |runtime| { + runtime.advance_expiry_for_test(deadline); + Ok(()) + }).await?; + let request_id = runtime_peers::RuntimePeers::expiry_request_id("controller", client.id(), deadline_signed, "expiring"); + let receipt = wait_expiry_receipt(&home, &request_id).await?; + assert_eq!(receipt.committed_revision, 2); + assert_eq!(receipt.cancellation.state, CancellationState::Complete); + assert!(!receipt.cancellation.admissions.is_empty()); + assert!(!receipt.cancellation.sessions.is_empty()); + assert!(receipt.removals[0].effective_trust_removed); + assert!(server.state().may(&client.id(), "work/0").await.is_err()); + let mut byte = [0]; + let ended = tokio::time::timeout(Duration::from_secs(1), stream.read(&mut byte)).await?; + assert!(!matches!(ended, Ok(1..))); + assert!(client.state().open_stream("server", b"work/0").await.is_err()); + client.shutdown().await?; + server.shutdown().await?; + task.abort(); + Ok(()) + } + + #[tokio::test] + async fn expiry_cancellation_bypasses_blocked_policy_publication_and_disk_writers() -> Result<()> { + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let server = FabricNode::start_with_daemon_options(home.clone(), options()).await?; + let state = server.state(); + let peer = iroh::SecretKey::from_bytes(&[42; 32]).public(); + let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 600; + let deadline_signed = i64::try_from(deadline)?; + mutate(&home, "expiring", 0, vec![ + Operation::AddPeer { node_id: peer, name: None, direction: None, addr: None }, + Operation::SetGrants { node_id: peer, grants: vec!["exec".into(), "echo".into()] }, + Operation::SetExpiry { node_id: peer, expires_at: deadline_signed }, + ]).await?; + let residual = request(&home, transaction("residual", "residual", 1, vec![ + Operation::AddPeer { node_id: peer, name: None, direction: None, addr: None }, + Operation::SetGrants { node_id: peer, grants: vec!["echo".into()] }, + ])).await?; + assert!(residual.error.is_none()); + let expiring = state.admissions.admit(peer, crate::admission::Grant::Service("exec".into())); + let residual = state.admissions.admit(peer, crate::admission::Grant::Service("echo".into())); + let connection = state.admissions.admit(peer, crate::admission::Grant::Trust); + let book = state.peer_book.read().await.clone(); + let policy = state.policy_update.lock().await; + let publication = state.peer_book.write().await; + let runtime = state.runtime_peers.as_ref().unwrap().clone(); + let (locked_tx, locked_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let disk_worker = tokio::task::spawn_blocking(move || { + let disk = runtime.lock().unwrap(); + disk.advance_expiry_for_test(deadline); + let _ = locked_tx.send(()); + let _ = release_rx.recv(); + }); + locked_rx.await?; + assert!(book.may(&peer, "exec").is_err()); + assert!(book.may(&peer, "echo").is_ok()); + tokio::time::timeout(Duration::from_secs(2), expiring.revoked().cancelled()).await?; + assert!(!residual.revoked().is_cancelled()); + assert!(!connection.revoked().is_cancelled()); + drop(expiring); + release_tx.send(())?; + disk_worker.await?; + drop(publication); + drop(policy); + let request_id = runtime_peers::RuntimePeers::expiry_request_id("controller", peer, deadline_signed, "expiring"); + let receipt = wait_expiry_receipt(&home, &request_id).await?; + assert!(!receipt.removals[0].effective_trust_removed); + assert!(!receipt.cancellation.admissions.is_empty()); + assert_eq!(receipt.cancellation.state, CancellationState::Complete); + drop(residual); + drop(connection); + server.shutdown().await?; + Ok(()) + } + + #[tokio::test] + async fn simultaneous_expiry_receipts_name_only_their_own_cancelled_work() -> Result<()> { + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let server = FabricNode::start_with_daemon_options(home.clone(), options()).await?; + let state = server.state(); + let peers = [44, 45].map(|byte| iroh::SecretKey::from_bytes(&[byte; 32]).public()); + let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 600; + let deadline_signed = i64::try_from(deadline)?; + let mut operations = Vec::new(); + for peer in peers { + operations.push(Operation::AddPeer { node_id: peer, name: None, direction: None, addr: None }); + operations.push(Operation::SetExpiry { node_id: peer, expires_at: deadline_signed }); + } + mutate(&home, "expiring-together", 0, operations).await?; + let first = state.admissions.admit(peers[0], crate::admission::Grant::Trust); + let second = state.admissions.admit(peers[1], crate::admission::Grant::Trust); + // Hold cleanup until the sweeper has selected both records' work. + let policy = state.policy_update.lock().await; + with_runtime(state.runtime_peers.as_ref().unwrap().clone(), move |runtime| { + runtime.advance_expiry_for_test(deadline); + Ok(()) + }).await?; + tokio::time::timeout(Duration::from_secs(2), first.revoked().cancelled()).await?; + tokio::time::timeout(Duration::from_secs(2), second.revoked().cancelled()).await?; + let book = state.peer_book.read().await.clone(); + let selected = state.admissions.select_withdrawn(&book); + let ids = peers.map(|peer| selected.iter().find(|entry| entry.peer == peer).unwrap().id); + drop(first); + drop(second); + drop(policy); + for (peer, admission_id) in peers.into_iter().zip(ids) { + let request_id = runtime_peers::RuntimePeers::expiry_request_id( + "controller", peer, deadline_signed, "expiring-together", + ); + let receipt = wait_expiry_receipt(&home, &request_id).await?; + assert_eq!(receipt.cancellation.admissions, [admission_id]); + assert!(receipt.cancellation.sessions.is_empty()); + assert_eq!(receipt.cancellation.state, CancellationState::Complete); + assert_eq!(receipt.removals.len(), 1); + assert_eq!(receipt.removals[0].node_id, peer); + } + server.shutdown().await?; + Ok(()) + } + + #[tokio::test] + async fn expiry_timeout_evidence_survives_a_delayed_cleanup_receipt() -> Result<()> { + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let server = FabricNode::start_with_daemon_options(home.clone(), options()).await?; + let state = server.state(); + let peer = iroh::SecretKey::from_bytes(&[43; 32]).public(); + let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 600; + let deadline_signed = i64::try_from(deadline)?; + mutate(&home, "expiring", 0, vec![ + Operation::AddPeer { node_id: peer, name: None, direction: None, addr: None }, + Operation::SetExpiry { node_id: peer, expires_at: deadline_signed }, + ]).await?; + let admission = state.admissions.admit(peer, crate::admission::Grant::Trust); + let policy = state.policy_update.lock().await; + with_runtime(state.runtime_peers.as_ref().unwrap().clone(), move |runtime| { + runtime.advance_expiry_for_test(deadline); + Ok(()) + }).await?; + tokio::time::timeout(Duration::from_secs(2), admission.revoked().cancelled()).await?; + // Keep the real admission alive past its cancellation bound while the + // durable sequencer is blocked. No mock completion or sleeping guess. + tokio::time::timeout(Duration::from_secs(3), async { + loop { + if state.pending_policy_revocations.lock().unwrap().iter() + .any(|selected| selected.timed_out.load(std::sync::atomic::Ordering::Relaxed)) + { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await?; + drop(admission); + drop(policy); + let request_id = runtime_peers::RuntimePeers::expiry_request_id( + "controller", peer, deadline_signed, "expiring", + ); + let receipt = wait_expiry_receipt(&home, &request_id).await?; + assert_eq!(receipt.cancellation.state, CancellationState::TimedOut); + assert!(receipt.cancellation.remaining_admissions.is_empty()); + assert_eq!(receipt.cancellation.admissions.len(), 1); + server.shutdown().await?; + Ok(()) + } + + #[tokio::test] + async fn expiry_sweep_refreshes_stale_policy_and_preserves_new_residual_sessions() -> Result<()> { + let _serial = SERIAL.lock().await; + let server_dir = tempfile::tempdir()?; + let client_dir = tempfile::tempdir()?; + let home = FabricHome::new(server_dir.path()); + let client_home = FabricHome::new(client_dir.path()); + let server = FabricNode::start_with_daemon_options(home.clone(), options()).await?; + let client = FabricNode::start(client_home.clone()).await?; + let state = server.state(); + trust_server(&client_home, &client, &server).await?; + let (echo, task) = echo_server().await?; + server.expose_tcp("kept/0", echo.clone()).await?; + server.expose_tcp("withdrawn/0", echo).await?; + let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 600; + mutate(&home, "expiring", 0, vec![ + add(&client, "client"), grant(&client, &["kept/0", "withdrawn/0"]), + Operation::SetExpiry { node_id: client.id(), expires_at: i64::try_from(deadline)? }, + ]).await?; + // Deterministically pause the sweep with the exact stale generation it + // would have cached before a concurrent residual-policy publication. + let mut stale = state.expiry_policy.read().await.clone(); + let reply = request(&home, transaction("residual", "residual", 1, vec![ + add(&client, "client"), grant(&client, &["kept/0"]), + ])).await?; + assert!(reply.error.is_none()); + let kept_addr = client.dial_tcp("server", "kept/0", "127.0.0.1:0".into()).await?; + let withdrawn_addr = client.dial_tcp("server", "withdrawn/0", "127.0.0.1:0".into()).await?; + let mut kept_stream = tokio::net::TcpStream::connect(&kept_addr).await?; + let mut withdrawn_stream = tokio::net::TcpStream::connect(&withdrawn_addr).await?; + round_trip(&mut kept_stream, b"new-residual-session").await?; + round_trip(&mut withdrawn_stream, b"last-expiring-source").await?; + let kept = state.admissions.admit(client.id(), crate::admission::Grant::Service("kept/0".into())); + let withdrawn = state.admissions.admit(client.id(), crate::admission::Grant::Service("withdrawn/0".into())); + let pending = std::sync::Mutex::new(Revocations { + admissions: vec![], sessions: vec![], + deadline: tokio::time::Instant::now() + Duration::from_secs(2), + timed_out: Arc::new(std::sync::atomic::AtomicBool::new(false)), + }); + let policy = state.policy_update.lock().await; + let (paused_tx, paused_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); + *state.expiry_sweep_pause.lock().unwrap() = Some((paused_tx, resume_rx)); + with_runtime(state.runtime_peers.as_ref().unwrap().clone(), move |runtime| { + runtime.advance_expiry_for_test(deadline); + Ok(()) + }).await?; + let sweep_state = state.clone(); + let sweep = tokio::spawn(async move { + let mut observed = None; + sweep_state.enforce_runtime_expiry(&mut stale, &mut observed, &pending).await; + }); + tokio::time::timeout(Duration::from_secs(2), paused_rx).await??; + // A publication cannot pass a sweep paused between refresh and signal. + assert!(state.expiry_policy.try_write().is_err()); + assert!(resume_tx.send(()).is_ok()); + sweep.await?; + tokio::time::timeout(Duration::from_secs(2), withdrawn.revoked().cancelled()).await?; + assert!(!kept.revoked().is_cancelled()); + round_trip(&mut kept_stream, b"still-permitted-after-expiry").await?; + let mut byte = [0]; + let ended = tokio::time::timeout(Duration::from_secs(2), withdrawn_stream.read(&mut byte)).await?; + assert!(!matches!(ended, Ok(1..))); + drop(withdrawn); + drop(kept); + drop(policy); + client.shutdown().await?; + server.shutdown().await?; + task.abort(); + Ok(()) + } + + #[tokio::test] + async fn expiry_forward_observation_then_rollback_cancels_live_work_after_600_elapsed_seconds() -> Result<()> { + let _serial = SERIAL.lock().await; + let dir = tempfile::tempdir()?; + let home = FabricHome::new(dir.path()); + let server = FabricNode::start_with_daemon_options(home.clone(), options()).await?; + let state = server.state(); + let peer = iroh::SecretKey::from_bytes(&[44; 32]).public(); + let forward = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs() + 3_600; + with_runtime(state.runtime_peers.as_ref().unwrap().clone(), move |runtime| { + runtime.advance_expiry_for_test(forward); + Ok(()) + }).await?; + let deadline = i64::try_from(forward + 600)?; + mutate(&home, "expiring", 0, vec![ + Operation::AddPeer { node_id: peer, name: None, direction: None, addr: None }, + Operation::SetExpiry { node_id: peer, expires_at: deadline }, + ]).await?; + let admission = state.admissions.admit(peer, crate::admission::Grant::Trust); + let book = state.peer_book.read().await.clone(); + let policy = state.policy_update.lock().await; + with_runtime(state.runtime_peers.as_ref().unwrap().clone(), |runtime| { + runtime.elapse_expiry_for_test(Duration::from_secs(600)); + Ok(()) + }).await?; + assert!(!book.accepts_inbound(&peer)); + tokio::time::timeout(Duration::from_secs(2), admission.revoked().cancelled()).await?; + drop(admission); + drop(policy); + let request_id = runtime_peers::RuntimePeers::expiry_request_id( + "controller", peer, deadline, "expiring", + ); + let receipt = wait_expiry_receipt(&home, &request_id).await?; + assert!(receipt.removals[0].effective_trust_removed); + assert!(!receipt.cancellation.admissions.is_empty()); + server.shutdown().await?; + Ok(()) + } + async fn pending_confirmation( state: &DaemonState, creation_id: &str, diff --git a/src/runtime_peers.rs b/src/runtime_peers.rs index 4f3a3be..5100b2e 100644 --- a/src/runtime_peers.rs +++ b/src/runtime_peers.rs @@ -13,7 +13,7 @@ use iroh::{EndpointAddr, EndpointId}; use serde::{Deserialize, Serialize}; use crate::{ - config::{FabricHome, Peer, PeerBook}, + config::{ExpiryClock, FabricHome, Peer, PeerBook}, error::{ErrorCode, FabricError}, }; @@ -27,6 +27,7 @@ const RECEIPT_SECONDS: u64 = 24 * 60 * 60; const MAX_RECEIPTS: usize = 1024; const WITHDRAWAL_RECEIPTS: usize = 64; const MAX_RECORDS: usize = 1024; +pub const MAX_EXPIRY_SECONDS: u64 = 600; #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -68,6 +69,10 @@ pub enum Operation { node_id: EndpointId, grants: Vec, }, + SetExpiry { + node_id: EndpointId, + expires_at: i64, + }, RemovePeer { node_id: EndpointId, }, @@ -77,6 +82,7 @@ impl Operation { match self { Self::AddPeer { node_id, .. } | Self::SetGrants { node_id, .. } + | Self::SetExpiry { node_id, .. } | Self::RemovePeer { node_id } => *node_id, } } @@ -121,6 +127,13 @@ pub enum Request { node_id: EndpointId, grants: Vec, }, + SetExpiry { + owner: String, + request_id: String, + expected_revision: u64, + node_id: EndpointId, + expires_at: i64, + }, RemovePeer { owner: String, request_id: String, @@ -166,6 +179,18 @@ impl Request { expected_revision, Operation::SetGrants { node_id, grants }, ), + Self::SetExpiry { + owner, + request_id, + expected_revision, + node_id, + expires_at, + } => ( + owner, + request_id, + expected_revision, + Operation::SetExpiry { node_id, expires_at }, + ), Self::RemovePeer { owner, request_id, @@ -192,7 +217,13 @@ impl Request { #[serde(tag = "kind", rename_all = "snake_case")] pub enum Source { Static { path: PathBuf }, - Runtime { owner: String }, + Runtime { + owner: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + expires_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + expiry_request_id: Option, + }, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct EffectivePeer { @@ -294,6 +325,10 @@ pub struct Reply { struct Record { owner: String, peer: Peer, + #[serde(default, skip_serializing_if = "Option::is_none")] + expires_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + expiry_request_id: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -318,6 +353,8 @@ struct Stored { pairing: Vec, #[serde(default, skip_serializing_if = "is_zero")] pairing_last_seen: u64, + #[serde(default, skip_serializing_if = "is_zero")] + expiry_last_seen: u64, } fn is_zero(value: &u64) -> bool { @@ -386,6 +423,7 @@ pub(crate) struct RuntimePeers { static_input: StaticInput, fault: Option, pairing_clock: (u64, std::time::Instant), + expiry_clock: std::sync::Arc, exchanges: BTreeMap, #[cfg(test)] fail_directory_sync: bool, @@ -424,12 +462,13 @@ fn validate_token(value: &str, field: &str) -> Result<()> { impl RuntimePeers { /// A bad runtime file keeps control available with all network policy closed. pub fn open(home: &FabricHome) -> (Self, PeerBook) { - Self::open_with_ended_work(home, true, (now(), std::time::Instant::now())) + Self::open_with_ended_work(home, true, (now(), std::time::Instant::now()), None) } fn open_with_ended_work( home: &FabricHome, restarted: bool, pairing_clock: (u64, std::time::Instant), + expiry_clock: Option>, ) -> (Self, PeerBook) { let path = home.root().join("runtime-peers.toml"); let placeholder = StaticInput { @@ -447,11 +486,13 @@ impl RuntimePeers { records: vec![], receipts: vec![], pairing: vec![], + expiry_last_seen: 0, pairing_last_seen: 0, }, static_input: placeholder, fault: None, pairing_clock, + expiry_clock: expiry_clock.unwrap_or_else(|| std::sync::Arc::new(ExpiryClock::new(pairing_clock))), exchanges: BTreeMap::new(), #[cfg(test)] fail_directory_sync: false, @@ -518,6 +559,16 @@ impl RuntimePeers { } let timestamp = store.pairing_timestamp(); changed |= pairing::recover(&mut store.stored, timestamp, restarted); + if now() < store.stored.expiry_last_seen { + let deadline = store.stored.records.iter().filter_map(|record| record.expires_at) + .filter_map(|deadline| u64::try_from(deadline).ok()).max().unwrap_or(0); + store.expiry_clock.timestamp(deadline); + } + loop { + let Some(transaction) = store.expired_transactions()?.into_iter().next() else { break; }; + let prepared = store.prepare_expiry(&transaction)?; + store.commit(prepared, &transaction.request_id, Cancellation::empty())?; + } let book = store.compose(&store.stored)?.0; if changed { store.persist(&store.stored)?; @@ -543,12 +594,34 @@ impl RuntimePeers { let mut records = BTreeSet::new(); for record in &self.stored.records { validate_token(&record.owner, "owner")?; + if record.expires_at.is_some_and(|deadline| deadline <= 0) { + // Reject impossible timestamps before converting them to a + // duration or serializing them through TOML's signed integers. + return Err(unavailable("invalid runtime record expires_at")); + } + if record.expires_at.is_some_and(|deadline| { + u64::try_from(deadline).is_ok_and(|deadline| { + deadline.saturating_sub(self.stored.expiry_last_seen) > MAX_EXPIRY_SECONDS + }) + }) { + return Err(unavailable("runtime expiry exceeds its persisted 600-second bound")); + } + if record.expires_at.is_some() != record.expiry_request_id.is_some() { + return Err(unavailable("runtime expiry requires its reserved cancellation receipt ID")); + } + if let Some(request_id) = &record.expiry_request_id { + validate_token(request_id, "expiry_request_id")?; + if self.stored.receipts.iter().any(|receipt| &receipt.request_id == request_id) { + return Err(unavailable("runtime expiry receipt ID is already committed")); + } + } if !records.insert((record.owner.clone(), record.peer.id)) { return Err(unavailable("duplicate owner/node runtime record")); } } - if self.stored.receipts.len() > MAX_RECEIPTS { - return Err(unavailable("too many retained runtime receipts")); + let reserved = self.stored.records.iter().filter(|record| record.expires_at.is_some()).count(); + if self.stored.receipts.len().saturating_add(reserved) > MAX_RECEIPTS { + return Err(unavailable("too many retained or reserved runtime receipts")); } let mut receipts = BTreeSet::new(); for receipt in &self.stored.receipts { @@ -577,7 +650,9 @@ impl RuntimePeers { pub fn reconcile(&mut self, home: &FabricHome) -> Result { // Reconciliation is still the same daemon lifetime. Carry the original // monotonic anchor into recovery, before it decides whether permits expired. - let (mut loaded, book) = Self::open_with_ended_work(home, false, self.pairing_clock); + let (mut loaded, book) = Self::open_with_ended_work( + home, false, self.pairing_clock, Some(self.expiry_clock.clone()), + ); if let Some(error) = loaded.fault.take() { self.fault = Some(error.clone()); return Err(unavailable(error)); @@ -610,6 +685,7 @@ impl RuntimePeers { static_input: input, fault: None, pairing_clock: self.pairing_clock, + expiry_clock: self.expiry_clock.clone(), exchanges: BTreeMap::new(), #[cfg(test)] fail_directory_sync: false, @@ -652,6 +728,19 @@ impl RuntimePeers { } fn compose(&self, stored: &Stored) -> Result<(PeerBook, Vec)> { + let timestamp = self.expiry_timestamp(); + let (book, views) = self.compose_at(stored, timestamp)?; + let deadlines: BTreeSet<_> = stored.records.iter() + .filter_map(|record| record.expires_at.and_then(|value| u64::try_from(value).ok())) + .collect(); + let generations = deadlines.into_iter().map(|deadline| { + self.compose_at(stored, deadline) + .map(|(book, _)| (deadline, book.into_effective_peers())) + }).collect::>>()?; + Ok((book.with_expiry_generations(self.expiry_clock.clone(), generations), views)) + } + + fn compose_at(&self, stored: &Stored, timestamp: u64) -> Result<(PeerBook, Vec)> { let mut nodes: BTreeMap> = BTreeMap::new(); for peer in self.static_input.book.peers() { nodes.entry(peer.id).or_default().push(( @@ -662,10 +751,17 @@ impl RuntimePeers { )); } for record in &stored.records { + if record.expires_at.is_some_and(|deadline| { + u64::try_from(deadline).is_ok_and(|deadline| deadline <= timestamp) + }) { + continue; + } nodes.entry(record.peer.id).or_default().push(( &record.peer, Source::Runtime { owner: record.owner.clone(), + expires_at: record.expires_at, + expiry_request_id: record.expiry_request_id.clone(), }, )); } @@ -722,10 +818,73 @@ impl RuntimePeers { Ok((self.static_input.book.with_effective_peers(peers)?, views)) } + #[cfg(test)] + pub(crate) fn advance_expiry_for_test(&self, timestamp: u64) { + self.expiry_clock.timestamp(timestamp); + } + + #[cfg(test)] + pub(crate) fn elapse_expiry_for_test(&self, elapsed: std::time::Duration) { + self.expiry_clock.elapse_for_test(elapsed); + } + + fn expiry_timestamp(&self) -> u64 { + self.expiry_clock.timestamp(self.stored.pairing_last_seen.max(self.stored.expiry_last_seen)) + } + + /// Deterministic IDs let a caller find automatic revocation receipts. + pub(crate) fn expiry_request_id(owner: &str, node_id: EndpointId, deadline: i64, request_id: &str) -> String { + let mut digest = blake3::Hasher::new(); + for value in [owner, &node_id.to_string(), &deadline.to_string(), request_id] { + digest.update(&(value.len() as u64).to_le_bytes()); + digest.update(value.as_bytes()); + } + format!("expiry:{}", digest.finalize().to_hex()) + } + + pub(crate) fn expired_transactions(&self) -> Result> { + self.ensure_available()?; + let timestamp = self.expiry_timestamp(); + Ok(self.stored.records.iter().filter_map(|record| { + let deadline = record.expires_at?; + if u64::try_from(deadline).ok()? > timestamp { + return None; + } + Some(Transaction { + owner: record.owner.clone(), + request_id: record.expiry_request_id.clone()?, + expected_revision: self.stored.revision, + operations: vec![Operation::RemovePeer { node_id: record.peer.id }], + }) + }).collect()) + } pub fn prepare(&self, tx: &Transaction) -> Result { + self.prepare_transaction(tx, false) + } + + pub(crate) fn prepare_expiry(&self, tx: &Transaction) -> Result { + self.prepare_transaction(tx, true) + } + + fn prepare_transaction(&self, tx: &Transaction, automatic_expiry: bool) -> Result { self.ensure_available()?; validate_token(&tx.owner, "owner")?; validate_token(&tx.request_id, "request_id")?; + if tx.request_id.starts_with("expiry:") != automatic_expiry { + return Err(invalid("the expiry: request_id prefix is reserved for automatic expiry")); + } + if let Some(record) = self.stored.records.iter().find(|record| { + record.expiry_request_id.as_deref() == Some(tx.request_id.as_str()) + }) { + let due = record.expires_at.and_then(|deadline| u64::try_from(deadline).ok()) + .is_some_and(|deadline| deadline <= self.expiry_timestamp()); + if !due || record.owner != tx.owner || !matches!( + tx.operations.as_slice(), + [Operation::RemovePeer { node_id }] if *node_id == record.peer.id + ) { + return Err(invalid("request_id is reserved for automatic expiry")); + } + } let encoded = serde_json::to_vec(tx)?; if encoded.len() > MAX_REQUEST_BYTES || tx.operations.is_empty() @@ -824,6 +983,8 @@ impl RuntimePeers { next.records.push(Record { owner: tx.owner.clone(), peer, + expires_at: None, + expiry_request_id: None, }); } } @@ -839,6 +1000,28 @@ impl RuntimePeers { grants.dedup(); next.records[index].peer.allow = grants; } + Operation::SetExpiry { expires_at, .. } => { + let Some(index) = index else { + return Err(self.ownership_error(&next, id)); + }; + let record = &mut next.records[index]; + if let Some(existing) = record.expires_at { + if existing != *expires_at { + return Err(invalid("expires_at is immutable once set")); + } + } else { + let deadline = u64::try_from(*expires_at) + .map_err(|_| invalid("expires_at must be positive Unix seconds"))?; + let timestamp = self.expiry_timestamp(); + if deadline <= timestamp + || deadline.saturating_sub(timestamp) > MAX_EXPIRY_SECONDS + { + return Err(invalid("expires_at must be within the next 600 seconds")); + } + record.expires_at = Some(*expires_at); + record.expiry_request_id = Some(Self::expiry_request_id(&tx.owner, id, *expires_at, &tx.request_id)); + } + } Operation::RemovePeer { .. } => { if index.is_none() && (self.static_input.book.peers().iter().any(|p| p.id == id) @@ -857,6 +1040,21 @@ impl RuntimePeers { } } } + let mut expiry_ids = BTreeSet::new(); + for record in &next.records { + if let Some(request_id) = &record.expiry_request_id { + if !expiry_ids.insert(request_id) + || request_id == &tx.request_id + || next.receipts.iter().any(|receipt| &receipt.request_id == request_id) + { + return Err(FabricError::new( + ErrorCode::RequestIdConflict, + "automatic expiry request_id conflicts with a retained or reserved receipt", + ).into()); + } + } + } + drop(expiry_ids); next.records .sort_by(|a, b| (&a.owner, a.peer.id).cmp(&(&b.owner, b.peer.id))); if next.records.len() > MAX_RECORDS { @@ -912,14 +1110,15 @@ impl RuntimePeers { .iter() .find(|r| r.owner == tx.owner && r.peer.id == *node_id) .is_some_and(|record| grants.iter().all(|grant| record.peer.allow.contains(grant))), - Operation::AddPeer { .. } => false, + Operation::AddPeer { .. } | Operation::SetExpiry { .. } => false, }); let limit = if withdrawal_only { MAX_RECEIPTS } else { MAX_RECEIPTS - WITHDRAWAL_RECEIPTS }; - if next.receipts.len() >= limit { + let reserved = next.records.iter().filter(|record| record.expires_at.is_some()).count(); + if next.receipts.len().saturating_add(reserved) >= limit { return Err(FabricError::new( ErrorCode::RuntimeReceiptCapacity, "unexpired runtime receipt capacity is full for this mutation", @@ -980,6 +1179,11 @@ impl RuntimePeers { return Ok(reply); } if !next.replay { + if next.stored.records.iter().any(|record| record.expires_at.is_some()) + || next.stored.expiry_last_seen != 0 + { + next.stored.expiry_last_seen = self.expiry_timestamp(); + } // Completion records can advance while validation runs off-thread. // Preserve that metadata when committing a later policy candidate. for receipt in &mut next.stored.receipts { @@ -1194,6 +1398,10 @@ mod tests { let next = store.prepare(tx)?; store.commit(next, &tx.request_id, Cancellation::empty()) } + fn commit_expiry(store: &mut RuntimePeers, tx: &Transaction) -> Result { + let next = store.prepare_expiry(tx)?; + store.commit(next, &tx.request_id, Cancellation::empty()) + } fn open(dir: &tempfile::TempDir) -> RuntimePeers { let (store, _) = RuntimePeers::open(&FabricHome::new(dir.path())); assert!(store.fault().is_none(), "{:?}", store.fault()); @@ -1358,7 +1566,7 @@ mod tests { ); assert_eq!( result.affected[0].services["exec"], - vec![Source::Runtime { owner: "b".into() }] + vec![Source::Runtime { owner: "b".into(), expires_at: None, expiry_request_id: None }] ); assert_eq!( crate::error::code_of( @@ -1816,4 +2024,275 @@ mod tests { assert_eq!(store.stored.records.len(), MAX_RECORDS - 1); Ok(()) } + #[test] + fn fresh_record_expiry_accepts_only_the_next_one_through_600_seconds() -> Result<()> { + for (offset, accepted) in [(0, false), (1, true), (600, true), (601, false)] { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + // A forward observation keeps the real clock from moving the boundary. + let timestamp = now() + 3_600; + store.advance_expiry_for_test(timestamp); + commit(&mut store, &tx("owner", "fresh", 0, vec![add(1, "one")]))?; + let request = tx("owner", "schedule", 1, vec![ + Operation::SetExpiry { node_id: id(1), expires_at: i64::try_from(timestamp + offset)? }, + ]); + if accepted { + let receipt = commit(&mut store, &request)?; + assert_eq!(receipt.committed_revision, 2); + assert_eq!(store.stored.records[0].expires_at, Some(i64::try_from(timestamp + offset)?)); + } else { + let error = store.prepare(&request).err().context("out-of-bounds deadline was accepted")?; + assert_eq!(crate::error::code_of(&error), ErrorCode::InvalidArgument); + assert_eq!(store.revision(), 1); + assert!(store.stored.records[0].expires_at.is_none()); + } + } + Ok(()) + } + #[test] + fn expiry_is_bounded_immutable_and_overlap_batch_replays_without_extension() -> Result<()> { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + commit(&mut store, &tx("owner", "old", 0, vec![add(1, "old"), grants(1, &["exec", "echo"])]))?; + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + let overlap = tx("owner", "overlap", 1, vec![ + add(2, "next"), grants(2, &["exec", "echo"]), + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ]); + let installed = commit(&mut store, &overlap)?; + assert_eq!(installed.committed_revision, 2); + assert_eq!(store.stored.records.iter().find(|record| record.peer.id == id(1)).unwrap().expires_at, Some(deadline)); + assert_eq!(commit(&mut store, &overlap)?.committed_revision, 2); + let mut changed_deadline = overlap.clone(); + if let Operation::SetExpiry { expires_at, .. } = &mut changed_deadline.operations[2] { + *expires_at -= 1; + } + let error = store.prepare(&changed_deadline).err().context("changed deadline replay was accepted")?; + assert_eq!(crate::error::code_of(&error), ErrorCode::RequestIdConflict); + for value in [deadline - 1, deadline + 1, 0, -1, i64::MAX] { + assert!(store.prepare(&tx("owner", &format!("change-{value}"), 2, vec![ + Operation::SetExpiry { node_id: id(1), expires_at: value }, + ])).is_err()); + } + assert!(store.prepare(&tx("other", "wrong-owner", 2, vec![ + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ])).is_err()); + assert!(store.prepare(&tx("owner", "too-far", 2, vec![ + Operation::SetExpiry { node_id: id(2), expires_at: deadline + 10 }, + ])).is_err()); + let unchanged = fs::read(&store.path)?; + assert!(store.prepare(&tx("owner", "bad-atomic", 2, vec![ + add(3, "third"), + Operation::SetExpiry { node_id: id(1), expires_at: deadline + 1 }, + ])).is_err()); + assert_eq!(fs::read(&store.path)?, unchanged); + store.advance_expiry_for_test(u64::try_from(deadline)?); + let expired = store.expired_transactions()?.remove(0); + let removed = commit_expiry(&mut store, &expired)?; + assert!(removed.removals[0].effective_trust_removed); + assert_eq!(removed.committed_revision, 3); + let replay = commit(&mut store, &overlap)?; + assert_eq!(replay.committed_revision, 2); + assert_eq!(replay.current_revision, 3); + assert_eq!(replay.affected.len(), 1); + assert_eq!(replay.affected[0].node_id, id(2)); + assert_eq!(store.stored.records.len(), 1); + Ok(()) + } + + #[test] + fn timed_policy_preserves_residual_sources_without_a_cleanup_commit() -> Result<()> { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + store.static_input.book.add_with_allow( + id(1), Some("one".into()), None, Some(vec!["static/0".into()]), + ); + commit(&mut store, &tx("expiring", "install", 0, vec![add(1, "one"), grants(1, &["exec"])]))?; + commit(&mut store, &tx("residual", "residual", 1, vec![add(1, "one"), grants(1, &["echo"])]))?; + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + commit(&mut store, &tx("expiring", "deadline", 2, vec![ + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ]))?; + let book = store.compose(&store.stored)?.0; + assert!(book.may(&id(1), "exec").is_ok()); + assert!(book.may(&id(1), "echo").is_ok()); + store.advance_expiry_for_test(u64::try_from(deadline)?); + assert!(book.may(&id(1), "exec").is_err()); + assert!(book.may(&id(1), "static/0").is_ok()); + assert!(book.may(&id(1), "echo").is_ok()); + assert!(book.accepts_inbound(&id(1))); + assert_eq!(store.revision(), 3); + assert_eq!(store.stored.records.len(), 2); + let transaction = store.expired_transactions()?.remove(0); + let removed = commit_expiry(&mut store, &transaction)?; + assert!(!removed.removals[0].effective_trust_removed); + assert_eq!(removed.affected[0].services.len(), 2); + Ok(()) + } + + #[test] + fn restart_expires_elapsed_records_and_clock_rollback_before_admission() -> Result<()> { + for rollback in [false, true] { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + commit(&mut store, &tx("owner", "install", 0, vec![ + add(1, "one"), grants(1, &["echo"]), + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ]))?; + if rollback { + store.stored.expiry_last_seen = u64::try_from(deadline)?; + } else { + store.stored.records[0].expires_at = Some(i64::try_from(now())? - 1); + } + let expiry_request_id = store.stored.records[0].expiry_request_id.clone().unwrap(); + store.persist(&store.stored)?; + let (restarted, book) = RuntimePeers::open(&FabricHome::new(dir.path())); + assert!(restarted.fault().is_none(), "{:?}", restarted.fault()); + assert!(book.peers().is_empty()); + assert_eq!(restarted.revision(), 2); + let receipt = restarted.receipt(&expiry_request_id)?; + assert_eq!(receipt.cancellation.state, CancellationState::Complete); + assert!(receipt.removals[0].effective_trust_removed); + } + Ok(()) + } + + #[test] + fn expiry_wire_rejects_overflow_and_reserves_its_receipt_identifier() -> Result<()> { + let encoded = format!( + r#"{{"operation":"set_expiry","owner":"owner","request_id":"set","expected_revision":0,"node_id":"{}","expires_at":9223372036854775808}}"#, + id(1), + ); + assert!(serde_json::from_str::(&encoded).is_err()); + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + for value in [-1, 0, i64::try_from(store.expiry_timestamp())?, i64::MAX] { + assert!(store.prepare(&tx("owner", &format!("invalid-{value}"), 0, vec![ + add(1, "one"), Operation::SetExpiry { node_id: id(1), expires_at: value }, + ])).is_err()); + } + let receipt = commit(&mut store, &tx("owner", "install", 0, vec![ + add(1, "one"), Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ]))?; + let reserved = store.stored.records[0].expiry_request_id.clone().unwrap(); + assert!(matches!(&receipt.affected[0].trust_sources[0], + Source::Runtime { expires_at: Some(value), expiry_request_id: Some(request_id), .. } + if *value == deadline && request_id == &reserved)); + assert!(store.prepare(&tx("owner", &reserved, 1, vec![grants(1, &[])])).is_err()); + let raw = fs::read_to_string(&store.path)?; + assert!(raw.contains("expires_at")); + let restarted = open(&dir); + assert_eq!(restarted.stored.records[0].expires_at, Some(deadline)); + assert_eq!(restarted.stored.records[0].expiry_request_id.as_deref(), Some(reserved.as_str())); + Ok(()) + } + + #[test] + fn expiry_request_id_prefix_is_system_only_before_and_after_cleanup() -> Result<()> { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + let reserved = RuntimePeers::expiry_request_id("owner", id(1), deadline, "schedule"); + for request_id in ["expiry:", "expiry:arbitrary", reserved.as_str()] { + let error = store.prepare(&tx("other", request_id, 0, vec![add(2, "two")])) + .err().context("external expiry prefix was accepted")?; + assert_eq!(crate::error::code_of(&error), ErrorCode::InvalidArgument); + } + commit(&mut store, &tx("owner", "schedule", 0, vec![ + add(1, "one"), + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ]))?; + store.advance_expiry_for_test(u64::try_from(deadline)?); + let expiry = store.expired_transactions()?.remove(0); + assert_eq!(expiry.request_id, reserved); + assert!(store.prepare(&expiry).is_err()); + let receipt = commit_expiry(&mut store, &expiry)?; + assert_eq!(receipt.committed_revision, 2); + assert!(store.prepare(&expiry).is_err()); + assert_eq!(commit_expiry(&mut store, &expiry)?.committed_revision, 2); + assert!(store.stored.records.is_empty()); + Ok(()) + } + + #[test] + fn expiry_reservation_collision_rejects_the_entire_batch_and_remains_restartable() -> Result<()> { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + let reserved = RuntimePeers::expiry_request_id("owner", id(1), deadline, "schedule"); + commit(&mut store, &tx("owner", "install", 0, vec![ + add(1, "one"), grants(1, &["echo"]), + ]))?; + // Retained state from before the prefix reservation remains collision-safe. + store.stored.receipts[0].request_id = reserved; + store.persist(&store.stored)?; + let before = fs::read(&store.path)?; + let error = store.prepare(&tx("owner", "schedule", 1, vec![ + add(2, "next"), grants(2, &["echo"]), + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ])).err().context("colliding expiry reservation was accepted")?; + assert_eq!(crate::error::code_of(&error), ErrorCode::RequestIdConflict); + assert_eq!(store.revision(), 1); + assert_eq!(store.stored.records.len(), 1); + assert!(store.stored.records[0].expires_at.is_none()); + assert_eq!(fs::read(&store.path)?, before); + let restarted = open(&dir); + assert_eq!(restarted.revision(), 1); + assert!(restarted.stored.records[0].expires_at.is_none()); + Ok(()) + } + + #[test] + fn expiry_forward_step_scheduling_survives_rollback_without_extending_lifetime() -> Result<()> { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + store.advance_expiry_for_test(now() + 3_600); + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + commit(&mut store, &tx("owner", "schedule", 0, vec![ + add(1, "one"), grants(1, &["echo"]), + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ]))?; + let book = store.compose(&store.stored)?.0; + assert!(book.may(&id(1), "echo").is_ok()); + // The real wall clock remains behind the injected forward observation. + store.elapse_expiry_for_test(std::time::Duration::from_secs(600)); + assert!(book.may(&id(1), "echo").is_err()); + let expired = store.expired_transactions()?.remove(0); + let receipt = commit_expiry(&mut store, &expired)?; + assert!(receipt.removals[0].effective_trust_removed); + assert_eq!(receipt.cancellation.state, CancellationState::Complete); + Ok(()) + } + + #[test] + fn expiry_receipt_capacity_accounts_for_reservations_and_preserves_withdrawal_reserve() -> Result<()> { + let dir = tempfile::tempdir()?; + let mut store = open(&dir); + let deadline = i64::try_from(store.expiry_timestamp() + 600)?; + commit(&mut store, &tx("owner", "install", 0, vec![ + add(1, "one"), add(2, "two"), add(3, "three"), + Operation::SetExpiry { node_id: id(1), expires_at: deadline }, + ]))?; + let template = store.stored.receipts[0].clone(); + store.stored.receipts = (0..MAX_RECEIPTS - 1).map(|index| Receipt { + request_id: format!("receipt-{index}"), ..template.clone() + }).collect(); + assert!(store.stored.receipts.len() < MAX_RECEIPTS); + assert_eq!(store.stored.receipts.len() + 1, MAX_RECEIPTS); + let error = store.prepare(&tx("owner", "remove", 1, vec![ + Operation::RemovePeer { node_id: id(2) }, + ])).err().context("withdrawal entered the full receipt pool")?; + assert_eq!(crate::error::code_of(&error), ErrorCode::RuntimeReceiptCapacity); + store.stored.receipts.truncate(MAX_RECEIPTS - WITHDRAWAL_RECEIPTS - 2); + let error = store.prepare(&tx("owner", "schedule", 1, vec![ + Operation::SetExpiry { node_id: id(2), expires_at: deadline }, + ])).err().context("expiry scheduling entered the withdrawal reserve")?; + assert_eq!(crate::error::code_of(&error), ErrorCode::RuntimeReceiptCapacity); + commit(&mut store, &tx("owner", "remove", 1, vec![ + Operation::RemovePeer { node_id: id(3) }, + ]))?; + Ok(()) + } } diff --git a/src/runtime_peers/pairing/confirm.rs b/src/runtime_peers/pairing/confirm.rs index 64e5a91..2324686 100644 --- a/src/runtime_peers/pairing/confirm.rs +++ b/src/runtime_peers/pairing/confirm.rs @@ -206,6 +206,8 @@ impl RuntimePeers { next.records.push(Record { owner: request.owner.clone(), peer, + expires_at: None, + expiry_request_id: None, }); next.records .sort_by(|a, b| (&a.owner, a.peer.id).cmp(&(&b.owner, b.peer.id))); @@ -1132,6 +1134,8 @@ mod tests { for index in 0..MAX_RECORDS { store.stored.records.push(Record { owner: format!("owner-{index}"), + expires_at: None, + expiry_request_id: None, peer: Peer { id: node(2), name: None, diff --git a/src/tunnel.rs b/src/tunnel.rs index e101838..b6eb0a8 100644 --- a/src/tunnel.rs +++ b/src/tunnel.rs @@ -2111,6 +2111,9 @@ pub(crate) struct SessionRevocation { closed: CancellationToken, } impl SessionRevocation { + pub fn peer(&self) -> EndpointId { + self.session.peer_id + } pub async fn complete(&self) -> bool { self.closed.is_cancelled() && self.session.local_reader_done.is_cancelled()