diff --git a/crates/engine/src/content/provider.rs b/crates/engine/src/content/provider.rs index 07e7e08b09..055cfb4ba8 100644 --- a/crates/engine/src/content/provider.rs +++ b/crates/engine/src/content/provider.rs @@ -279,7 +279,7 @@ fn headers(config: &ByoIpfsConfig, content_type: Option) -> Vec<(String, /// Why a provider connection test did not succeed. The first four are policy /// verdicts reached before any request is issued, kept distinct so a host can /// say which rule refused the config instead of showing a bare failure. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ProviderError { /// The endpoint is not an absolute `http(s)` URL whose authority is /// host-and-port bytes. It is spliced into a request URL, so a `file:`, diff --git a/crates/engine/src/facade.rs b/crates/engine/src/facade.rs index 9a02e6d8b5..6c588932ed 100644 --- a/crates/engine/src/facade.rs +++ b/crates/engine/src/facade.rs @@ -77,7 +77,7 @@ use crate::sync::provision::{ use crate::sync::rebase::{QueueScan, QueueScanMemo, decode_queue}; use cipherbox_core::hex::lower as hex_lower; -pub use crate::sync::drain::BlockedOp; +pub use crate::sync::drain::{BlockedOp, SettingsHold}; pub use crate::sync::rebase::DeadLetterReason; use crate::sync::record::{RecordReader, RecordSeal}; use crate::sync::refresh::{ManualRefresh, RefreshVerdict}; @@ -293,6 +293,10 @@ pub struct SessionStatus { /// this is a state that *clears*, and a lost "resumed" would strand a host /// on a blockage that is gone. pub blocked: Option, + /// The settings-refused hold, if the drain has one. Read for the same + /// reason as `blocked`, and it names the rule so a host can tell the member + /// which part of their own provider config to fix. + pub settings_hold: Option, /// How many durable queue entries this session holds but cannot read /// (CONTEXT.md "Retained record"). Deliberately unattributed — it says the /// device is not empty, never whose work it holds — and it exists so an @@ -324,6 +328,8 @@ pub struct SnapshotView { pub dead_letters: Vec, /// See [`SessionStatus::blocked`]. pub blocked: Option, + /// See [`SessionStatus::settings_hold`]. + pub settings_hold: Option, /// See [`SessionStatus::retained_records`]. pub retained_records: usize, /// See [`SessionStatus::staleness`]. @@ -1769,6 +1775,10 @@ pub struct Engine { /// [`snapshot`](Self::snapshot). In-memory: a restart re-derives it from the /// next drain attempt's own 413 rather than trusting a stale verdict. blocked: Rc>>, + /// The drain's settings-refused hold, on the same in-memory terms as + /// [`blocked`](Self::blocked): a restart re-derives it from the next drain + /// attempt's own verdict. + settings_hold: Rc>>, /// Pinned bytes a published prune still owes the registry, written by the /// drain tick and read by [`pending_reclaim_bytes`](Self::pending_reclaim_bytes). /// In-memory: the durable record is the retire ledger, which every pass re-reads. @@ -1855,6 +1865,7 @@ impl Engine { dead_letters: Rc::new(RefCell::new(BTreeMap::new())), queue_scan: RefCell::new(QueueScanMemo::default()), blocked: Rc::new(RefCell::new(None)), + settings_hold: Rc::new(RefCell::new(None)), pending_reclaim: Rc::new(Cell::new(0)), orphan_heads: Rc::new(OrphanHeads::default()), alive: Rc::new(Cell::new(true)), @@ -2378,6 +2389,7 @@ where { let scope_write_seeds = self.scope_write_seeds.clone(); let dead_letters = self.dead_letters.clone(); let blocked = self.blocked.clone(); + let settings_hold = self.settings_hold.clone(); let pending_reclaim = self.pending_reclaim.clone(); let content_profile = self.content_profile; let orphan_heads = self.orphan_heads.clone(); @@ -2579,6 +2591,7 @@ where { base: &base, held: &held, blocked: &blocked, + settings_hold: &settings_hold, pending_reclaim: &pending_reclaim, orphan_heads: &orphan_heads, cancels: &cancels, @@ -3382,6 +3395,7 @@ where { Ok(SessionStatus { dead_letters: self.retained_dead_letters(), blocked: *self.blocked.borrow(), + settings_hold: *self.settings_hold.borrow(), retained_records, staleness: self.staleness_now(), }) @@ -3461,6 +3475,7 @@ where { ancestors, dead_letters: self.retained_dead_letters(), blocked: *self.blocked.borrow(), + settings_hold: *self.settings_hold.borrow(), retained_records: scan.retained, staleness: self.staleness_now(), }) @@ -4778,6 +4793,27 @@ mod tests { ); } + /// A hold is a state that *clears*, so it is read off both surfaces rather + /// than evented: a lost "released" would strand a host on a refusal the + /// member has already fixed in their settings. + #[test] + fn a_settings_refused_hold_reaches_both_read_surfaces() { + let (engine, _events) = started(); + let root = engine.root(); + let hold = SettingsHold { + op_id: OpId(1), + node: root, + refusal: crate::content::ProviderError::InsecureTransport, + }; + *engine.settings_hold.borrow_mut() = Some(hold); + + assert_eq!( + block_on(engine.snapshot(root)).unwrap().settings_hold, + Some(hold) + ); + assert_eq!(block_on(engine.status()).unwrap().settings_hold, Some(hold)); + } + #[test] fn delete_removes_the_node_from_the_view() { let (mut engine, _events) = started(); diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 8f1462be79..2b77cc9da7 100644 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -102,10 +102,10 @@ pub use sync::{ AppliedOp, BlockedOp, Connectivity, DeadLetterReason, DropReason, FocusTarget, FocusWindow, HeadReconciliation, Link, NewNode, NodeMeta, Op, OpKind, OpRecordError, OpResolution, PointerError, PointerFetch, RecordClass, RecordReader, RecordSeal, Repair, Replaced, - ReplayReport, ScopeCrossing, SessionRole, Snapshot, StagedContent, TickCause, TickControl, - VaultPointerAdoption, apply_overlay, apply_repairs, classify, decode_queue, encode_op_record, - focus_set, observed_repair, rebase_one, reconcile_head, record_content_root_cid, replay, - resolve_vault_pointer, stage_op, + ReplayReport, ScopeCrossing, SessionRole, SettingsHold, Snapshot, StagedContent, TickCause, + TickControl, VaultPointerAdoption, apply_overlay, apply_repairs, classify, decode_queue, + encode_op_record, focus_set, observed_repair, rebase_one, reconcile_head, + record_content_root_cid, replay, resolve_vault_pointer, stage_op, }; /// Placeholder identity item; kept for the sibling crate stubs' dependency diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 1d09e39084..a3095f890d 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -42,9 +42,9 @@ use crate::api::{ApiClient, ApiError, QUOTA_EXCEEDED, REGISTRY_BATCH_REFUSED, UP use crate::content::{ ContentPlane, ContentProfile, ContentVersion, Expansion, Gateway, ProviderError, RootPlacement, SealedContent, expand_retire_targets, place_block, plan_prune, pre_flight_quota_check, - read_block, version_cids, + read_block, validate_byo_config, version_cids, }; -use crate::entropy::Entropy; +use crate::entropy::{Entropy, fresh_nonce}; use crate::facade::{BlockProgress, Event, NodeId, OpPhase}; use crate::gate::floor; use crate::grants::{UndoDestAdd, undo_dest_add_versioned}; @@ -235,8 +235,20 @@ enum Halt { /// exhausting the budget may retire what the op uploaded — the op's target /// is still unreachable, so no record a parent links names it. UploadAttempt, + /// The authored head is over the block ceiling its own ingress enforces. + /// Charged like an attempt — no re-author shrinks it, since a fresh nonce + /// moves the sealed bytes and never their count — but exhausting the budget + /// **preserves** the staged version instead of releasing it: the version is + /// intact and openable, and only the record naming it was too large. What + /// it does owe back is a create's own derived name, since the record that + /// would have referenced it is the one that never published. + HeadOversized, /// Classified-permanent: the same bytes are refused on every retry. Permanent(DeadLetterReason), + /// The member's own provider settings were refused before any request was + /// built. Not a failure of the op — it holds the head and its staging + /// reservation until those settings change ([`SettingsHold`]). + HeldBySettings(ProviderError), /// Over the account quota. Not a failure of the op — it holds the head and /// its staging reservation until a quota probe reports room. Blocked { @@ -346,6 +358,24 @@ pub struct BlockedOp { pub needed_bytes: u64, } +/// The queue head is held over rather than failed: [`validate_byo_config`] +/// refused the member's own provider settings before any request was built, so +/// every retry reaches the same verdict and charging one would spend the +/// version's budget and then release its staged blocks. It keeps its place and +/// its staging reservation until the settings name a placement that clears the +/// refusal. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SettingsHold { + /// The held op. + pub op_id: OpId, + /// The node the op targets, so a host can point at it. + pub node: NodeId, + /// Which rule refused the config. Render it through + /// [`ProviderError::check`], which names the rule and never the endpoint or + /// the bearer the settings carry. + pub refusal: ProviderError, +} + /// The owner-scope material one drain pass publishes under. Every field is /// borrowed from the live session; the drain zeroizes none of it. pub(crate) struct DrainScope<'a> { @@ -391,6 +421,10 @@ pub(crate) struct Drain<'a, T, H: Http, C: CredentialStore, F, S, St, Sch> { /// The over-quota hold, shared with the facade's read surface. It clears /// only here, on a quota probe reporting room. pub(crate) blocked: &'a RefCell>, + /// The settings-refused hold, shared with the facade's read surface. It + /// clears only here, once `placement` no longer carries the refusal that + /// took it. + pub(crate) settings_hold: &'a RefCell>, /// Pinned bytes the retire ledger still owes, shared with the facade's read /// surface. Rewritten at the end of every pass from the ledger itself. pub(crate) pending_reclaim: &'a Cell, @@ -600,6 +634,7 @@ where let queued = mine; if queued.is_empty() { self.clear_block(); + self.clear_settings_hold(); return report; } let Ok(mut attempts) = self.load_attempts(&all_ids).await else { @@ -621,6 +656,9 @@ where if !self.quota_admits_the_held_head(queued).await { return Ok(()); } + if !self.settings_admit_the_held_head(queued) { + return Ok(()); + } let mut pass = self.open_pass(scope).await?; let rebased = { @@ -698,19 +736,35 @@ where self.retire_cancelled(op_id).await; } } - Halt::Attempt | Halt::UploadAttempt => { + Halt::Attempt | Halt::UploadAttempt | Halt::HeadOversized => { if attempts.charge(op_id) < ATTEMPT_BUDGET { return; } let reason = DeadLetterReason::AttemptsExhausted; - if halt == Halt::UploadAttempt { - self.dead_letter(scope, op_id, op, reason, report).await; - } else if self.dequeue_op(op_id).await.is_ok() { + match halt { + Halt::UploadAttempt => { + self.dead_letter(scope, op_id, op, reason, report).await; + } + // On the terms `Halt::HeadOversized` carries: the version + // stays staged and openable, and only the name no published + // record reached goes back. + Halt::HeadOversized => { + if self.retire_unreferenced_name(scope, op).await.is_ok() + && self.preserve_dead_letter(op_id).await.is_ok() + && self.dequeue_op(op_id).await.is_ok() + { + report.dead_letters.push((op_id, op.target, reason)); + } + } // An acked PUT may be resolvable at the name, so nothing is // retired: unpinning content a live record still names is // loss, where leaving the rows charged is only a leak. - self.release_staged_blocks(op).await; - report.dead_letters.push((op_id, op.target, reason)); + _ => { + if self.dequeue_op(op_id).await.is_ok() { + self.release_staged_blocks(op).await; + report.dead_letters.push((op_id, op.target, reason)); + } + } } } // A conditional-edit loser keeps its staged version and retires @@ -727,13 +781,25 @@ where Halt::Permanent(reason) => { self.dead_letter(scope, op_id, op, reason, report).await; } + // One pass raises one halt, and each hold's own gate is what lets + // go of it — so taking one drops the other rather than leaving two + // cells claiming the same head for different reasons. Halt::Blocked { needed_bytes } => { + self.clear_settings_hold(); *self.blocked.borrow_mut() = Some(BlockedOp { op_id, node: op.target, needed_bytes, }); } + Halt::HeldBySettings(refusal) => { + self.clear_block(); + *self.settings_hold.borrow_mut() = Some(SettingsHold { + op_id, + node: op.target, + refusal, + }); + } } } @@ -744,7 +810,7 @@ where let Some(blocked) = *self.blocked.borrow() else { return true; }; - if !queued.iter().any(|(op_id, _)| *op_id == blocked.op_id) { + if !still_queued(queued, blocked.op_id) { self.clear_block(); return true; } @@ -772,6 +838,26 @@ where *self.blocked.borrow_mut() = None; } + /// Whether a settings-held head may be tried again this tick: only once the + /// placement this pass runs under stops reaching the verdict that took the + /// hold. + fn settings_admit_the_held_head(&self, queued: &[(OpId, Op)]) -> bool { + let Some(hold) = *self.settings_hold.borrow() else { + return true; + }; + if still_queued(queued, hold.op_id) + && placement_refusal(self.placement) == Some(hold.refusal) + { + return false; + } + self.clear_settings_hold(); + true + } + + fn clear_settings_hold(&self) { + *self.settings_hold.borrow_mut() = None; + } + /// This identity's queued ops, minus restore residue: an op at or below the /// durable drained-op mark already left this queue once, so the queue it /// came back in predates the completion record. @@ -2295,7 +2381,8 @@ where completes: Option, ) -> Result { let read_key = self.node_read_key(scope, &node.0); - let nonce = self.nonce().map_err(PublishHalt::before_the_put)?; + let nonce = fresh_nonce(&mut *self.entropy.borrow_mut()) + .map_err(|_| PublishHalt::before_the_put(Halt::UploadAttempt))?; let authoring = EnvelopeAuthoring { node_id: node.0, scope_id: scope.root.0, @@ -2533,26 +2620,41 @@ where } } + /// Retire only the name half of [`Self::registered_by`], for an abandonment + /// that keeps what the op uploaded. + async fn retire_unreferenced_name(&self, scope: &DrainScope<'_>, op: &Op) -> Result<(), Halt> { + let Some(name) = self.unreferenced_create_name(scope, op) else { + return Ok(()); + }; + retire(self.api, &[name]) + .await + .map_err(|_| Halt::UploadAttempt) + } + + /// The name a create derived, where nothing published references it yet: a + /// name some published record already references would leave a reference + /// outliving its referent, and the gate-passing base is the evidence — a + /// created node reaches it only once a parent record naming it published. + fn unreferenced_create_name(&self, scope: &DrainScope<'_>, op: &Op) -> Option { + let target_published = self.base.borrow().contains(op.target); + (!target_published && matches!(op.kind, OpKind::Create { .. })).then(|| { + derive_write_name(scope.write_scope_seed, &op.target.0) + .as_str() + .to_owned() + }) + } + /// The registry rows one op's publish registered, mirroring what the publish /// pipeline sends (`PublishRequest::registration`). /// - /// The child name goes only with an abandoned create whose target never - /// became reachable: a name some published record already references would - /// leave a reference outliving its referent, and the gate-passing base is - /// the evidence — a created node reaches it only once a parent record naming - /// it published. The content CIDs go with **any** content-bearing op that - /// reaches here: an abandonment only retires while the op's target is - /// unreachable, so no record a parent links can name the version. + /// The content CIDs go with **any** content-bearing op that reaches here: an + /// abandonment only retires while the op's target is unreachable, so no + /// record a parent links can name the version. /// /// Reads the manifest before [`Self::release_staged_blocks`] drops it: after /// that the leaf CIDs are recoverable from nowhere. async fn registered_by(&self, scope: &DrainScope<'_>, op: &Op) -> Vec { - let target_published = self.base.borrow().contains(op.target); - let name = (!target_published && matches!(op.kind, OpKind::Create { .. })).then(|| { - derive_write_name(scope.write_scope_seed, &op.target.0) - .as_str() - .to_owned() - }); + let name = self.unreferenced_create_name(scope, op); let content = match op.content_root_cid() { Some(root_cid) => version_cids( root_cid, @@ -2675,17 +2777,6 @@ where let node_seed = kdf::node_seed(scope.read_scope_seed, node_id); Zeroizing::new(*kdf::read_key(node_seed.as_bytes()).as_bytes()) } - - /// A fresh injected seal nonce. Fails closed: a reused nonce under one key - /// is a confidentiality break, never a degraded mode. - fn nonce(&self) -> Result<[u8; 24], Halt> { - let mut nonce = [0u8; 24]; - self.entropy - .borrow_mut() - .fill(&mut nonce) - .map_err(|_| Halt::UploadAttempt)?; - Ok(nonce) - } } /// The `contentCid` of a file body's head version — the conditional-edit @@ -2709,18 +2800,38 @@ fn seam(_: crate::seams::SeamError) -> Halt { Halt::Unclassified } -/// Classify an authoring refusal for the valve, off -/// [`AuthorError::is_trust_refusal`]. +/// Whether the op a hold names is still in this identity's queue — the shared +/// half of both hold gates, since a hold on an op that left is stale. +fn still_queued(queued: &[(OpId, Op)], held: OpId) -> bool { + queued.iter().any(|(op_id, _)| *op_id == held) +} + +/// Classify an authoring refusal for the valve. Exhaustive by construction: an +/// unclassified refusal retries free and forever, so a new variant must be +/// judged here rather than inheriting that arm. /// -/// Charged, not dead-lettered on sight: the scope root a trust refusal is +/// A trust refusal is charged, not dead-lettered on sight: the scope root it is /// authored from comes from the snapshot cache, which a later resolve replaces, /// so an immediate permanent verdict would abandon a user's ops over a cache -/// another tick repairs. The budget bounds the spin either way. +/// another tick repairs. +/// +/// An over-length head is charged on [`Halt::HeadOversized`]'s terms rather +/// than judged permanent, because the attacker-influenced side of a body must +/// never refuse an owner's publish outright (blueprint/core.md: an over-length +/// carry is truncated, never refused). +/// +/// [`AuthorError::Seal`] is the one refusal left uncharged: it judges the body +/// *this* pass built, which a rebase onto other state may not build again. fn classify_author(error: AuthorError) -> Halt { - if error.is_trust_refusal() { - Halt::UploadAttempt - } else { - Halt::Unclassified + match error { + AuthorError::GrantSectionOnChild + | AuthorError::MissingGrantSection + | AuthorError::InvalidGrantSection + | AuthorError::CommitmentNameMismatch + | AuthorError::CommitmentSignatureInvalid + | AuthorError::SectionSignatureInvalid => Halt::UploadAttempt, + AuthorError::HeadTooLarge { .. } => Halt::HeadOversized, + AuthorError::Seal(_) => Halt::Unclassified, } } @@ -2919,13 +3030,31 @@ fn classify_upload(error: ApiError, refused_bytes: u64) -> Halt { /// transport failure carries no verdict about these bytes, and charging the /// attempt budget for one would spend the version's five tries on a condition /// that repairs itself. Everything the provider *answered* is charged. +/// +/// A policy verdict is neither: it is deterministic, so it holds the op rather +/// than charging it ([`SettingsHold`]). fn classify_placement(error: ProviderError) -> Halt { match error { ProviderError::Unreachable => Halt::Unclassified, + ProviderError::InvalidEndpoint + | ProviderError::InsecureTransport + | ProviderError::BlockedAddress + | ProviderError::InvalidCredential => Halt::HeldBySettings(error), _ => Halt::UploadAttempt, } } +/// The verdict this session's placement reaches on the member's own config +/// before any request is built, which is what a [`Halt::HeldBySettings`] hold +/// waits on changing. Only the external-only leg can hold an op on it: a dual +/// write's mirror is best-effort and never fails the op. +fn placement_refusal(placement: &PlacementDecision) -> Option { + match placement { + Ok(Placement::External(config)) => validate_byo_config(config).err(), + _ => None, + } +} + /// A block count as [`BlockProgress`] carries it; the root manifest's own /// ceiling bounds a version's leaves far below `u32::MAX`. fn blocks(count: usize) -> u32 { @@ -2933,18 +3062,20 @@ fn blocks(count: usize) -> u32 { } /// The key-free classification an [`OpPhase::UploadFailed`] carries, or `None` -/// where the halt is not a failed attempt: an over-quota hold keeps the op and -/// its reservation, and the host reads it from `SnapshotView::blocked`. +/// where the halt is not a failed attempt: either hold keeps the op and its +/// reservation, and the host reads them from `SnapshotView::blocked` and +/// `SnapshotView::settings_hold`. fn upload_failure(halt: Halt) -> Option<&'static str> { match halt { // A cancel reports `UploadCancelled` from the facade that ordered it. - Halt::Blocked { .. } | Halt::Cancelled => None, + Halt::Blocked { .. } | Halt::HeldBySettings(_) | Halt::Cancelled => None, Halt::Unclassified => Some("the upload did not complete"), // Both charge the attempt budget; which one it is decides only what // exhausting that budget retires, not what the host is told. Halt::Attempt | Halt::UploadAttempt => { Some("the network refused it without a classification") } + Halt::HeadOversized => Some("the record this change publishes is over the size limit"), Halt::Permanent(DeadLetterReason::PayloadRefused) => { Some("the network refused the payload") } @@ -3190,12 +3321,49 @@ mod tests { ProviderError::NoVerdict, ProviderError::Rejected { status: 500 }, ProviderError::AddressMismatch, - ProviderError::InvalidCredential, ] { assert_eq!(classify_placement(answered), Halt::UploadAttempt); } } + #[test] + fn a_config_refused_before_the_request_holds_the_op_rather_than_spending_its_budget() { + for settings in [ + ProviderError::InvalidEndpoint, + ProviderError::InsecureTransport, + ProviderError::BlockedAddress, + ProviderError::InvalidCredential, + ] { + assert_eq!( + classify_placement(settings), + Halt::HeldBySettings(settings), + "{}", + settings.check(), + ); + } + } + + /// The hold's exit is the config, not a timer: it stands exactly while the + /// placement this session runs under still reaches the same verdict. + #[test] + fn a_settings_hold_lets_go_only_once_the_placement_stops_refusing() { + let refused = byo("file:///etc/passwd"); + assert_eq!( + placement_refusal(&Ok(Placement::External(refused.clone()))), + Some(ProviderError::InvalidEndpoint), + ); + for admitted in [ + Ok(Placement::External(byo("https://node.example"))), + Ok(Placement::Hosted), + // A dual write's mirror is best-effort, so no verdict on it holds + // the op that the hosted leg is already carrying. + Ok(Placement::Dual(refused)), + Err(crate::settings::PlacementRefusal::NoProvider), + ] { + assert_eq!(placement_refusal(&admitted), None); + } + } + /// The report names what the member must fix. A verdict reached before any /// request is built is their own settings, not their node. #[test] @@ -3405,10 +3573,12 @@ mod tests { } /// A trust refusal is charged so the queue stops at the budget instead of - /// spinning free; a refusal of the body *this pass* built is not, or the - /// reclassification would turn a rebasable failure into a permanent halt. + /// spinning free, and so is an over-length head, which no re-author can + /// shrink — but the two spend that budget differently, so the size refusal + /// keeps its own verdict. A refusal of the body *this pass* built is + /// charged neither way: a rebase onto other state may never build it again. #[test] - fn only_a_produce_side_trust_refusal_is_charged_against_the_attempt_budget() { + fn only_a_refusal_a_rebase_cannot_shed_is_charged_against_the_attempt_budget() { for (error, expected) in [ (AuthorError::GrantSectionOnChild, Halt::UploadAttempt), (AuthorError::MissingGrantSection, Halt::UploadAttempt), @@ -3422,7 +3592,7 @@ mod tests { ), ( AuthorError::HeadTooLarge { size: 2, limit: 1 }, - Halt::Unclassified, + Halt::HeadOversized, ), ] { let check = error.check(); diff --git a/crates/engine/src/sync/mod.rs b/crates/engine/src/sync/mod.rs index 2eae0cc9c2..3030d82f51 100644 --- a/crates/engine/src/sync/mod.rs +++ b/crates/engine/src/sync/mod.rs @@ -35,8 +35,8 @@ pub mod tick; pub use boot::{ColdStartError, ColdStartOutcome, ColdStartParams, RootResolve, cold_start}; pub use drain::{ - BlockedOp, DRAINED_OP_MARK_PREFIX, OP_ATTEMPTS_KEY, PUBLISHED_OP_MARK_PREFIX, UPLOAD_MARK_KEY, - owner_scoped_key, owner_tag, + BlockedOp, DRAINED_OP_MARK_PREFIX, OP_ATTEMPTS_KEY, PUBLISHED_OP_MARK_PREFIX, SettingsHold, + UPLOAD_MARK_KEY, owner_scoped_key, owner_tag, }; pub use model::{Link, NodeMeta, Snapshot, collation_key, suffix_name}; pub use op::{NewNode, Op, OpDecodeError, OpKind, Replaced, ScopeCrossing, StagedContent}; diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index cda2a777e2..05442ffdaa 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -61,10 +61,11 @@ use cipherbox_engine::testkit::{ }; use cipherbox_engine::{ 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, + DeadLetter, DeadLetterReason, DefaultsReason, Engine, EngineError, Entropy, EntropyError, + 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]; @@ -579,9 +580,16 @@ fn seed_account(world: &FakeWorld, blocks: &Blocks) -> IpnsName { } fn engine_on(device: &FakeDevice, entropy_seed: u64) -> (Engine, EventStream) { + engine_with(device, Box::new(SeededEntropy::new(entropy_seed))) +} + +fn engine_with( + device: &FakeDevice, + entropy: Box, +) -> (Engine, EventStream) { Engine::new( device.seam_set(), - Box::new(SeededEntropy::new(entropy_seed)), + entropy, SyncTimingProfile::CI, ContentProfile::CI, StoragePolicy::CI, @@ -720,15 +728,47 @@ fn boot( blocks: &Blocks, device: &FakeDevice, entropy_seed: u64, +) -> (Engine, EventStream, Vec) { + boot_with( + world, + blocks, + device, + Box::new(SeededEntropy::new(entropy_seed)), + ) +} + +/// The same, over a caller-supplied entropy source. +fn boot_with( + world: &FakeWorld, + blocks: &Blocks, + device: &FakeDevice, + entropy: Box, ) -> (Engine, EventStream, Vec) { serve_http(device, blocks, 400); - let (mut engine, events) = engine_on(device, entropy_seed); + let (mut engine, events) = engine_with(device, entropy); block_on(engine.start(secret())).expect("cold start adopts the owner root"); let mut tasks = world.scheduler.take_spawned_tasks(); poll_each(&mut tasks); (engine, events, tasks) } +/// A real seeded source that can be silenced mid-scenario: once armed it +/// reports success having written nothing, which is the seam failure every +/// fresh draw in the engine is required to refuse. +struct SilenceableEntropy { + inner: SeededEntropy, + silent: Arc, +} + +impl Entropy for SilenceableEntropy { + fn fill(&mut self, dest: &mut [u8]) -> Result<(), EntropyError> { + match self.silent.load(Ordering::Relaxed) { + true => Ok(()), + false => self.inner.fill(dest), + } + } +} + // --------------------------------------------------------------------------- // Record-plane inspection: what a node's published record actually carries. // --------------------------------------------------------------------------- @@ -2015,6 +2055,106 @@ fn a_registration_400_from_an_intermediary_is_charged_not_permanent() { ); } +/// A head over the block ceiling is refused identically on every retry: a fresh +/// nonce moves the sealed bytes and never their count, so no re-author shrinks +/// it. Uncharged it would hold the strict-FIFO queue head forever with nothing +/// reported anywhere, so it spends the budget and every op behind it drains — +/// but only the record was over the ceiling, so ending it keeps the version it +/// would have named rather than unpinning and erasing it, and owes back only the +/// child name no parent record ever reached. +#[test] +fn an_authored_head_over_the_block_ceiling_dead_letters_with_its_version_intact() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + + // A child ref carries its name verbatim, so a name past the 2 MiB IPFS + // block ceiling is a parent folder record this engine can author and its + // own ingress can never hold. + let name = "n".repeat(2 * 1024 * 1024 + 4096); + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: name.clone(), + }, + &(0..200u8).collect::>(), + ) + .expect("the write commits"); + let doomed = child_id(&engine, ROOT, &name); + + let (dead_letters, passes) = tick_until_dead_lettered(&world, &engine, &mut tasks); + assert!( + passes > 1, + "a size refusal is charged against the budget, not permanent on sight" + ); + assert_eq!( + dead_letters, + vec![DeadLetter { + op_id, + reason: DeadLetterReason::AttemptsExhausted + }] + ); + assert!( + block_on(StagingStore::queued_ops(&alice.staging_store)) + .unwrap() + .is_empty(), + "the head no retry could publish leaves the queue" + ); + assert_eq!( + retire_targets(&alice), + vec![write_name(doomed).as_str().to_owned()], + "the version the oversized record would have named stays pinned — unpinning \ + it is loss no retry undoes — while the child name the parent never came to \ + reference is owed back like any abandoned create's" + ); +} + +/// The drain's seal nonce is a fresh draw or nothing. A seam reporting success +/// having written nothing would seal every body on the engine's highest-volume +/// plane under one fixed nonce, and two seals under one key at one nonce is a +/// confidentiality break — so the pass halts with the op still queued. +#[test] +fn a_seam_that_draws_a_silent_nonce_publishes_no_record() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + let alice = world.device(b"alice"); + let silent = Arc::new(AtomicBool::new(false)); + let (mut engine, _events, mut tasks) = boot_with( + &world, + &blocks, + &alice, + Box::new(SilenceableEntropy { + inner: SeededEntropy::new(42), + silent: silent.clone(), + }), + ); + + block_on(engine.command(Command::Create { + parent: ROOT, + name: "photos".into(), + kind: NodeKind::Folder, + })) + .expect("the create queues"); + silent.store(true, Ordering::Relaxed); + tick(&world, &engine, &mut tasks); + + assert!( + published_names(&world.record_store, &blocks, ROOT).is_empty(), + "no record is sealed under a nonce the seam never wrote" + ); + assert_eq!( + block_on(StagingStore::queued_ops(&alice.staging_store)) + .unwrap() + .len(), + 1, + "the op keeps its place for a later draw" + ); +} + /// The `pushChunk` total is cross-checked against the `beginWrite` declaration: /// a backing file truncated mid-read fails the commit rather than publishing a /// short version as a success. diff --git a/crates/wasm/src/lib.rs b/crates/wasm/src/lib.rs index 3c916352fb..c41d15f278 100644 --- a/crates/wasm/src/lib.rs +++ b/crates/wasm/src/lib.rs @@ -580,6 +580,35 @@ impl BlockedOp { } } +/// The queue head held over the member's own provider settings, keeping its +/// place and its staging reservation until those settings change. +#[wasm_bindgen] +pub struct SettingsHold { + inner: facade::SettingsHold, +} + +#[wasm_bindgen] +impl SettingsHold { + /// The held op id (a `u64`, crossing as a `bigint`). + #[wasm_bindgen(getter, js_name = opId)] + pub fn op_id(&self) -> u64 { + self.inner.op_id.0 + } + + /// The 16 raw bytes of the node the held op targets. + #[wasm_bindgen(getter)] + pub fn node(&self) -> Vec { + self.inner.node.0.to_vec() + } + + /// The stable check name of the rule that refused the config. Never the + /// endpoint or the bearer those settings carry. + #[wasm_bindgen(getter)] + pub fn check(&self) -> String { + self.inner.refusal.check().to_owned() + } +} + /// A key-free snapshot of one folder for a host UI paint: children, breadcrumb /// trail, retained dead letters, and the staleness rung. #[wasm_bindgen] @@ -646,6 +675,12 @@ impl SnapshotView { self.inner.blocked.map(|inner| BlockedOp { inner }) } + /// The drain's settings-refused hold, or `undefined`. + #[wasm_bindgen(getter, js_name = settingsHold)] + pub fn settings_hold(&self) -> Option { + self.inner.settings_hold.map(|inner| SettingsHold { inner }) + } + /// Durable queue entries this session holds but cannot read — another /// identity's, or written by a newer build. A host reports these instead of /// leaving an over-budget rejection unexplained on a vault that looks empty. @@ -1203,6 +1238,11 @@ mod tests { node: facade::NodeId([5u8; 16]), needed_bytes: 4096, }), + settings_hold: Some(facade::SettingsHold { + op_id: OpId(13), + node: facade::NodeId([6u8; 16]), + refusal: cipherbox_engine::ProviderError::InsecureTransport, + }), retained_records: 3, staleness: facade::Staleness::Reconciling, }); @@ -1225,6 +1265,10 @@ mod tests { assert_eq!(blocked.op_id(), 12); assert_eq!(blocked.node(), vec![5u8; 16]); assert_eq!(blocked.needed_bytes(), 4096); + let held = view.settings_hold().expect("the view carries the hold"); + assert_eq!(held.op_id(), 13); + assert_eq!(held.node(), vec![6u8; 16]); + assert_eq!(held.check(), "byo-endpoint-insecure"); assert_eq!(view.retained_records(), 3); assert_eq!(view.staleness(), Staleness::Reconciling); diff --git a/crates/wasm/tests/boundary.rs b/crates/wasm/tests/boundary.rs index 09ab6a98c5..a9c8a9b00f 100644 --- a/crates/wasm/tests/boundary.rs +++ b/crates/wasm/tests/boundary.rs @@ -258,6 +258,11 @@ fn snapshot_view_getters_cross_with_boundary_shapes() { node: facade::NodeId([6u8; 16]), needed_bytes: u64::MAX, }), + settings_hold: Some(facade::SettingsHold { + op_id: OpId(13), + node: facade::NodeId([7u8; 16]), + refusal: cipherbox_engine::ProviderError::BlockedAddress, + }), retained_records: 0, staleness: facade::Staleness::Fresh, }) @@ -337,6 +342,22 @@ fn snapshot_view_getters_cross_with_boundary_shapes() { vec![6u8; 16] ); + let held = get(&view, "settingsHold"); + assert_eq!( + get(&held, "opId").js_typeof(), + JsValue::from_str("bigint"), + "a held op's opId must cross as a JS bigint, never a number" + ); + assert_eq!( + get(&held, "node").unchecked_into::().to_vec(), + vec![7u8; 16] + ); + assert_eq!( + get(&held, "check"), + JsValue::from_str("byo-endpoint-blocked"), + "the refusing rule crosses by its stable check name" + ); + let children = get(&view, "children"); assert!(children.is_instance_of::()); let children = children.unchecked_into::();