From 06150818c8393b10bcf0758edde6e2c9acb0c7a0 Mon Sep 17 00:00:00 2001 From: "developing-today[bot]" <5454422+developing-today[bot]@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:51:22 -0600 Subject: [PATCH 1/2] feat(id): journaled native-spill write lease entries + epoch fencing (#5527) Phase 2 of the design #2819 protocol (packet issue #3804), the first implementation phase of issue #5527: LeaseGranted/LeaseRenewed/LeaseHandedOff/LeaseRevoked/LeaseWriteRejected journal entries, the WriteLease state machine (DO bootstrap epoch 0, epoch advances exactly +1 on grant/handoff/revoke, renewal keeps epoch and never moves the term backwards), the append-gate check (holder + epoch + term, expiry fences before any revocation), validated + deduplicated rejection audits, checkpoint-carried lease state/history/audit validated by replay-from-bootstrap at restore, and journal version 7 under the #227 bump-on-first-write contract. Native spill stays disabled; no new writer is enabled. Phase 3 (issue #3805) remains. --- pkgs/id/src/world/core.rs | 89 +++++++ pkgs/id/src/world/mod.rs | 524 ++++++++++++++++++++++++++++++++++++- pkgs/id/src/world/tests.rs | 404 ++++++++++++++++++++++++++++ pkgs/id/src/world_store.rs | 47 ++++ 4 files changed, 1062 insertions(+), 2 deletions(-) diff --git a/pkgs/id/src/world/core.rs b/pkgs/id/src/world/core.rs index 4695edc9d..25ab34fc0 100644 --- a/pkgs/id/src/world/core.rs +++ b/pkgs/id/src/world/core.rs @@ -188,6 +188,12 @@ pub struct WorldCore { /// rebuilt by journal replay and preserved across compaction by the /// checkpoint's `transfers` field. pub(crate) handoff_transfers: Vec, + /// The journal-resident native-spill write lease (design #2819, + /// phase 2 of issue #3804): rebuilt by journal replay from the + /// lease control entries and preserved across compaction by the + /// checkpoint's `lease` field. Bootstrap (DO holder, epoch 0) for + /// every world that has never spilled. + pub(crate) write_lease: WriteLease, /// Author keys bound to participants for signed move capture (#2814): /// participant id to the player's ed25519 verifying key bytes. A bound /// participant's inputs must arrive player-signed (see @@ -341,6 +347,7 @@ impl WorldCore { directory: crate::directory::Directory::new(), petnames: crate::petname::Petnames::new(), handoff_transfers: Vec::new(), + write_lease: WriteLease::bootstrap(), author_keys: HashMap::new(), move_chains: HashMap::new(), signed_moves: Vec::new(), @@ -811,6 +818,61 @@ impl WorldCore { replay.push(destination_event); core.since_checkpoint = core.since_checkpoint.saturating_add(1); } + JournalEntry::LeaseGranted { + holder, + epoch, + expires_at_ms, + } => { + core.write_lease + .apply_grant(&holder, epoch, expires_at_ms)?; + core.since_checkpoint = core.since_checkpoint.saturating_add(1); + } + JournalEntry::LeaseRenewed { + holder, + epoch, + expires_at_ms, + } => { + core.write_lease + .apply_renew(&holder, epoch, expires_at_ms)?; + core.since_checkpoint = core.since_checkpoint.saturating_add(1); + } + JournalEntry::LeaseHandedOff { + from, + to, + epoch, + expires_at_ms, + } => { + core.write_lease + .apply_handoff(&from, &to, epoch, expires_at_ms)?; + core.since_checkpoint = core.since_checkpoint.saturating_add(1); + } + JournalEntry::LeaseRevoked { + former_holder, + epoch, + reason, + } => { + core.write_lease + .apply_revoke(&former_holder, epoch, &reason)?; + core.since_checkpoint = core.since_checkpoint.saturating_add(1); + } + JournalEntry::LeaseWriteRejected { + writer, + presented_epoch, + operation_id, + observed_epoch, + reason, + received_at_ms, + } => { + core.write_lease.apply_rejection(LeaseWriteRejection { + writer, + presented_epoch, + operation_id, + observed_epoch, + reason, + received_at_ms, + })?; + core.since_checkpoint = core.since_checkpoint.saturating_add(1); + } JournalEntry::CheckpointSnapshotPart { sequence, part_index, @@ -913,6 +975,7 @@ impl WorldCore { caps_ceiling, schedules, transfers, + lease, mut snapshot, mut events, mut admin_history, @@ -1056,6 +1119,24 @@ impl WorldCore { validated_transfers.push(transfer); } core.handoff_transfers = validated_transfers; + // A checkpoint stands in for the journal it + // compacted, so its lease field is the whole lease + // state: validate the carried history reproduces + // the claimed state before adopting it, exactly as + // replaying the un-compacted lease entries would. + // Lease entries are never written before their + // checkpoint, so the lease here must still be the + // bootstrap state the checkpoint replaces. + if let Some(lease) = lease { + ensure!( + core.write_lease.is_bootstrap(), + "journal checkpoint carries a write lease after lease entries" + ); + lease + .validate_history() + .context("invalid write lease in journal checkpoint")?; + core.write_lease = lease; + } // The writer emits one `ScheduleSet` per pending schedule // before the checkpoint, so the two must agree exactly. let listed = schedules.unwrap_or_default(); @@ -1234,6 +1315,7 @@ impl WorldCore { caps_ceiling: caps_ceiling.clone(), schedules: schedules.clone(), transfers: transfers.clone(), + lease: (!self.write_lease.is_bootstrap()).then(|| self.write_lease.clone()), snapshot, events, admin_history: Some(JournalAdminHistory { @@ -1416,6 +1498,13 @@ impl WorldCore { &self.handoff_transfers } + /// The world's journal-resident native-spill write lease + /// (design #2819, phase 2 of issue #3804). + #[must_use] + pub(crate) fn write_lease(&self) -> &WriteLease { + &self.write_lease + } + /// The per-world resource requests, independent of current server ceilings. /// The world's total-memory budget ledger (#557): the effective /// limit, the logical charge measured at restore, its high-water diff --git a/pkgs/id/src/world/mod.rs b/pkgs/id/src/world/mod.rs index e75f8590f..bd82616c7 100644 --- a/pkgs/id/src/world/mod.rs +++ b/pkgs/id/src/world/mod.rs @@ -601,6 +601,431 @@ pub struct HandoffTransferState { pub event: WorldEvent, } +/// The lease holder every world starts with: the hosted actor itself +/// (design #2819 bootstrap — existing worlds with no lease records are +/// DO-owned at epoch 0). +pub const DO_WRITE_LEASE_HOLDER: &str = "do"; + +/// Recommended native lease term (design #2819): 30 seconds, renewed by +/// the DO every 10 seconds, judged only on the DO authority's clock. +pub const WRITE_LEASE_TERM_MS: u64 = 30_000; + +/// Recommended interval between DO renewals of a native lease +/// (design #2819). +pub const WRITE_LEASE_RENEWAL_INTERVAL_MS: u64 = 10_000; + +/// Why the append gate refused a writer (design #2819, phase 2 of +/// issue #3804). A refusal is retryable in every case: the writer catches +/// up, waits for a grant, or retries under the new holder. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LeaseRejectionReason { + /// The writer is not the current lease holder (or there is none). + NotHolder, + /// The writer presented an epoch other than the current one. + StaleEpoch, + /// The current lease term had expired on the authority's clock. + Expired, +} + +/// Which lease transition a [`WriteLeaseTransition`] records +/// (design #2819). +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LeaseTransitionKind { + /// A lease was granted while no holder stood (bootstrap is not a + /// grant; the first grant follows a revocation). + Granted, + /// The current holder's term was extended; the epoch is unchanged. + Renewed, + /// The lease was handed from one holder to another. + HandedOff, + /// The lease was revoked; no holder stands until the next grant. + Revoked, +} + +/// One auditable lease transition, in journal order (design #2819: +/// compaction must retain every transition needed to prove lease +/// ownership at represented positions). `epoch` is the epoch in force +/// after the transition (for a renewal, the unchanged epoch). +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WriteLeaseTransition { + /// The kind of transition. + pub kind: LeaseTransitionKind, + /// The holder before the transition (`None` when none stood). + pub from: Option, + /// The holder after the transition (`None` after a revocation). + pub to: Option, + /// The epoch in force after the transition. + pub epoch: u64, + /// The term expiry in force after the transition, in unix + /// milliseconds (`None` for an untermed holder such as the DO). + pub expires_at_ms: Option, + /// Why a revocation happened; absent for other kinds. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason: Option, +} + +/// One rejected stale append attempt, as the append authority audited +/// it (design #2819). Deduplicated by `(writer, presented_epoch, +/// operation_id)`: repeated deliveries of the same attempt produce one +/// audit record, while a client retry of the same operation under a new +/// holder is a distinct attempt. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LeaseWriteRejection { + /// The authenticated writer identity that was refused. + pub writer: String, + /// The epoch the writer presented. + pub presented_epoch: u64, + /// The client operation ID the writer presented. + pub operation_id: String, + /// The authority's epoch when the attempt was refused. + pub observed_epoch: u64, + /// Why the attempt was refused. + pub reason: LeaseRejectionReason, + /// When the authority received the attempt, in unix milliseconds + /// on the authority's clock. + pub received_at_ms: u64, +} + +/// The journal-resident write lease for one world (design #2819, +/// phase 2 of issue #3804): the derived state the append gate fences +/// on, rebuilt by journal replay and carried whole in checkpoints so +/// compaction preserves the current state, the monotonic epoch, the +/// transition history, and the rejection audit. +/// +/// Safety invariants: epochs never decrease or repeat across grant, +/// handoff, and revocation (a renewal keeps the epoch); there is at +/// most one holder at any epoch; expiry fences the old holder even +/// before a revocation is journaled, because [`WriteLease::check_writer`] +/// judges the term at the append decision. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WriteLease { + /// The current holder's authenticated identity; `None` between a + /// revocation and the next grant. + pub holder: Option, + /// The current epoch. Bootstrap is epoch 0. + pub epoch: u64, + /// When the current term expires, in unix milliseconds on the + /// authority's clock; `None` for an untermed holder. + pub expires_at_ms: Option, + /// Every transition applied, in journal order. + pub transitions: Vec, + /// Every audited rejected append attempt, in journal order. + pub rejections: Vec, +} + +impl Default for WriteLease { + fn default() -> Self { + Self::bootstrap() + } +} + +impl WriteLease { + /// The pre-spill state of every existing world: the DO actor is the + /// incumbent holder at epoch 0, with no term and no history. + #[must_use] + pub fn bootstrap() -> Self { + Self { + holder: Some(DO_WRITE_LEASE_HOLDER.to_owned()), + epoch: 0, + expires_at_ms: None, + transitions: Vec::new(), + rejections: Vec::new(), + } + } + + /// Whether this lease is still the untouched bootstrap state (a + /// checkpoint then omits it, exactly as older checkpoints did). + #[must_use] + pub fn is_bootstrap(&self) -> bool { + *self == Self::bootstrap() + } + + /// Whether `now_ms` is past the current term. An untermed holder + /// never expires; a world with no holder has nothing to expire. + #[must_use] + pub fn is_expired(&self, now_ms: u64) -> bool { + self.holder.is_some() && self.expires_at_ms.is_some_and(|expiry| now_ms >= expiry) + } + + /// The append-gate check (design #2819): a mutation append is + /// admitted only from the current holder, presenting the current + /// epoch, before the term expires on the authority's clock. This + /// is the check the serialized append path performs in the same + /// operation as the append itself; a caller that passes it and + /// then appends through any other path has no fencing. + /// + /// # Errors + /// + /// The [`LeaseRejectionReason`] the gate records for the attempt. + pub fn check_writer( + &self, + writer: &str, + presented_epoch: u64, + now_ms: u64, + ) -> Result<(), LeaseRejectionReason> { + if self.holder.as_deref() != Some(writer) { + return Err(LeaseRejectionReason::NotHolder); + } + if presented_epoch != self.epoch { + return Err(LeaseRejectionReason::StaleEpoch); + } + if self.is_expired(now_ms) { + return Err(LeaseRejectionReason::Expired); + } + Ok(()) + } + + /// Apply a `LeaseGranted` record: legal only while no holder + /// stands, and the granted epoch must be exactly the next epoch. + /// + /// # Errors + /// + /// A grant over a standing holder (that is a handoff, not a grant), + /// or an epoch that is not `current + 1`. + pub fn apply_grant( + &mut self, + holder: &str, + epoch: u64, + expires_at_ms: Option, + ) -> Result<()> { + ensure!( + self.holder.is_none(), + "journal grants the write lease to {holder:?} while {:?} holds it", + self.holder + ); + self.advance_epoch(epoch, "grant")?; + self.holder = Some(holder.to_owned()); + self.expires_at_ms = expires_at_ms; + self.transitions.push(WriteLeaseTransition { + kind: LeaseTransitionKind::Granted, + from: None, + to: Some(holder.to_owned()), + epoch, + expires_at_ms, + reason: None, + }); + Ok(()) + } + + /// Apply a `LeaseRenewed` record: only the current holder renews, + /// the epoch is unchanged, and the term never moves backwards. + /// + /// # Errors + /// + /// A renewal by a non-holder, under a different epoch, or moving + /// the expiry backwards. + pub fn apply_renew(&mut self, holder: &str, epoch: u64, expires_at_ms: u64) -> Result<()> { + ensure!( + self.holder.as_deref() == Some(holder), + "journal renews the write lease for {holder:?}, held by {:?}", + self.holder + ); + ensure!( + epoch == self.epoch, + "journal renews the write lease at epoch {epoch}, current epoch is {}", + self.epoch + ); + ensure!( + self.expires_at_ms + .is_none_or(|current| expires_at_ms >= current), + "journal renewal moves the write lease expiry backwards" + ); + self.expires_at_ms = Some(expires_at_ms); + self.transitions.push(WriteLeaseTransition { + kind: LeaseTransitionKind::Renewed, + from: Some(holder.to_owned()), + to: Some(holder.to_owned()), + epoch, + expires_at_ms: Some(expires_at_ms), + reason: None, + }); + Ok(()) + } + + /// Apply a `LeaseHandedOff` record: the handoff names the current + /// holder as its source and advances the epoch by exactly one; + /// that journal position is the ownership boundary. + /// + /// # Errors + /// + /// A handoff from a non-holder, or an epoch that is not + /// `current + 1`. + pub fn apply_handoff( + &mut self, + from: &str, + to: &str, + epoch: u64, + expires_at_ms: Option, + ) -> Result<()> { + ensure!( + self.holder.as_deref() == Some(from), + "journal hands the write lease off from {from:?}, held by {:?}", + self.holder + ); + self.advance_epoch(epoch, "handoff")?; + self.holder = Some(to.to_owned()); + self.expires_at_ms = expires_at_ms; + self.transitions.push(WriteLeaseTransition { + kind: LeaseTransitionKind::HandedOff, + from: Some(from.to_owned()), + to: Some(to.to_owned()), + epoch, + expires_at_ms, + reason: None, + }); + Ok(()) + } + + /// Apply a `LeaseRevoked` record: revocation names the current + /// holder, advances the epoch by exactly one, and leaves no + /// holder standing. + /// + /// # Errors + /// + /// A revocation of a non-holder, or an epoch that is not + /// `current + 1`. + pub fn apply_revoke(&mut self, former_holder: &str, epoch: u64, reason: &str) -> Result<()> { + ensure!( + self.holder.as_deref() == Some(former_holder), + "journal revokes the write lease from {former_holder:?}, held by {:?}", + self.holder + ); + self.advance_epoch(epoch, "revocation")?; + self.holder = None; + self.expires_at_ms = None; + self.transitions.push(WriteLeaseTransition { + kind: LeaseTransitionKind::Revoked, + from: Some(former_holder.to_owned()), + to: None, + epoch, + expires_at_ms: None, + reason: Some(reason.to_owned()), + }); + Ok(()) + } + + /// Apply a `LeaseWriteRejected` audit record. The audited attempt + /// must actually be refusable under the state it names: a writer + /// that would have passed the gate cannot be audited as rejected. + /// Duplicate deliveries of one attempt (same writer, presented + /// epoch, and operation ID) are refused, so a journal carries at + /// most one audit record per attempt. + /// + /// # Errors + /// + /// An audit whose observed epoch is not the current epoch, whose + /// stated reason disagrees with the gate check (expiry is judged + /// at `received_at_ms`, the authority's own clock reading), or + /// that duplicates an earlier audit. + pub fn apply_rejection(&mut self, rejection: LeaseWriteRejection) -> Result<()> { + ensure!( + rejection.observed_epoch == self.epoch, + "journal audits a rejected write at observed epoch {}, current epoch is {}", + rejection.observed_epoch, + self.epoch + ); + ensure!( + self.check_writer( + &rejection.writer, + rejection.presented_epoch, + rejection.received_at_ms + ) == Err(rejection.reason), + "journal audits a rejected write by {:?} whose stated reason {:?} disagrees with the lease gate", + rejection.writer, + rejection.reason + ); + ensure!( + !self.rejections.iter().any(|prior| { + prior.writer == rejection.writer + && prior.presented_epoch == rejection.presented_epoch + && prior.operation_id == rejection.operation_id + }), + "journal audits rejected write {:?} twice", + rejection.operation_id + ); + self.rejections.push(rejection); + Ok(()) + } + + /// Validate a checkpoint-carried lease as a whole: replaying its + /// transitions and rejections from bootstrap must reproduce the + /// exact state the checkpoint claims. Used at restore, where a + /// checkpoint stands in for the journal it compacted. + /// + /// # Errors + /// + /// Any transition or rejection that replay refuses, or a final + /// state that differs from the claimed one. + pub fn validate_history(&self) -> Result<()> { + let mut replayed = Self::bootstrap(); + for transition in &self.transitions { + match transition.kind { + LeaseTransitionKind::Granted => replayed.apply_grant( + transition + .to + .as_deref() + .context("lease grant transition names no holder")?, + transition.epoch, + transition.expires_at_ms, + )?, + LeaseTransitionKind::Renewed => replayed.apply_renew( + transition + .to + .as_deref() + .context("lease renewal transition names no holder")?, + transition.epoch, + transition + .expires_at_ms + .context("lease renewal transition names no expiry")?, + )?, + LeaseTransitionKind::HandedOff => replayed.apply_handoff( + transition + .from + .as_deref() + .context("lease handoff transition names no source")?, + transition + .to + .as_deref() + .context("lease handoff transition names no target")?, + transition.epoch, + transition.expires_at_ms, + )?, + LeaseTransitionKind::Revoked => replayed.apply_revoke( + transition + .from + .as_deref() + .context("lease revocation transition names no holder")?, + transition.epoch, + transition + .reason + .as_deref() + .context("lease revocation transition names no reason")?, + )?, + } + } + for rejection in &self.rejections { + replayed.apply_rejection(rejection.clone())?; + } + ensure!( + replayed == *self, + "checkpoint write lease history does not reproduce its claimed state" + ); + Ok(()) + } + + fn advance_epoch(&mut self, epoch: u64, what: &str) -> Result<()> { + ensure!( + self.epoch.checked_add(1) == Some(epoch), + "journal {what} advances the write lease epoch from {} to {epoch}", + self.epoch + ); + self.epoch = epoch; + Ok(()) + } +} + /// One durable fact in the region coordinator's own journal (#3870 phase 1, /// design #2818 §3). The coordinator records and recovers handoff progress; /// it never stores entity payloads and never emulates a cross-region @@ -915,6 +1340,76 @@ pub enum JournalEntry { /// the payload, hex-encoded, per design #2818 §3). destination_event: WorldEvent, }, + /// A write lease was granted while no holder stood (design #2819, + /// phase 2 of issue #3804). Authority-only: the append authority + /// writes this, never a lease holder on its own behalf. `epoch` is + /// the resulting epoch (`previous + 1`). + LeaseGranted { + /// The new holder's authenticated identity. + holder: String, + /// The resulting lease epoch. + epoch: u64, + /// When the granted term expires, in unix milliseconds on the + /// authority's clock; absent for an untermed holder. + #[serde(default, skip_serializing_if = "Option::is_none")] + expires_at_ms: Option, + }, + /// The current holder's lease term was renewed by the authority + /// (design #2819). The epoch is unchanged. + LeaseRenewed { + /// The holder whose term was renewed. + holder: String, + /// The (unchanged) lease epoch. + epoch: u64, + /// The new term expiry, in unix milliseconds. + expires_at_ms: u64, + }, + /// The write lease was handed from one holder to another + /// (design #2819). This entry's journal position is the ownership + /// boundary: before it only the old epoch passes the append gate, + /// after it only the new one. `epoch` is the resulting epoch. + LeaseHandedOff { + /// The holder giving up execution. + from: String, + /// The holder taking over execution. + to: String, + /// The resulting lease epoch. + epoch: u64, + /// When the new holder's term expires, in unix milliseconds; + /// absent for an untermed holder. + #[serde(default, skip_serializing_if = "Option::is_none")] + expires_at_ms: Option, + }, + /// The write lease was revoked by the authority (design #2819): + /// the former holder is fenced at the resulting epoch and no + /// holder stands until the next grant. + LeaseRevoked { + /// The holder whose lease was revoked. + former_holder: String, + /// The resulting lease epoch. + epoch: u64, + /// Why the lease was revoked. + reason: String, + }, + /// An authenticated append attempt was refused by the append gate + /// and the authority journaled the attempt as an audit record + /// (design #2819). The attempted entry itself was not applied. + /// Control metadata: replay rebuilds the audit, never the program. + LeaseWriteRejected { + /// The authenticated writer identity that was refused. + writer: String, + /// The epoch the writer presented. + presented_epoch: u64, + /// The client operation ID the writer presented. + operation_id: String, + /// The authority's epoch when the attempt was refused. + observed_epoch: u64, + /// Why the attempt was refused. + reason: LeaseRejectionReason, + /// When the authority received the attempt, in unix + /// milliseconds on the authority's clock. + received_at_ms: u64, + }, /// One UTF-8-safe part of a compacted program snapshot. The complete set /// immediately precedes its checkpoint and is keyed by that checkpoint's /// sequence. @@ -980,6 +1475,15 @@ pub enum JournalEntry { /// checkpoints written before journal version 3. #[serde(default, skip_serializing_if = "Option::is_none")] transfers: Option>, + /// The journal-resident write lease as of this checkpoint + /// (design #2819, phase 2 of issue #3804): current holder, + /// epoch and term, the full transition history, and the + /// rejection audit, so compaction cannot reset the epoch or + /// lose proof of ownership. Absent when the lease is still the + /// bootstrap state, and in checkpoints written before journal + /// version 7. + #[serde(default, skip_serializing_if = "Option::is_none")] + lease: Option, /// The program's own serialization of its state. snapshot: String, /// Recent events kept for reconnect catch-up. @@ -1056,10 +1560,18 @@ pub const JOURNAL_VERSION_OPEN_PRESENCE: u32 = 5; /// restoring an incomplete checkpoint. pub const JOURNAL_VERSION_CHECKPOINT_PARTS: u32 = 6; +/// Header version of a journal that contains a native-spill write +/// lease entry ([`JournalEntry::LeaseGranted`] and its siblings), or a +/// checkpoint carrying a non-bootstrap [`WriteLease`] (design #2819, +/// phase 2 of issue #3804). Version-6 binaries cannot read the lease +/// records and would restore a world with no fencing state, so they +/// must refuse the journal at the header check. +pub const JOURNAL_VERSION_WRITE_LEASE: u32 = 7; + /// Newest journal version this binary can read. Journals with a higher /// header version are refused at the header check, before any entry is /// applied. -pub const MAX_JOURNAL_VERSION: u32 = JOURNAL_VERSION_CHECKPOINT_PARTS; +pub const MAX_JOURNAL_VERSION: u32 = JOURNAL_VERSION_WRITE_LEASE; /// The header version a journal must carry once it contains `entry`: version 2 /// for execution-profile entries, version 3 for region-handoff entries, @@ -1082,6 +1594,11 @@ pub fn journal_version_required_by(entry: &JournalEntry) -> u32 { JournalEntry::EntityExported { .. } | JournalEntry::EntityImported { .. } => { JOURNAL_VERSION_REGION_HANDOFF } + JournalEntry::LeaseGranted { .. } + | JournalEntry::LeaseRenewed { .. } + | JournalEntry::LeaseHandedOff { .. } + | JournalEntry::LeaseRevoked { .. } + | JournalEntry::LeaseWriteRejected { .. } => JOURNAL_VERSION_WRITE_LEASE, JournalEntry::CheckpointSnapshotPart { .. } | JournalEntry::CheckpointEventsPart { .. } | JournalEntry::CheckpointAdminPart { .. } => JOURNAL_VERSION_CHECKPOINT_PARTS, @@ -1096,12 +1613,15 @@ pub fn journal_version_required_by(entry: &JournalEntry) -> u32 { // PR #3922). JournalEntry::Checkpoint { transfers, + lease, events, open_presence, execution_profile, .. } => { - let base = if open_presence.is_some() { + let base = if lease.is_some() { + JOURNAL_VERSION_WRITE_LEASE + } else if open_presence.is_some() { JOURNAL_VERSION_OPEN_PRESENCE } else if transfers.is_some() || events diff --git a/pkgs/id/src/world/tests.rs b/pkgs/id/src/world/tests.rs index 4c529d98a..9ed610758 100644 --- a/pkgs/id/src/world/tests.rs +++ b/pkgs/id/src/world/tests.rs @@ -149,6 +149,7 @@ mod tests { caps_ceiling: None, schedules: None, transfers: None, + lease: None, snapshot: "s".to_owned(), events, open_presence: None, @@ -314,6 +315,7 @@ mod tests { caps_ceiling: None, schedules: None, transfers: Some(transfers), + lease: None, snapshot: "state".to_owned(), events: Vec::new(), open_presence: None, @@ -379,6 +381,7 @@ mod tests { caps_ceiling: None, schedules: None, transfers: None, + lease: None, admin_history: None, open_presence: None, snapshot: "s".to_owned(), @@ -460,6 +463,404 @@ mod tests { assert_eq!(restored.core.handoff_transfers(), transfers.as_slice()); } + fn lease_header() -> JournalEntry { + JournalEntry::Created { + world_id: "w".to_owned(), + version: JOURNAL_VERSION_WRITE_LEASE, + } + } + + fn spill_handoff() -> JournalEntry { + JournalEntry::LeaseHandedOff { + from: DO_WRITE_LEASE_HOLDER.to_owned(), + to: "native-1".to_owned(), + epoch: 1, + expires_at_ms: Some(30_000), + } + } + + fn spill_renewal() -> JournalEntry { + JournalEntry::LeaseRenewed { + holder: "native-1".to_owned(), + epoch: 1, + expires_at_ms: 40_000, + } + } + + fn stale_rejection() -> JournalEntry { + // The native holder presenting its pre-handoff epoch: the gate + // refuses it as stale, and the authority journals the attempt. + JournalEntry::LeaseWriteRejected { + writer: "native-1".to_owned(), + presented_epoch: 0, + operation_id: "op-1".to_owned(), + observed_epoch: 1, + reason: LeaseRejectionReason::StaleEpoch, + received_at_ms: 35_000, + } + } + + #[test] + fn write_lease_entries_require_journal_version_7() { + // Design #2819 phase 2 (#3804): lease kinds are the first + // entries a version-6 binary cannot read, so they pin the + // header to version 7 under the #227 bump-on-first-write + // contract. + for entry in [ + JournalEntry::LeaseGranted { + holder: "native-1".to_owned(), + epoch: 1, + expires_at_ms: Some(30_000), + }, + spill_renewal(), + spill_handoff(), + JournalEntry::LeaseRevoked { + former_holder: "native-1".to_owned(), + epoch: 2, + reason: "expired".to_owned(), + }, + stale_rejection(), + ] { + assert_eq!( + journal_version_required_by(&entry), + JOURNAL_VERSION_WRITE_LEASE, + "{entry:?}" + ); + } + assert_eq!(MAX_JOURNAL_VERSION, JOURNAL_VERSION_WRITE_LEASE); + // A checkpoint carrying a non-bootstrap lease also requires + // version 7; one without a lease keeps its existing version. + let mut lease = WriteLease::bootstrap(); + lease + .apply_handoff(DO_WRITE_LEASE_HOLDER, "native-1", 1, Some(30_000)) + .unwrap(); + let checkpoint = |lease: Option| JournalEntry::Checkpoint { + sequence: 0, + module_hash: "h".to_owned(), + seed: 1, + provenance: None, + execution_profile: None, + caps_ceiling: None, + schedules: None, + transfers: None, + lease, + snapshot: "s".to_owned(), + events: Vec::new(), + open_presence: None, + admin_history: None, + }; + assert_eq!( + journal_version_required_by(&checkpoint(Some(lease))), + JOURNAL_VERSION_WRITE_LEASE + ); + assert_eq!( + journal_version_required_by(&checkpoint(None)), + JOURNAL_VERSION + ); + } + + #[test] + fn write_lease_wire_shape_round_trips() { + for entry in [spill_handoff(), spill_renewal(), stale_rejection()] { + let json = serde_json::to_string(&entry).unwrap(); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + entry, + "{json}" + ); + } + // A lease entry in a journal whose header predates version 7 is + // refused at restore, never applied mid-read. + let error = WorldCore::restore( + "w", + WorldLimits::default(), + [ + JournalEntry::Created { + world_id: "w".to_owned(), + version: JOURNAL_VERSION_CHECKPOINT_PARTS, + }, + spill_handoff(), + ], + ) + .unwrap_err(); + assert!( + error.to_string().contains("requires version 7"), + "{error:#}" + ); + } + + #[test] + fn write_lease_replay_walks_grant_renew_revoke_handoff() { + // Bootstrap: an untouched world is DO-owned at epoch 0. + let restored = WorldCore::restore( + "w", + WorldLimits::default(), + [JournalEntry::Created { + world_id: "w".to_owned(), + version: JOURNAL_VERSION, + }], + ) + .unwrap(); + assert_eq!(restored.core.write_lease(), &WriteLease::bootstrap()); + assert_eq!( + restored + .core + .write_lease() + .check_writer(DO_WRITE_LEASE_HOLDER, 0, u64::MAX), + Ok(()) + ); + + let restored = WorldCore::restore( + "w", + WorldLimits::default(), + [ + lease_header(), + spill_handoff(), + spill_renewal(), + JournalEntry::LeaseRevoked { + former_holder: "native-1".to_owned(), + epoch: 2, + reason: "term expired without renewal".to_owned(), + }, + JournalEntry::LeaseGranted { + holder: DO_WRITE_LEASE_HOLDER.to_owned(), + epoch: 3, + expires_at_ms: None, + }, + ], + ) + .unwrap(); + let lease = restored.core.write_lease(); + assert_eq!(lease.holder.as_deref(), Some(DO_WRITE_LEASE_HOLDER)); + assert_eq!(lease.epoch, 3); + assert_eq!(lease.expires_at_ms, None); + assert_eq!(lease.transitions.len(), 4); + assert_eq!(lease.transitions[0].kind, LeaseTransitionKind::HandedOff); + assert_eq!(lease.transitions[1].kind, LeaseTransitionKind::Renewed); + assert_eq!(lease.transitions[2].kind, LeaseTransitionKind::Revoked); + assert_eq!(lease.transitions[3].kind, LeaseTransitionKind::Granted); + // The ex-holder is fenced at the new epoch, and the returned + // holder presenting an old epoch is stale. + assert_eq!( + lease.check_writer("native-1", 1, 0), + Err(LeaseRejectionReason::NotHolder) + ); + assert_eq!( + lease.check_writer(DO_WRITE_LEASE_HOLDER, 2, 0), + Err(LeaseRejectionReason::StaleEpoch) + ); + assert_eq!( + lease.check_writer(DO_WRITE_LEASE_HOLDER, 3, u64::MAX), + Ok(()) + ); + } + + #[test] + fn write_lease_gate_fences_an_expired_holder_before_any_revoke() { + let restored = WorldCore::restore( + "w", + WorldLimits::default(), + [lease_header(), spill_handoff(), spill_renewal()], + ) + .unwrap(); + let lease = restored.core.write_lease(); + assert_eq!(lease.check_writer("native-1", 1, 39_999), Ok(())); + // Expiry alone fences the holder: no revocation entry needs to + // land first, because the gate judges the term at append time. + assert_eq!( + lease.check_writer("native-1", 1, 40_000), + Err(LeaseRejectionReason::Expired) + ); + assert!(lease.is_expired(40_000)); + assert!(!lease.is_expired(39_999)); + } + + #[test] + fn write_lease_rejected_attempt_is_audited_and_never_applied() { + let restored = WorldCore::restore( + "w", + WorldLimits::default(), + [lease_header(), spill_handoff(), stale_rejection()], + ) + .unwrap(); + let lease = restored.core.write_lease(); + // The audit landed; the attempted write did not change the + // holder, the epoch, or any program state. + assert_eq!(lease.holder.as_deref(), Some("native-1")); + assert_eq!(lease.epoch, 1); + assert_eq!(lease.rejections.len(), 1); + assert_eq!(lease.rejections[0].operation_id, "op-1"); + assert_eq!(restored.core.current_sequence(), 0); + assert!(restored.replay.is_empty()); + + // A second audit of the same attempt is a duplicate. + let error = WorldCore::restore( + "w", + WorldLimits::default(), + [ + lease_header(), + spill_handoff(), + stale_rejection(), + stale_rejection(), + ], + ) + .unwrap_err(); + assert!(error.to_string().contains("twice"), "{error:#}"); + + // An audit whose stated reason disagrees with the gate is + // refused: at 35s the native term (40s after renewal) had not + // expired, so this attempt was stale, not expired. + let lying = JournalEntry::LeaseWriteRejected { + writer: "native-1".to_owned(), + presented_epoch: 0, + operation_id: "op-1".to_owned(), + observed_epoch: 1, + reason: LeaseRejectionReason::Expired, + received_at_ms: 35_000, + }; + let error = WorldCore::restore( + "w", + WorldLimits::default(), + [lease_header(), spill_handoff(), lying], + ) + .unwrap_err(); + assert!( + error.to_string().contains("disagrees with the lease gate"), + "{error:#}" + ); + } + + #[test] + fn write_lease_transitions_refuse_epoch_reuse_skips_and_non_holders() { + let cases: Vec<(Vec, &str)> = vec![ + ( + // A handoff that skips epoch 1. + vec![JournalEntry::LeaseHandedOff { + from: DO_WRITE_LEASE_HOLDER.to_owned(), + to: "native-1".to_owned(), + epoch: 2, + expires_at_ms: Some(30_000), + }], + "advances the write lease epoch from 0 to 2", + ), + ( + // A grant while the DO still holds the lease. + vec![JournalEntry::LeaseGranted { + holder: "native-1".to_owned(), + epoch: 1, + expires_at_ms: None, + }], + "while", + ), + ( + // A renewal by someone who does not hold the lease. + vec![ + spill_handoff(), + JournalEntry::LeaseRenewed { + holder: DO_WRITE_LEASE_HOLDER.to_owned(), + epoch: 1, + expires_at_ms: 50_000, + }, + ], + "held by", + ), + ( + // A renewal that moves the expiry backwards. + vec![ + spill_handoff(), + spill_renewal(), + JournalEntry::LeaseRenewed { + holder: "native-1".to_owned(), + epoch: 1, + expires_at_ms: 35_000, + }, + ], + "backwards", + ), + ( + // A revocation naming the wrong former holder. + vec![ + spill_handoff(), + JournalEntry::LeaseRevoked { + former_holder: DO_WRITE_LEASE_HOLDER.to_owned(), + epoch: 2, + reason: "operator".to_owned(), + }, + ], + "held by", + ), + ( + // A handoff from someone who does not hold the lease. + vec![JournalEntry::LeaseHandedOff { + from: "native-9".to_owned(), + to: "native-1".to_owned(), + epoch: 1, + expires_at_ms: None, + }], + "held by", + ), + ]; + for (entries, needle) in cases { + let mut journal = vec![lease_header()]; + journal.extend(entries); + let error = WorldCore::restore("w", WorldLimits::default(), journal).unwrap_err(); + assert!(error.to_string().contains(needle), "{error:#}"); + } + } + + #[test] + fn write_lease_survives_restore_and_checkpoint_compaction() { + let restored = WorldCore::restore( + "w", + WorldLimits::default(), + [ + lease_header(), + spill_handoff(), + spill_renewal(), + stale_rejection(), + ], + ) + .unwrap(); + let lease = restored.core.write_lease().clone(); + + // Compaction carries the lease whole on the checkpoint: the + // header rises to version 7 and a restore of the compacted + // journal rebuilds the identical lease — epoch, holder, term, + // transition history and rejection audit all preserved. + let mut core = restored.core; + core.active_module_hash = Some("h".to_owned()); + let compacted = core.checkpoint_entries("state".to_owned()).unwrap(); + assert!(matches!( + compacted[0], + JournalEntry::Created { + version: JOURNAL_VERSION_WRITE_LEASE, + .. + } + )); + let checkpoint_lease = compacted.iter().find_map(|entry| match entry { + JournalEntry::Checkpoint { lease, .. } => lease.clone(), + _ => None, + }); + assert_eq!(checkpoint_lease, Some(lease.clone())); + let restored = WorldCore::restore("w", WorldLimits::default(), compacted.clone()).unwrap(); + assert_eq!(restored.core.write_lease(), &lease); + + // A checkpoint whose carried history does not reproduce its + // claimed state (here: the epoch was bumped without a matching + // transition) is refused whole. + let mut tampered = compacted; + for entry in &mut tampered { + if let JournalEntry::Checkpoint { + lease: Some(lease), .. + } = entry + { + lease.epoch += 1; + } + } + let error = WorldCore::restore("w", WorldLimits::default(), tampered).unwrap_err(); + assert!(error.to_string().contains("write lease"), "{error:#}"); + } + #[test] fn checkpoint_handoff_transfers_validate_duplicates_epochs_and_events() { for (first, second) in [ @@ -778,6 +1179,7 @@ mod tests { caps_ceiling: None, schedules: None, transfers: None, + lease: None, snapshot: "s".to_owned(), events: vec![], open_presence: None, @@ -796,6 +1198,7 @@ mod tests { caps_ceiling: None, schedules: None, transfers: None, + lease: None, snapshot: "s".to_owned(), events: vec![], open_presence: None, @@ -6556,6 +6959,7 @@ mod schedule_tests { caps_ceiling: None, schedules, transfers: None, + lease: None, snapshot: "s".to_owned(), events: Vec::new(), admin_history: None, diff --git a/pkgs/id/src/world_store.rs b/pkgs/id/src/world_store.rs index e593227bb..e98c762ab 100644 --- a/pkgs/id/src/world_store.rs +++ b/pkgs/id/src/world_store.rs @@ -1191,6 +1191,52 @@ mod tests { ); } + #[test] + fn write_lease_entries_survive_jsonl_reopen_with_epoch_intact() { + // Design #2819 phase 2 (#3804): lease entries appended to a + // fresh (version 1) journal bump its header to version 7 on + // first write, and a reopen + replay rebuilds the same lease: + // holder, monotonic epoch, term, and transition history. + use crate::world::{ + JOURNAL_VERSION_WRITE_LEASE, JournalEntry as Entry, WorldCore, WorldJournal, + WorldLimits, + }; + let dir = TempDir::new().unwrap(); + let path = dir.path().join("journal.jsonl"); + let (mut journal, _) = FileJournal::open(&path, "w").unwrap(); + journal + .append(&Entry::LeaseHandedOff { + from: "do".to_owned(), + to: "native-1".to_owned(), + epoch: 1, + expires_at_ms: Some(30_000), + }) + .unwrap(); + journal + .append(&Entry::LeaseRenewed { + holder: "native-1".to_owned(), + epoch: 1, + expires_at_ms: 40_000, + }) + .unwrap(); + drop(journal); + + let (_journal, entries) = FileJournal::open(&path, "w").unwrap(); + assert!(matches!( + entries[0], + Entry::Created { + version: JOURNAL_VERSION_WRITE_LEASE, + .. + } + )); + let restored = WorldCore::restore("w", WorldLimits::default(), entries).unwrap(); + let lease = restored.core.write_lease(); + assert_eq!(lease.holder.as_deref(), Some("native-1")); + assert_eq!(lease.epoch, 1); + assert_eq!(lease.expires_at_ms, Some(40_000)); + assert_eq!(lease.transitions.len(), 2); + } + #[test] fn replication_snapshot_repairs_an_unterminated_torn_tail() { let dir = TempDir::new().unwrap(); @@ -2087,6 +2133,7 @@ mod tests { caps_ceiling: None, schedules: None, transfers: None, + lease: None, snapshot: "s".to_owned(), events: Vec::new(), admin_history: None, From 3cff3c3f093d5a15c01e6aeec3036f74001b6caf Mon Sep 17 00:00:00 2001 From: "developing-today[bot]" <5454422+developing-today[bot]@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:08:53 -0600 Subject: [PATCH 2/2] fix(id): interleave lease transitions and rejection audits in checkpoint history (Codex P1, #5527) WriteLease kept transitions and rejections in two separate lists, and validate_history replayed every transition before any audit: a stale write audited at epoch 1 followed by a revocation to epoch 2 restored fine from the original journal but failed checkpoint validation, bricking the world after compaction. Transitions and audits now share one journal-ordered history list, validation replays it in order, checkpoint_entries refuses to compact a lease history that would not validate, and a regression test walks handoff -> audit -> revoke -> grant through restore, validation, and compaction. --- pkgs/id/src/world/core.rs | 9 ++ pkgs/id/src/world/mod.rs | 209 ++++++++++++++++++++++--------------- pkgs/id/src/world/tests.rs | 64 ++++++++++-- pkgs/id/src/world_store.rs | 2 +- 4 files changed, 190 insertions(+), 94 deletions(-) diff --git a/pkgs/id/src/world/core.rs b/pkgs/id/src/world/core.rs index 25ab34fc0..4bb0d4321 100644 --- a/pkgs/id/src/world/core.rs +++ b/pkgs/id/src/world/core.rs @@ -1296,6 +1296,15 @@ impl WorldCore { // entity, forget an import receipt, or drop an owner epoch. let transfers: Option> = (!self.handoff_transfers.is_empty()).then(|| self.handoff_transfers.clone()); + // Never compact a lease history that would fail restore + // validation into the replacement journal: refuse the + // compaction and keep the un-compacted journal instead + // (Codex review on PR #5571). + if !self.write_lease.is_bootstrap() { + self.write_lease + .validate_history() + .context("write lease history does not validate; keep the journal untrimmed")?; + } let checkpoint_events: Vec = self.events.iter().cloned().collect(); let checkpoint_admin_events: Vec = self.admin_history.iter().cloned().collect(); // A checkpoint that already fits the per-entry bound stays a single diff --git a/pkgs/id/src/world/mod.rs b/pkgs/id/src/world/mod.rs index bd82616c7..8411a6041 100644 --- a/pkgs/id/src/world/mod.rs +++ b/pkgs/id/src/world/mod.rs @@ -688,6 +688,22 @@ pub struct LeaseWriteRejection { pub received_at_ms: u64, } +/// One entry in a [`WriteLease`]'s audit history, in journal order. +/// Transitions and rejection audits share one ordered list because +/// each rejection audit is only valid against the lease state at its +/// own journal position: a stale write audited at epoch 1 stays valid +/// after a revocation advances the epoch to 2, and a validator that +/// replays every transition before any audit would refuse it +/// (Codex review on PR #5571). +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "record", rename_all = "snake_case")] +pub enum WriteLeaseHistory { + /// A lease transition (grant, renewal, handoff, revocation). + Transition(WriteLeaseTransition), + /// A rejected append attempt the authority audited. + Rejection(LeaseWriteRejection), +} + /// The journal-resident write lease for one world (design #2819, /// phase 2 of issue #3804): the derived state the append gate fences /// on, rebuilt by journal replay and carried whole in checkpoints so @@ -709,10 +725,10 @@ pub struct WriteLease { /// When the current term expires, in unix milliseconds on the /// authority's clock; `None` for an untermed holder. pub expires_at_ms: Option, - /// Every transition applied, in journal order. - pub transitions: Vec, - /// Every audited rejected append attempt, in journal order. - pub rejections: Vec, + /// Every transition and audited rejection, interleaved in + /// journal order — the order is load-bearing (see + /// [`WriteLeaseHistory`]). + pub history: Vec, } impl Default for WriteLease { @@ -730,11 +746,26 @@ impl WriteLease { holder: Some(DO_WRITE_LEASE_HOLDER.to_owned()), epoch: 0, expires_at_ms: None, - transitions: Vec::new(), - rejections: Vec::new(), + history: Vec::new(), } } + /// Every transition applied, in journal order. + pub fn transitions(&self) -> impl Iterator { + self.history.iter().filter_map(|entry| match entry { + WriteLeaseHistory::Transition(transition) => Some(transition), + WriteLeaseHistory::Rejection(_) => None, + }) + } + + /// Every audited rejected append attempt, in journal order. + pub fn rejections(&self) -> impl Iterator { + self.history.iter().filter_map(|entry| match entry { + WriteLeaseHistory::Transition(_) => None, + WriteLeaseHistory::Rejection(rejection) => Some(rejection), + }) + } + /// Whether this lease is still the untouched bootstrap state (a /// checkpoint then omits it, exactly as older checkpoints did). #[must_use] @@ -798,14 +829,15 @@ impl WriteLease { self.advance_epoch(epoch, "grant")?; self.holder = Some(holder.to_owned()); self.expires_at_ms = expires_at_ms; - self.transitions.push(WriteLeaseTransition { - kind: LeaseTransitionKind::Granted, - from: None, - to: Some(holder.to_owned()), - epoch, - expires_at_ms, - reason: None, - }); + self.history + .push(WriteLeaseHistory::Transition(WriteLeaseTransition { + kind: LeaseTransitionKind::Granted, + from: None, + to: Some(holder.to_owned()), + epoch, + expires_at_ms, + reason: None, + })); Ok(()) } @@ -833,14 +865,15 @@ impl WriteLease { "journal renewal moves the write lease expiry backwards" ); self.expires_at_ms = Some(expires_at_ms); - self.transitions.push(WriteLeaseTransition { - kind: LeaseTransitionKind::Renewed, - from: Some(holder.to_owned()), - to: Some(holder.to_owned()), - epoch, - expires_at_ms: Some(expires_at_ms), - reason: None, - }); + self.history + .push(WriteLeaseHistory::Transition(WriteLeaseTransition { + kind: LeaseTransitionKind::Renewed, + from: Some(holder.to_owned()), + to: Some(holder.to_owned()), + epoch, + expires_at_ms: Some(expires_at_ms), + reason: None, + })); Ok(()) } @@ -867,14 +900,15 @@ impl WriteLease { self.advance_epoch(epoch, "handoff")?; self.holder = Some(to.to_owned()); self.expires_at_ms = expires_at_ms; - self.transitions.push(WriteLeaseTransition { - kind: LeaseTransitionKind::HandedOff, - from: Some(from.to_owned()), - to: Some(to.to_owned()), - epoch, - expires_at_ms, - reason: None, - }); + self.history + .push(WriteLeaseHistory::Transition(WriteLeaseTransition { + kind: LeaseTransitionKind::HandedOff, + from: Some(from.to_owned()), + to: Some(to.to_owned()), + epoch, + expires_at_ms, + reason: None, + })); Ok(()) } @@ -895,14 +929,15 @@ impl WriteLease { self.advance_epoch(epoch, "revocation")?; self.holder = None; self.expires_at_ms = None; - self.transitions.push(WriteLeaseTransition { - kind: LeaseTransitionKind::Revoked, - from: Some(former_holder.to_owned()), - to: None, - epoch, - expires_at_ms: None, - reason: Some(reason.to_owned()), - }); + self.history + .push(WriteLeaseHistory::Transition(WriteLeaseTransition { + kind: LeaseTransitionKind::Revoked, + from: Some(former_holder.to_owned()), + to: None, + epoch, + expires_at_ms: None, + reason: Some(reason.to_owned()), + })); Ok(()) } @@ -937,7 +972,7 @@ impl WriteLease { rejection.reason ); ensure!( - !self.rejections.iter().any(|prior| { + !self.rejections().any(|prior| { prior.writer == rejection.writer && prior.presented_epoch == rejection.presented_epoch && prior.operation_id == rejection.operation_id @@ -945,7 +980,7 @@ impl WriteLease { "journal audits rejected write {:?} twice", rejection.operation_id ); - self.rejections.push(rejection); + self.history.push(WriteLeaseHistory::Rejection(rejection)); Ok(()) } @@ -960,54 +995,56 @@ impl WriteLease { /// state that differs from the claimed one. pub fn validate_history(&self) -> Result<()> { let mut replayed = Self::bootstrap(); - for transition in &self.transitions { - match transition.kind { - LeaseTransitionKind::Granted => replayed.apply_grant( - transition - .to - .as_deref() - .context("lease grant transition names no holder")?, - transition.epoch, - transition.expires_at_ms, - )?, - LeaseTransitionKind::Renewed => replayed.apply_renew( - transition - .to - .as_deref() - .context("lease renewal transition names no holder")?, - transition.epoch, - transition - .expires_at_ms - .context("lease renewal transition names no expiry")?, - )?, - LeaseTransitionKind::HandedOff => replayed.apply_handoff( - transition - .from - .as_deref() - .context("lease handoff transition names no source")?, - transition - .to - .as_deref() - .context("lease handoff transition names no target")?, - transition.epoch, - transition.expires_at_ms, - )?, - LeaseTransitionKind::Revoked => replayed.apply_revoke( - transition - .from - .as_deref() - .context("lease revocation transition names no holder")?, - transition.epoch, - transition - .reason - .as_deref() - .context("lease revocation transition names no reason")?, - )?, + for entry in &self.history { + match entry { + WriteLeaseHistory::Transition(transition) => match transition.kind { + LeaseTransitionKind::Granted => replayed.apply_grant( + transition + .to + .as_deref() + .context("lease grant transition names no holder")?, + transition.epoch, + transition.expires_at_ms, + )?, + LeaseTransitionKind::Renewed => replayed.apply_renew( + transition + .to + .as_deref() + .context("lease renewal transition names no holder")?, + transition.epoch, + transition + .expires_at_ms + .context("lease renewal transition names no expiry")?, + )?, + LeaseTransitionKind::HandedOff => replayed.apply_handoff( + transition + .from + .as_deref() + .context("lease handoff transition names no source")?, + transition + .to + .as_deref() + .context("lease handoff transition names no target")?, + transition.epoch, + transition.expires_at_ms, + )?, + LeaseTransitionKind::Revoked => replayed.apply_revoke( + transition + .from + .as_deref() + .context("lease revocation transition names no holder")?, + transition.epoch, + transition + .reason + .as_deref() + .context("lease revocation transition names no reason")?, + )?, + }, + WriteLeaseHistory::Rejection(rejection) => { + replayed.apply_rejection(rejection.clone())?; + } } } - for rejection in &self.rejections { - replayed.apply_rejection(rejection.clone())?; - } ensure!( replayed == *self, "checkpoint write lease history does not reproduce its claimed state" diff --git a/pkgs/id/src/world/tests.rs b/pkgs/id/src/world/tests.rs index 9ed610758..5946b3865 100644 --- a/pkgs/id/src/world/tests.rs +++ b/pkgs/id/src/world/tests.rs @@ -634,11 +634,19 @@ mod tests { assert_eq!(lease.holder.as_deref(), Some(DO_WRITE_LEASE_HOLDER)); assert_eq!(lease.epoch, 3); assert_eq!(lease.expires_at_ms, None); - assert_eq!(lease.transitions.len(), 4); - assert_eq!(lease.transitions[0].kind, LeaseTransitionKind::HandedOff); - assert_eq!(lease.transitions[1].kind, LeaseTransitionKind::Renewed); - assert_eq!(lease.transitions[2].kind, LeaseTransitionKind::Revoked); - assert_eq!(lease.transitions[3].kind, LeaseTransitionKind::Granted); + let kinds: Vec = lease + .transitions() + .map(|transition| transition.kind) + .collect(); + assert_eq!( + kinds, + [ + LeaseTransitionKind::HandedOff, + LeaseTransitionKind::Renewed, + LeaseTransitionKind::Revoked, + LeaseTransitionKind::Granted, + ] + ); // The ex-holder is fenced at the new epoch, and the returned // holder presenting an old epoch is stale. assert_eq!( @@ -688,8 +696,8 @@ mod tests { // holder, the epoch, or any program state. assert_eq!(lease.holder.as_deref(), Some("native-1")); assert_eq!(lease.epoch, 1); - assert_eq!(lease.rejections.len(), 1); - assert_eq!(lease.rejections[0].operation_id, "op-1"); + assert_eq!(lease.rejections().count(), 1); + assert_eq!(lease.rejections().next().unwrap().operation_id, "op-1"); assert_eq!(restored.core.current_sequence(), 0); assert!(restored.replay.is_empty()); @@ -808,6 +816,48 @@ mod tests { } } + #[test] + fn write_lease_rejection_audit_stays_valid_across_later_transitions() { + // Codex P1 on PR #5571: an audit is valid against the lease + // state at its own journal position, not the final state. + // Handoff (epoch 1) -> audit a stale write (observed epoch 1) + // -> revoke (epoch 2) -> grant (epoch 3) must restore from the + // original journal AND from its compaction: a validator that + // replays every transition before any audit would refuse the + // audit for naming observed epoch 1 against a final epoch of + // 3, bricking the world after compaction. + let restored = WorldCore::restore( + "w", + WorldLimits::default(), + [ + lease_header(), + spill_handoff(), + stale_rejection(), + JournalEntry::LeaseRevoked { + former_holder: "native-1".to_owned(), + epoch: 2, + reason: "term expired without renewal".to_owned(), + }, + JournalEntry::LeaseGranted { + holder: DO_WRITE_LEASE_HOLDER.to_owned(), + epoch: 3, + expires_at_ms: None, + }, + ], + ) + .unwrap(); + let lease = restored.core.write_lease().clone(); + assert_eq!(lease.epoch, 3); + assert_eq!(lease.rejections().count(), 1); + lease.validate_history().unwrap(); + + let mut core = restored.core; + core.active_module_hash = Some("h".to_owned()); + let compacted = core.checkpoint_entries("state".to_owned()).unwrap(); + let restored = WorldCore::restore("w", WorldLimits::default(), compacted).unwrap(); + assert_eq!(restored.core.write_lease(), &lease); + } + #[test] fn write_lease_survives_restore_and_checkpoint_compaction() { let restored = WorldCore::restore( diff --git a/pkgs/id/src/world_store.rs b/pkgs/id/src/world_store.rs index e98c762ab..b6587f965 100644 --- a/pkgs/id/src/world_store.rs +++ b/pkgs/id/src/world_store.rs @@ -1234,7 +1234,7 @@ mod tests { assert_eq!(lease.holder.as_deref(), Some("native-1")); assert_eq!(lease.epoch, 1); assert_eq!(lease.expires_at_ms, Some(40_000)); - assert_eq!(lease.transitions.len(), 2); + assert_eq!(lease.transitions().count(), 2); } #[test]