diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 4ef590618..cda2a777e 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -16,8 +16,8 @@ use cipherbox_core::ipns::{IpnsName, IpnsRecord}; use cipherbox_core::kdf; use cipherbox_core::payload::RepointObject; use cipherbox_core::seal::{ - ChildRef, NodeKind as CoreNodeKind, PreservedFields, ReadBody, Version as CoreVersion, - decode_envelope, open_read_body, + ChildRef, GrantSetCommitment, NodeKind as CoreNodeKind, PreservedFields, ReadBody, + Version as CoreVersion, decode_envelope, open_read_body, sign_grant_set, }; use cipherbox_core::suite::ecdsa::EcdsaSigner; use cipherbox_core::suite::ed25519::Ed25519Signer; @@ -32,7 +32,10 @@ use cipherbox_engine::content::{ }; use cipherbox_engine::facade::PendingClass; use cipherbox_engine::net::OrphanHeads; -use cipherbox_engine::net::author::{AuthoredHead, EnvelopeAuthoring, author_child_envelope}; +use cipherbox_engine::net::author::{ + AuthoredHead, ENVELOPE_V, EnvelopeAuthoring, author_child_envelope, + author_scope_root_with_section, +}; use cipherbox_engine::net::{ ChildAdopter, REGISTRY_BATCH_MAX, ResolveOutcome, StagingRetireLedger, resolve, }; @@ -52,15 +55,16 @@ use cipherbox_engine::sync::{ }; use cipherbox_engine::testkit::fakes::{InMemoryRecordStore, InMemoryStagingStore}; use cipherbox_engine::testkit::{ - FakeDevice, FakeSeamTypes, FakeWorld, OWNER_ROOT_EPOCH as EPOCH, + FakeDevice, FakeSeamTypes, FakeWorld, OWNER_ROOT_EPOCH as EPOCH, OWNER_ROOT_PSEUDONYM_SEED, OWNER_ROOT_SCOPE_SEED as READ_SCOPE_SEED, OWNER_ROOT_WRITE_SCOPE_SEED as WRITE_SCOPE_SEED, OwnerRootSpec, SeededEntropy, block_on, frame_version as frame, owner_root_fixture, }; use cipherbox_engine::{ - ApiBaseUrl, ApiClient, BlockProgress, Command, CommandOutcome, ContentProfile, DeadLetter, - DeadLetterReason, DefaultsReason, Engine, EngineError, Event, EventStream, GatewayConfig, - LoginSecret, MAX_OPEN_STREAMS, NodeId, NodeKind, Op, OpPhase, OverBudgetCause, Placement, - PlacementRefusal, RecordSeal, StoragePolicy, SyncTimingProfile, WriteTarget, stage_op, + ApiBaseUrl, ApiClient, BlockProgress, Command, CommandOutcome, CommittedSet, ContentProfile, + DeadLetter, DeadLetterReason, DefaultsReason, Engine, EngineError, Event, EventStream, + GatewayConfig, LoginSecret, MAX_OPEN_STREAMS, NodeId, NodeKind, Op, OpPhase, OverBudgetCause, + Placement, PlacementRefusal, PrevEpochSeed, RecordSeal, ResealSeeds, ScopeRootIdentity, + StoragePolicy, SyncTimingProfile, WriteHistory, WriteTarget, reseal_scope_root, stage_op, }; const SECRET: [u8; 32] = [7u8; 32]; @@ -69,6 +73,11 @@ const SCOPE: [u8; 16] = [0u8; 16]; const ROOT: NodeId = NodeId(SCOPE); /// The sole v2 re-point payload version (`facade::POINTER_PAYLOAD_VERSION`). const POINTER_PAYLOAD_VERSION: u64 = 1; +/// The override seed a rotation mints for `SCOPE`'s second read epoch. +const ROTATED_READ_SCOPE_SEED: [u8; 32] = [0xA5; 32]; +/// The stable per-scope pointer read key the owner-root fixture's grant blobs +/// carry. +const POINTER_READ_KEY: [u8; 32] = [0x88; 32]; const TTL_NANOS: u64 = 2_000_000_000; /// The destination set the upload mark opens on. const DESTINATIONS_LEN: usize = Destinations::LEN; @@ -734,9 +743,15 @@ fn write_signer(node: NodeId) -> Ed25519Signer { kdf::ipns_keypair(kdf::write_seed(&WRITE_SCOPE_SEED, &node.0).as_bytes()) } -/// A node's per-node read key (`nodeSeed(readScopeSeed, id)` → `readKey`). +/// A node's per-node read key (`nodeSeed(scopeSeed, id)` → `readKey`) under +/// `scope_seed` — the epoch's seed is the only thing that moves it. +fn read_key_under(scope_seed: &[u8; 32], node: NodeId) -> [u8; 32] { + *kdf::read_key(kdf::node_seed(scope_seed, &node.0).as_bytes()).as_bytes() +} + +/// A node's per-node read key under the account's first read-scope seed. fn read_key_of(node: NodeId) -> [u8; 32] { - *kdf::read_key(kdf::node_seed(&READ_SCOPE_SEED, &node.0).as_bytes()).as_bytes() + read_key_under(&READ_SCOPE_SEED, node) } /// The `(sequence, headCid)` of the record currently published under `node`'s @@ -3657,6 +3672,109 @@ fn a_planted_focus_record_never_renders() { } } +/// Rotate `SCOPE`'s read plane: the vault root republishes at the next read +/// epoch under a freshly minted override seed, carrying the history link a +/// current-seed holder walks backward through (CONTEXT.md "History link"). The +/// write plane stands still, so the root keeps its name and its write epoch. +fn rotate_read_epoch(records: &InMemoryRecordStore, blocks: &Blocks) { + let owner_identity = owner_identity(); + let owner_verifier = owner_identity.verifying_key(); + let owner_pseudonym = Ed25519Signer::from_seed(OWNER_ROOT_PSEUDONYM_SEED); + let owner_enc = kdf::enc_subkey(&SECRET); + let owner_enc_pub = owner_enc.public(); + let name = write_name(ROOT); + + let commitment = GrantSetCommitment { + ipns_name: name.as_str().as_bytes().to_vec(), + owner_pseudonym_pk: owner_pseudonym.verifying_key().to_bytes(), + entries: Vec::new(), + unknown: PreservedFields::new(), + }; + let commitment_sig = sign_grant_set(&owner_identity, &commitment) + .expect("the owner signs its own grant set") + .to_compact(); + let section = reseal_scope_root( + &mut SeededEntropy::new(EPOCH + 1), + &ScopeRootIdentity { + v: ENVELOPE_V, + scope_id: SCOPE, + ipns_name: name.as_str().as_bytes(), + owner_enc_pub: &owner_enc_pub, + owner_enc_secret: None, + parent_node_seed: None, + owes_ascent_link: false, + pseudonym_signer: &owner_pseudonym, + }, + &ResealSeeds { + override_seed: &ROTATED_READ_SCOPE_SEED, + read_epoch: EPOCH + 1, + prev: Some(PrevEpochSeed { + seed: &READ_SCOPE_SEED, + epoch: EPOCH, + }), + write_scope_seed: &WRITE_SCOPE_SEED, + write_epoch: EPOCH, + write_history: WriteHistory::Carried(&[]), + pointer_read_key: &POINTER_READ_KEY, + }, + &CommittedSet { + commitment: &commitment, + commitment_sig: &commitment_sig, + grant_ledger: &[], + direct_child_scope_index: &[], + }, + &[], + ) + .expect("the root re-seals at the next read epoch"); + + // A cut re-seals the scope root and nothing else: its children keep the + // records — and the epoch — they already published under. + let head = author_scope_root_with_section( + EnvelopeAuthoring { + node_id: ROOT.0, + scope_id: SCOPE, + epoch: EPOCH + 1, + read_key: &read_key_under(&ROTATED_READ_SCOPE_SEED, ROOT), + nonce: &[0x7E; 24], + body: &ReadBody::Folder { + created_at: 0, + modified_at: 0, + children: published_children(records, blocks, ROOT), + unknown: PreservedFields::new(), + }, + carried_unknown: PreservedFields::new(), + carried_epoch_tag_unknown: PreservedFields::new(), + }, + &name, + §ion, + &owner_verifier, + ) + .expect("the rotated root authors"); + publish_next_record(records, blocks, ROOT, &head); +} + +/// The lazy wave reaches `folder`: republish exactly the children it lists +/// today, re-sealed at the scope's current epoch under the rotation's seed. +fn sweep_folder(records: &InMemoryRecordStore, blocks: &Blocks, folder: NodeId) { + let head = author_child_envelope(EnvelopeAuthoring { + node_id: folder.0, + scope_id: SCOPE, + epoch: EPOCH + 1, + read_key: &read_key_under(&ROTATED_READ_SCOPE_SEED, folder), + nonce: &[0x6D; 24], + body: &ReadBody::Folder { + created_at: 0, + modified_at: 0, + children: published_children(records, blocks, folder), + unknown: PreservedFields::new(), + }, + carried_unknown: PreservedFields::new(), + carried_epoch_tag_unknown: PreservedFields::new(), + }) + .expect("the sweep authors a valid record"); + publish_next_record(records, blocks, folder, &head); +} + /// Epoch lag is sweep-pending staleness, not abuse (CONTEXT.md "Epoch lag"): a /// focused folder the lazy wave has not swept yet rejects fail-closed, but the /// owner's own rotation must not read as an attack on the host's abuse channel. @@ -3664,7 +3782,7 @@ fn a_planted_focus_record_never_renders() { fn an_epoch_lagged_focus_folder_rejects_without_raising_abuse() { let DeepCreate { world, - bob, + blocks, mut engine_b, mut events_b, mut tasks_b, @@ -3674,9 +3792,16 @@ fn an_epoch_lagged_focus_folder_rejects_without_raising_abuse() { block_on(engine_b.command(Command::SetFocus { node: Some(photos) })).unwrap(); assert_eq!(listed_names(&engine_b, photos), ["2026"]); - // A rotation raised the scope's read-epoch floor past the epoch this folder - // still publishes under. - block_on(bob.floor_store.raise_epoch_floor(&SCOPE, EPOCH + 1)).unwrap(); + // A real rotation: the root republishes at the next read epoch under a + // fresh seed, which is what raises this device's read-epoch floor. `photos` + // is not swept, so its own writer keeps publishing at the old epoch. + rotate_read_epoch(&world.record_store, &blocks); + concurrent_add( + &world.record_store, + &blocks, + photos, + child_ref([0x27; 16], "2027", CoreNodeKind::Folder), + ); let _ = events_so_far(&mut events_b); tick(&world, &engine_b, &mut tasks_b); @@ -3691,6 +3816,17 @@ fn an_epoch_lagged_focus_folder_rejects_without_raising_abuse() { .all(|event| !matches!(event, Event::AttributableAbuse { .. })), "an unswept folder is not an attacker" ); + + // The control: the same children, re-sealed at the current epoch, do + // render — so the leg above ran and rejected. + sweep_folder(&world.record_store, &blocks, photos); + tick(&world, &engine_b, &mut tasks_b); + + assert_eq!( + listed_names(&engine_b, photos), + ["2026", "2027"], + "the wave's re-seal at the current epoch is adopted" + ); } /// An unreachable record plane is availability staleness, never data loss: the diff --git a/crates/fuse/src/cache.rs b/crates/fuse/src/cache.rs index 1ee214e33..3e79362d8 100644 --- a/crates/fuse/src/cache.rs +++ b/crates/fuse/src/cache.rs @@ -4,7 +4,9 @@ //! //! Blocks are keyed by the engine [`StreamHandle`] that produced them. A stream //! pins one head version for its whole life, so a block cached under its handle -//! can never be served as a slice of a different version. +//! can never be served as a slice of a different version. They are held whole: +//! a truncate floor belongs to the handle that set it, not to the stream, so a +//! reader clamps at use rather than the cache clamping at store. //! //! Every block is owned here and nowhere else, so this cache is the terminal //! owner that zeroizes it (security rule 7). Zeroization rides `Drop`, not an @@ -131,6 +133,13 @@ impl ChunkCache { self.retained } + /// Serve a block without moving it in the recency order — what a one-shot + /// pass reads through, so walking a whole file cannot re-rank a reader's + /// hot blocks behind the blocks that walk touched once. + pub fn peek(&self, key: BlockKey) -> Option<&[u8]> { + Some(self.blocks.get(&key)?.plaintext.as_slice()) + } + /// Serve a block, making it the most recently used. pub fn get(&mut self, key: BlockKey) -> Option<&[u8]> { let tick = self.next_tick; @@ -144,13 +153,15 @@ impl ChunkCache { /// Retain `plaintext` under `key`, evicting until it fits. A block larger /// than the whole budget is dropped rather than cached: the ceiling is the - /// mount's promise, not a target to overshoot. + /// mount's promise, not a target to overshoot. An empty block — the answer + /// past the end of a version — is dropped too: it would grow the table + /// without spending a byte of the budget that bounds it. pub fn insert(&mut self, key: BlockKey, plaintext: Vec) { // Wrapped before any early return: a block this cache refuses is still // plaintext it took ownership of, and must leave memory zeroed. let plaintext = Zeroizing::new(plaintext); let len = plaintext.len(); - if len > self.budget.max_bytes { + if len == 0 || len > self.budget.max_bytes { return; } self.drop_block(key); @@ -306,6 +317,34 @@ mod tests { assert!(cache.get((STREAM, 3)).is_some()); } + #[test] + fn an_empty_block_is_never_retained() { + let mut cache = ChunkCache::new(tiny()); + cache.insert((STREAM, 0), Vec::new()); + assert_eq!(cache.get((STREAM, 0)), None); + } + + #[test] + fn peeking_serves_a_block_without_reordering_the_recency_it_evicts_by() { + let mut cache = ChunkCache::new(tiny()); + for index in 0..3 { + cache.insert((STREAM, index), block(index as u8, 16)); + } + + // Newest-first, so a serve that promoted would leave the block the + // walk saw last as the eviction candidate instead of the oldest. + for index in (0..3).rev() { + assert!(cache.peek((STREAM, index)).is_some()); + } + cache.insert((STREAM, 3), block(3, 16)); + + assert_eq!( + cache.get((STREAM, 0)), + None, + "the peek pass promoted the oldest block instead of leaving it first out" + ); + } + #[test] fn a_block_larger_than_the_whole_budget_is_never_retained() { let mut cache = ChunkCache::new(tiny()); diff --git a/crates/fuse/src/ops.rs b/crates/fuse/src/ops.rs index beea8b94a..952e9a6c0 100644 --- a/crates/fuse/src/ops.rs +++ b/crates/fuse/src/ops.rs @@ -83,6 +83,16 @@ impl Pending { } } +/// How a block-fetching path uses the chunk cache: a path that will come back +/// to these bytes retains and promotes them; a one-shot pass reads through, +/// leaving neither the budget nor the recency order spent on blocks it will +/// never ask for again. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Retain { + Hot, + Scan, +} + /// The operation core for one mount session: the engine it projects, the host /// adapter it pushes invalidation to, and the session's inode and handle maps. pub struct OperationCore { @@ -321,11 +331,7 @@ impl OperationCore { .await?; let taken = take_from(&mut out, &block, within, want); let whole = block.len() as u64 == block_bytes; - // Caching an empty block past the end would grow the table - // without spending a byte of the budget that bounds it. - if !block.is_empty() { - self.cache.insert((stream, index), block); - } + self.cache.insert((stream, index), block); (taken, whole) } }; @@ -377,7 +383,7 @@ impl OperationCore { if within == 0 && take as u64 == block_bytes { self.spill_mut(handle)?.put(index, &rest[..take])?; } else { - let mut block = self.version_block(handle, index).await?; + let mut block = self.version_block(handle, index, Retain::Scan).await?; block[within..within + take].copy_from_slice(&rest[..take]); self.spill_mut(handle)?.put(index, &block)?; } @@ -558,7 +564,7 @@ impl OperationCore { while offset < len { let index = offset / block_bytes; let take = block_bytes.min(len - offset) as usize; - let block = self.version_block(handle, index).await?; + let block = self.version_block(handle, index, Retain::Scan).await?; self.engine.push_chunk(write, &block[..take]).await?; offset += take as u64; } @@ -581,7 +587,7 @@ impl OperationCore { let index = cursor / block_bytes; let within = (cursor - index * block_bytes) as usize; let want = (end - cursor) as usize; - let block = self.version_block(handle, index).await?; + let block = self.version_block(handle, index, Retain::Hot).await?; cursor += take_from(&mut out, &block, within, want) as u64; } Ok(core::mem::take(&mut *out)) @@ -590,10 +596,15 @@ impl OperationCore { /// Block `index` of the version a handle's writes are building: its spill /// block if it took one, else the base version's bytes, else the zeros a /// hole reads as. Always a whole block wide. + /// + /// The base block is cached whole and clamped at use: the truncate floor is + /// per-handle `Pending` state, not a property of the stream every handle on + /// the file shares. async fn version_block( &mut self, handle: HandleId, index: u64, + retain: Retain, ) -> Result>, VfsError> { let block_bytes = self.cache.block_bytes(); let pending = self.pending.get_mut(&handle).ok_or(VfsError::BadHandle)?; @@ -610,11 +621,22 @@ impl OperationCore { } // Clamped to the floor a truncate left: past it the version's own bytes // are gone, and the file reads as the hole they left. - let want = block_bytes.min(base_len - at); + let want = block_bytes.min(base_len - at) as usize; let stream = self.stream_for(handle).await?; - let base = Zeroizing::new(self.engine.read_stream(stream, at, want).await?); - let take = base.len().min(out.len()); - out[..take].copy_from_slice(&base[..take]); + let held = match retain { + Retain::Hot => self.cache.get((stream, index)), + Retain::Scan => self.cache.peek((stream, index)), + }; + if let Some(block) = held { + clamp_into(&mut out, block, want); + return Ok(out); + } + let mut base = Zeroizing::new(self.engine.read_stream(stream, at, block_bytes).await?); + clamp_into(&mut out, &base, want); + if retain == Retain::Hot { + self.cache + .insert((stream, index), core::mem::take(&mut *base)); + } Ok(out) } @@ -820,6 +842,13 @@ impl OperationCore { } } +/// Copy the first `want` bytes of `block` over the head of `out`, leaving the +/// rest of `out` as the zeros a truncate's floor reads as. +fn clamp_into(out: &mut [u8], block: &[u8], want: usize) { + let take = want.min(block.len()); + out[..take].copy_from_slice(&block[..take]); +} + /// Append at most `want` bytes of `block` from `within` onto `out`, reporting /// how many the block actually held. fn take_from(out: &mut Zeroizing>, block: &[u8], within: usize, want: usize) -> usize { diff --git a/crates/fuse/tests/fuse_op_core.rs b/crates/fuse/tests/fuse_op_core.rs index dd1ebe502..561bab4c3 100644 --- a/crates/fuse/tests/fuse_op_core.rs +++ b/crates/fuse/tests/fuse_op_core.rs @@ -1798,6 +1798,100 @@ mod published { } } + #[test] + fn a_sub_block_write_merges_over_the_block_the_read_path_cached() { + let plaintext = clip_bytes(); + let mut mount = mount_published(&plaintext, CacheBudget::CI); + let handle = + block_on(mount.core.open(mount.ino, Access::ReadWrite)).expect("the file opens"); + + block_on(mount.core.read(handle, 0, 4)).expect("the read caches the chunk it framed"); + let after_read = block_fetches(&mount.device); + + block_on(mount.core.write(handle, 1, b"a")).expect("the write merges into that chunk"); + + assert_eq!( + block_fetches(&mount.device), + after_read, + "the merge re-fetched a base block the mount was already holding" + ); + } + + #[test] + fn reading_through_pending_writes_fetches_each_base_block_once() { + let plaintext = clip_bytes(); + let mut mount = mount_published(&plaintext, CacheBudget::CI); + let handle = + block_on(mount.core.open(mount.ino, Access::ReadWrite)).expect("the file opens"); + + // One sub-block write, so every later read renders through the pending + // overlay and every untouched block still comes off the base version. + block_on(mount.core.write(handle, 1, b"a")).expect("the write lands"); + + let whole = plaintext.len() as u32; + let first = block_on(mount.core.read(handle, 0, whole)).expect("the first pass"); + let after_first = block_fetches(&mount.device); + let second = block_on(mount.core.read(handle, 0, whole)).expect("the second pass"); + + assert_eq!(first, second, "both passes render the same file"); + assert_eq!( + block_fetches(&mount.device), + after_first, + "the second pass re-fetched base blocks the first had already cached" + ); + } + + #[test] + fn a_cached_base_block_is_still_clamped_to_the_floor_a_shrink_left() { + let plaintext = clip_bytes(); + let mut mount = mount_published(&plaintext, CacheBudget::CI); + let ino = mount.ino; + let handle = block_on(mount.core.open(ino, Access::ReadWrite)).expect("the file opens"); + block_on(mount.core.read(handle, 0, chunk() as u32)).expect("the read caches the chunk"); + + block_on(mount.core.truncate(ino, 4, Some(handle))).expect("the shrink"); + block_on(mount.core.write(handle, 6, b"xy")).expect("the write past the gap"); + block_on(mount.core.release(handle)).expect("the release commits"); + advance_and_pump(&mut mount); + + let mut expected = plaintext[..4].to_vec(); + expected.extend_from_slice(b"\0\0xy"); + let reader = opened(&mut mount); + assert_eq!( + block_on(mount.core.read(reader, 0, 16)).expect("the read"), + expected + ); + } + + #[test] + fn the_commit_walk_leaves_a_readers_hot_blocks_in_the_cache() { + let plaintext = clip_bytes(); + let budget = CacheBudget::for_profile(ContentProfile::CI, 5).expect("five chunks"); + let mut mount = mount_published(&plaintext, budget); + let reader = opened(&mut mount); + for index in 0..3 { + block_on(mount.core.read(reader, index * chunk(), chunk() as u32)) + .expect("the reader's chunks"); + } + + let writer = + block_on(mount.core.open(mount.ino, Access::ReadWrite)).expect("the file opens"); + block_on(mount.core.write(writer, 1, b"a")).expect("the write lands"); + block_on(mount.core.release(writer)).expect("the release commits the whole version"); + advance_and_pump(&mut mount); + + let before = block_fetches(&mount.device); + for index in 0..3 { + block_on(mount.core.read(reader, index * chunk(), chunk() as u32)) + .expect("the reader's chunks re-serve"); + } + assert_eq!( + block_fetches(&mount.device), + before, + "the commit walk evicted the reader's hot blocks" + ); + } + #[test] fn unmounting_releases_every_stream_and_the_plaintext_they_cached() { let plaintext = clip_bytes();