diff --git a/CHANGELOG.md b/CHANGELOG.md index 5dc475f7..298f3647 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,7 +38,7 @@ - `IngressEnvelope::to_retained_bytes_v2` is now `to_retained_bytes` (#880). It writes `EINGR002`, byte for byte as before, for an envelope without a Continuum occurrence, and `EINGR003` for one with an occurrence. - `from_retained_bytes` reads `EINGR001` to `EINGR003`. + `from_retained_bytes` reads `EINGR001` to `EINGR003` (and, from #885, `EINGR004`). - Commit hash v3 binds a tick's receipt digest, and through it the tick's attribution section, into the tick's commit identity (#873). Every new scheduler tick's receipt is commit-bound (receipt format 4, retained as @@ -105,6 +105,18 @@ ### Added +- Causal intents get an identity recipe that a plain intent cannot alias + (#885). Under the original recipe a plain intent could reproduce a causal + intent's hash input exactly and take its ingress and submission IDs. New + causal intents are identified by `"echo.ingress.causal.v3\0" || ...`, record + `IngressIdentityVersion::SeparatedCausal`, and are retained as `EINGR004`. + `EINGR001` to `EINGR003` decode exactly as before, as `Original`. A resend of + a causal request accepted before this change resolves to its original + submission when its retained envelope states the same request, and is staged + with that envelope; an alias is not a match. Missing legacy material is + `OriginalIngressIdentityUnverifiable`, a request with both identities is + `IngressIdentityAmbiguous`, and new admission in the original causal recipe + is `OriginalCausalIdentityNotAdmissible`. - A strand inherits the Continuum occurrences decided in its causal past (#886). Redelivering such an occurrence to the strand returns `AlreadyPresentInInheritedHistory` (from `submit_intent` and `ingest`) diff --git a/crates/warp-core/src/contract_obstruction.rs b/crates/warp-core/src/contract_obstruction.rs index eb8f6439..af9542b2 100644 --- a/crates/warp-core/src/contract_obstruction.rs +++ b/crates/warp-core/src/contract_obstruction.rs @@ -228,7 +228,17 @@ impl ContractObstruction { } => Self::admission_obstruction(ContractObstructionSubject::Submission { submission_id: *existing_submission_id, }), - RuntimeError::AdmissionCoverageUnestablished { .. } => Self::new( + RuntimeError::IngressIdentityAlias { submission_id } => { + Self::admission_obstruction(ContractObstructionSubject::Submission { + submission_id: *submission_id, + }) + } + RuntimeError::IngressIdentityAmbiguous { .. } + | RuntimeError::OriginalCausalIdentityNotAdmissible { .. } => { + Self::admission_obstruction(ContractObstructionSubject::Unspecified) + } + RuntimeError::OriginalIngressIdentityUnverifiable { .. } + | RuntimeError::AdmissionCoverageUnestablished { .. } => Self::new( ContractObstructionKind::MissingRetention, ContractObstructionSubject::Unspecified, ), diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index 1544b4a1..1ca170af 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -290,6 +290,54 @@ pub enum RuntimeError { /// Frontier tick of the restored basis. basis_tick: WorldlineTick, }, + /// A causal request exists both as a submission identified by the + /// original recipe and as one identified by the separated recipe. + /// + /// Echo does not choose between them silently. + #[error( + "causal request has two submissions: {original_submission_id:?} (original identity) and {separated_submission_id:?} (separated identity)" + )] + IngressIdentityAmbiguous { + /// Submission identified by the original recipe. + original_submission_id: Hash, + /// Submission identified by the separated causal recipe. + separated_submission_id: Hash, + }, + /// A request accepted under the original identity recipe may exist, but + /// Echo lacks the retained material to tell whether this is its resend. + /// + /// Absence of evidence never authorizes a second act. + #[error( + "cannot verify whether {head_key:?} already accepted this request under original ingress id {original_ingress_id:?}" + )] + OriginalIngressIdentityUnverifiable { + /// Resolved writer head. + head_key: WriterHeadKey, + /// Ingress id the request has under the original recipe. + original_ingress_id: Hash, + }, + /// The presented request has the ingress id of a retained submission that + /// states a different request: an alias under the original causal recipe. + /// + /// Echo neither treats it as a retry nor overwrites the retained request. + #[error( + "request aliases retained submission {submission_id:?} but states a different request" + )] + IngressIdentityAlias { + /// Retained submission whose ingress id the request shares. + submission_id: Hash, + }, + /// A causal request identified by the original recipe can only name an + /// existing submission; new causal requests use the separated recipe. + #[error( + "causal request {ingress_id:?} on {head_key:?} uses the original identity recipe and names no existing submission" + )] + OriginalCausalIdentityNotAdmissible { + /// Resolved writer head. + head_key: WriterHeadKey, + /// Original-recipe ingress id that was presented. + ingress_id: Hash, + }, /// Ticketed runtime ingress referenced a submission Echo has not witnessed. #[error("unknown intent submission: {0:?}")] UnknownIntentSubmission(Hash), @@ -3375,7 +3423,16 @@ impl WorldlineRuntime { let retained_envelope = envelope.clone(); let head_key = self.resolve_target(envelope.target())?; self.check_admission_coverage(head_key.worldline_id)?; + if let Some(record) = self.original_identity_submission(head_key, &envelope)? { + return Ok(IntentSubmissionDisposition::Duplicate { + ingress_id: record.ingress_id, + head_key, + submission_id: record.submission_id, + submission_generation: record.submission_generation, + }); + } self.check_continuum_occurrence_binding(head_key, &envelope)?; + self.refuse_retained_alias(head_key, &envelope)?; if let Some(inherited) = self.inherited_continuum_occurrence(head_key, &envelope)? { return Ok( IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { @@ -3425,6 +3482,7 @@ impl WorldlineRuntime { }); } + refuse_new_original_causal_identity(head_key, &retained_envelope)?; let record = self.record_witnessed_submission(head_key, ingress_id)?; self.bind_continuum_occurrence(&record, &retained_envelope); self.witnessed_submission_envelopes @@ -3489,6 +3547,21 @@ impl WorldlineRuntime { let Some(submission) = self.witnessed_submissions.get(&submission_id) else { return Err(RuntimeError::UnknownIntentSubmission(submission_id)); }; + // A submission accepted under the original recipe is staged with its + // retained envelope, never with a separated form whose identity + // differs from the submission's. + let envelope = match ( + self.witnessed_submission_envelopes.get(&submission_id), + envelope.original_identity_form(), + ) { + (Some(retained), Some(original)) + if retained.ingress_id() == submission.ingress_id + && retained.same_request_as(&original) => + { + retained.clone() + } + _ => envelope, + }; let ingress_id = envelope.ingress_id(); if submission.ingress_id != ingress_id || submission.head_key != head_key { return Err(RuntimeError::TicketedIngressSubmissionMismatch( @@ -3671,10 +3744,25 @@ impl WorldlineRuntime { envelope: IngressEnvelope, ) -> Result { let ingress_id = envelope.ingress_id(); - let retained_envelope = envelope.clone(); let head_key = self.resolve_target(envelope.target())?; self.check_admission_coverage(head_key.worldline_id)?; + // A resend of a request accepted under the original recipe continues + // as that submission, with its retained envelope and identity. + let (envelope, ingress_id) = match self.original_identity_submission(head_key, &envelope)? { + Some(record) => ( + self.witnessed_submission_envelopes + .get(&record.submission_id) + .cloned() + .ok_or(RuntimeError::WitnessedSubmissionEnvelopeUnavailable( + record.submission_id, + ))?, + record.ingress_id, + ), + None => (envelope, ingress_id), + }; + let retained_envelope = envelope.clone(); self.check_continuum_occurrence_binding(head_key, &envelope)?; + self.refuse_retained_alias(head_key, &envelope)?; if let Some(inherited) = self.inherited_continuum_occurrence(head_key, &envelope)? { return Ok(IngressDisposition::AlreadyPresentInInheritedHistory { ingress_id, @@ -3701,6 +3789,12 @@ impl WorldlineRuntime { }); } + if !self + .submission_by_target + .contains_key(&(head_key, ingress_id)) + { + refuse_new_original_causal_identity(head_key, &envelope)?; + } let outcome = self .heads .inbox_mut(&head_key) @@ -3890,10 +3984,7 @@ impl WorldlineRuntime { .witnessed_submissions .get(&submission_id) .zip(self.witnessed_submission_envelopes.get(&submission_id)) - .filter(|(record, existing)| { - record.ingress_id == envelope.ingress_id() - && existing.occurrence() == Some(occurrence) - }) + .filter(|(_, existing)| is_same_request(existing, envelope)) .map(|(record, _)| record.clone()); let Some(record) = same_binding else { return Err(RuntimeError::ContinuumOccurrenceConflict { @@ -3937,6 +4028,107 @@ impl WorldlineRuntime { Ok(None) } + /// Finds the submission of a causal request accepted under the original + /// identity recipe, when `envelope` is that request's separated form. + /// + /// Exact legacy matches take precedence over a new separated submission, + /// so a resend does not become a second act. The retained request must + /// state the same kind, bytes, parents, occurrence and head: the original + /// recipe can be aliased by a plain intent, and an alias is not a match. + /// Missing retained material obstructs instead of admitting. When both a + /// legacy and a separated submission exist, the request is ambiguous. + fn original_identity_submission( + &self, + head_key: WriterHeadKey, + envelope: &IngressEnvelope, + ) -> Result, RuntimeError> { + let Some(original) = envelope.original_identity_form() else { + return Ok(None); + }; + let original_ingress_id = original.ingress_id(); + let committed = self + .worldlines + .get(&head_key.worldline_id) + .is_some_and(|frontier| { + frontier + .state() + .contains_committed_ingress(&head_key, &original_ingress_id) + }); + let Some(record) = self + .submission_by_target + .get(&(head_key, original_ingress_id)) + .and_then(|submission_id| self.witnessed_submissions.get(submission_id)) + else { + if committed { + return Err(RuntimeError::OriginalIngressIdentityUnverifiable { + head_key, + original_ingress_id, + }); + } + return Ok(None); + }; + let Some(retained) = self + .witnessed_submission_envelopes + .get(&record.submission_id) + else { + return Err(RuntimeError::OriginalIngressIdentityUnverifiable { + head_key, + original_ingress_id, + }); + }; + if !retained.same_request_as(&original) + || self.resolve_target(retained.target()).ok() != Some(head_key) + { + return Ok(None); + } + let separated_ingress_id = envelope.ingress_id(); + if let Some(separated_submission_id) = self + .submission_by_target + .get(&(head_key, separated_ingress_id)) + .copied() + .or_else(|| { + self.worldlines + .get(&head_key.worldline_id) + .is_some_and(|frontier| { + frontier + .state() + .contains_committed_ingress(&head_key, &separated_ingress_id) + }) + .then(|| derive_intent_submission_id(head_key, separated_ingress_id)) + }) + { + return Err(RuntimeError::IngressIdentityAmbiguous { + original_submission_id: record.submission_id, + separated_submission_id, + }); + } + Ok(Some(record.clone())) + } + + /// Refuses a request whose ingress id names a retained submission that + /// states a different request. Only the original causal recipe can alias + /// (#885); an alias is neither a retry nor a reason to overwrite the + /// retained request. + fn refuse_retained_alias( + &self, + head_key: WriterHeadKey, + envelope: &IngressEnvelope, + ) -> Result<(), RuntimeError> { + let Some(submission_id) = self + .submission_by_target + .get(&(head_key, envelope.ingress_id())) + .copied() + else { + return Ok(()); + }; + match self.witnessed_submission_envelopes.get(&submission_id) { + Some(retained) if !retained.same_request_as(envelope) => { + Err(RuntimeError::IngressIdentityAlias { submission_id }) + } + _ => Ok(()), + } + } + /// Applies Continuum's occurrence binding law at admission. /// /// An envelope without an occurrence, or with an occurrence this worldline @@ -3979,9 +4171,7 @@ impl WorldlineRuntime { .get(&existing_submission_id), ) .is_some_and(|(record, existing)| { - record.head_key == head_key - && record.ingress_id == envelope.ingress_id() - && existing.occurrence() == Some(occurrence) + record.head_key == head_key && is_same_request(existing, envelope) }); if retry { Ok(()) @@ -4362,6 +4552,32 @@ fn attributed_outcome_decision( }) } +/// Refuses to create a submission for a causal request identified by the +/// original recipe: only resends of retained requests may use it. +fn refuse_new_original_causal_identity( + head_key: WriterHeadKey, + envelope: &IngressEnvelope, +) -> Result<(), RuntimeError> { + if envelope.identity_version() == crate::IngressIdentityVersion::Original + && !envelope.causal_parents().is_empty() + { + return Err(RuntimeError::OriginalCausalIdentityNotAdmissible { + head_key, + ingress_id: envelope.ingress_id(), + }); + } + Ok(()) +} + +/// Whether `presented` restates the retained request `existing`, directly or +/// as the separated form of a request retained under the original recipe. +fn is_same_request(existing: &IngressEnvelope, presented: &IngressEnvelope) -> bool { + existing.same_request_as(presented) + || presented + .original_identity_form() + .is_some_and(|original| existing.same_request_as(&original)) +} + fn derive_intent_submission_id(head_key: WriterHeadKey, ingress_id: Hash) -> Hash { let mut hasher = blake3::Hasher::new(); hasher.update(b"echo.intent-submission.v0"); @@ -4478,6 +4694,10 @@ fn scheduler_fault_scope_for_error( | RuntimeError::ContinuumOccurrenceConflict { .. } | RuntimeError::ContinuumOccurrenceCoverageUnavailable { .. } | RuntimeError::AdmissionCoverageUnestablished { .. } + | RuntimeError::IngressIdentityAmbiguous { .. } + | RuntimeError::OriginalIngressIdentityUnverifiable { .. } + | RuntimeError::OriginalCausalIdentityNotAdmissible { .. } + | RuntimeError::IngressIdentityAlias { .. } | RuntimeError::UnknownIntentSubmission(_) | RuntimeError::TicketedIngressSubmissionMismatch(_) | RuntimeError::TicketedIngressAlreadyStaged(_) @@ -5265,6 +5485,36 @@ fn scheduler_error_cause_digest(err: &RuntimeError) -> Hash { hasher.update(worldline_id.as_bytes()); hash_worldline_tick(&mut hasher, *basis_tick); } + RuntimeError::IngressIdentityAmbiguous { + original_submission_id, + separated_submission_id, + } => { + hasher.update(b"ingress-identity-ambiguous"); + hasher.update(original_submission_id); + hasher.update(separated_submission_id); + } + RuntimeError::OriginalIngressIdentityUnverifiable { + head_key, + original_ingress_id, + } => { + hasher.update(b"original-ingress-identity-unverifiable"); + hasher.update(head_key.worldline_id.as_bytes()); + hasher.update(head_key.head_id.as_bytes()); + hasher.update(original_ingress_id); + } + RuntimeError::IngressIdentityAlias { submission_id } => { + hasher.update(b"ingress-identity-alias"); + hasher.update(submission_id); + } + RuntimeError::OriginalCausalIdentityNotAdmissible { + head_key, + ingress_id, + } => { + hasher.update(b"original-causal-identity-not-admissible"); + hasher.update(head_key.worldline_id.as_bytes()); + hasher.update(head_key.head_id.as_bytes()); + hasher.update(ingress_id); + } RuntimeError::UnknownIntentSubmission(submission_id) => { hasher.update(b"unknown-intent-submission"); hasher.update(submission_id); @@ -6920,6 +7170,266 @@ mod tests { accepted_submission(runtime.ingest(occurred(early, 4, b"four")).unwrap()); } + /// Retains `envelope` as an accepted submission of generation `generation`, + /// as restore does for a request accepted by an earlier writer. + fn retain_submission( + runtime: &mut WorldlineRuntime, + envelope: IngressEnvelope, + generation: u64, + ) -> Hash { + let head_key = runtime.resolve_target(envelope.target()).unwrap(); + let submission_id = derive_intent_submission_id(head_key, envelope.ingress_id()); + runtime + .restore_witnessed_submission_persistence(WitnessedSubmissionPersistenceSnapshot::new( + vec![WitnessedSubmissionPersistenceRecord { + submission: IntentSubmissionRecord { + submission_id, + ingress_id: envelope.ingress_id(), + head_key, + submission_generation: IngressSubmissionGeneration::from_raw(generation), + }, + envelope, + }], + )) + .unwrap(); + submission_id + } + + fn causal_request(worldline_id: WorldlineId, parent_seed: u8) -> IngressEnvelope { + IngressEnvelope::local_intent_with_causal_parents( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/causal"), + b"causal payload".to_vec(), + vec![crate::IngressCausalParent::TickReceipt { + receipt_ref: CausalTickReceiptRef { + worldline_id, + worldline_tick_after: wt(1), + commit_global_tick: gt(1), + commit_hash: hash(parent_seed), + submission_id: hash(parent_seed.wrapping_add(1)), + ticket_digest: hash(parent_seed.wrapping_add(2)), + receipt_content_digest: hash(parent_seed.wrapping_add(3)), + }, + }], + ) + } + + /// Issue #885: a resend of a causal request accepted under the original + /// recipe resolves to that submission, and staging uses its retained + /// envelope. A plain alias of the original recipe, or a request with a + /// different parent, is not a match and is admitted under the separated + /// recipe. New requests cannot use the original causal recipe. + #[test] + fn a_causal_resend_resolves_to_its_original_identity_submission() { + let (mut runtime, _engine, worldline_id) = toy_contract_runtime(); + let separated = causal_request(worldline_id, 40); + let original = separated.original_identity_form().unwrap(); + assert_ne!(original.ingress_id(), separated.ingress_id()); + + // An earlier writer could create the original form; this one cannot. + assert!(matches!( + runtime.submit_intent(original.clone()), + Err(RuntimeError::OriginalCausalIdentityNotAdmissible { .. }) + )); + assert!(matches!( + runtime.ingest(original.clone()), + Err(RuntimeError::OriginalCausalIdentityNotAdmissible { .. }) + )); + let legacy = retain_submission(&mut runtime, original.clone(), 1); + + // Genuine resend: the retained submission, nothing new witnessed. + match runtime.submit_intent(separated.clone()).unwrap() { + IntentSubmissionDisposition::Duplicate { + submission_id, + ingress_id, + .. + } => { + assert_eq!(submission_id, legacy); + assert_eq!(ingress_id, original.ingress_id()); + } + other => panic!("a resend must resolve to the retained submission, got {other:?}"), + } + assert_eq!(runtime.witnessed_submission_count(), 1); + // A resend of the original form itself is an ordinary retry. + assert!(matches!( + runtime.submit_intent(original.clone()).unwrap(), + IntentSubmissionDisposition::Duplicate { submission_id, .. } if submission_id == legacy + )); + + // Staging with the separated form stages the retained request. + let authority = TicketedRuntimeIngressAuthority::assume_runtime_owner(); + match runtime + .ingest_ticketed_invocation(&authority, legacy, &test_ticket(21), separated) + .unwrap() + { + TicketedRuntimeIngressDisposition::Staged { record, .. } => { + assert_eq!(record.ingress_id, original.ingress_id()); + } + TicketedRuntimeIngressDisposition::Duplicate { .. } => panic!("first staging"), + } + + // A different parent is a different request. + accepted_intent( + runtime + .submit_intent(causal_request(worldline_id, 50)) + .unwrap(), + ); + } + + #[test] + fn a_plain_alias_of_the_original_recipe_is_not_a_resend() { + let (mut runtime, _engine, worldline_id) = toy_contract_runtime(); + let separated = causal_request(worldline_id, 40); + let original = separated.original_identity_form().unwrap(); + let IngressPayload::LocalIntent { + intent_kind, + intent_bytes, + } = separated.payload(); + let parent = separated.causal_parents()[0].receipt_ref(); + let mut alias_kind = [0u8; 32]; + alias_kind[..10].copy_from_slice(b"causal:v2\0"); + alias_kind[10..].copy_from_slice(&intent_kind.as_hash()[..22]); + let mut alias_bytes = intent_kind.as_hash()[22..].to_vec(); + alias_bytes.extend_from_slice(&(intent_bytes.len() as u64).to_le_bytes()); + alias_bytes.extend_from_slice(intent_bytes); + alias_bytes.extend_from_slice(&1u64.to_le_bytes()); + alias_bytes.extend_from_slice(b"tick-receipt\0"); + alias_bytes.extend_from_slice(&parent.to_canonical_bytes()); + let alias = IngressEnvelope::local_intent( + IngressTarget::DefaultWriter { worldline_id }, + crate::IntentKind::from_hash(alias_kind), + alias_bytes, + ); + assert_eq!(alias.ingress_id(), original.ingress_id()); + + let planted = accepted_intent(runtime.submit_intent(alias).unwrap()); + let genuine = accepted_intent(runtime.submit_intent(separated).unwrap()); + assert_ne!( + genuine, planted, + "the alias neither counts as a retry nor reserves the request" + ); + } + + /// PR #900 review: an ingress id shared by two different requests (an + /// alias under the original causal recipe) is neither a retry nor a + /// reason to overwrite the retained request, in either direction. + #[test] + fn a_retained_alias_is_refused_and_never_overwritten() { + let (mut runtime, _engine, worldline_id) = toy_contract_runtime(); + let separated = causal_request(worldline_id, 40); + let original = separated.original_identity_form().unwrap(); + let IngressPayload::LocalIntent { + intent_kind, + intent_bytes, + } = separated.payload(); + let parent = separated.causal_parents()[0].receipt_ref(); + let mut alias_kind = [0u8; 32]; + alias_kind[..10].copy_from_slice(b"causal:v2\0"); + alias_kind[10..].copy_from_slice(&intent_kind.as_hash()[..22]); + let mut alias_bytes = intent_kind.as_hash()[22..].to_vec(); + alias_bytes.extend_from_slice(&(intent_bytes.len() as u64).to_le_bytes()); + alias_bytes.extend_from_slice(intent_bytes); + alias_bytes.extend_from_slice(&1u64.to_le_bytes()); + alias_bytes.extend_from_slice(b"tick-receipt\0"); + alias_bytes.extend_from_slice(&parent.to_canonical_bytes()); + let alias = IngressEnvelope::local_intent( + IngressTarget::DefaultWriter { worldline_id }, + crate::IntentKind::from_hash(alias_kind), + alias_bytes, + ); + + // A planted plain alias, then the original causal form through ingest. + let planted = accepted_intent(runtime.submit_intent(alias.clone()).unwrap()); + assert!(matches!( + runtime.ingest(original.clone()), + Err(RuntimeError::IngressIdentityAlias { submission_id }) if submission_id == planted + )); + assert_eq!( + runtime.witnessed_submission_envelope(&planted), + Some(&alias) + ); + + // A retained legacy causal request, then its plain alias. + let (mut runtime, _engine, _) = toy_contract_runtime(); + let legacy = retain_submission(&mut runtime, original.clone(), 1); + assert!(matches!( + runtime.submit_intent(alias.clone()), + Err(RuntimeError::IngressIdentityAlias { submission_id }) if submission_id == legacy + )); + assert!(matches!( + runtime.ingest(alias), + Err(RuntimeError::IngressIdentityAlias { submission_id }) if submission_id == legacy + )); + assert_eq!( + runtime.witnessed_submission_envelope(&legacy), + Some(&original) + ); + } + + #[test] + fn original_identity_resends_obstruct_on_ambiguity_or_missing_material() { + // Both identities of one request retained: Echo will not choose. + let (mut runtime, _engine, worldline_id) = toy_contract_runtime(); + let separated = causal_request(worldline_id, 40); + let original = separated.original_identity_form().unwrap(); + let legacy = retain_submission(&mut runtime, original.clone(), 1); + let current = retain_submission(&mut runtime, separated.clone(), 2); + assert!(matches!( + runtime.submit_intent(separated.clone()), + Err(RuntimeError::IngressIdentityAmbiguous { + original_submission_id, + separated_submission_id, + .. + }) if original_submission_id == legacy && separated_submission_id == current + )); + + // A legacy submission without retained material cannot be compared. + let (mut replayed, _engine, _) = toy_contract_runtime(); + let head_key = replayed + .resolve_target(&IngressTarget::DefaultWriter { worldline_id }) + .unwrap(); + replayed + .replay_witnessed_submissions([IntentSubmissionRecord { + submission_id: legacy, + ingress_id: original.ingress_id(), + head_key, + submission_generation: IngressSubmissionGeneration::from_raw(1), + }]) + .unwrap(); + assert!(matches!( + replayed.submit_intent(separated), + Err(RuntimeError::OriginalIngressIdentityUnverifiable { original_ingress_id, .. }) + if original_ingress_id == original.ingress_id() + )); + assert_eq!(replayed.witnessed_submission_count(), 1); + } + + #[test] + fn a_causal_occurrence_resend_keeps_its_original_binding() { + let (mut runtime, _engine, worldline_id) = toy_contract_runtime(); + let occurred = |applied: Hash| { + IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("test/causal"), + b"causal payload".to_vec(), + causal_request(worldline_id, 40).causal_parents().to_vec(), + crate::ContinuumOccurrence::new([9; 32], applied), + ) + }; + let separated = occurred([0xa1; 32]); + let original = separated.original_identity_form().unwrap(); + let legacy = retain_submission(&mut runtime, original, 1); + assert!(matches!( + runtime.submit_intent(separated).unwrap(), + IntentSubmissionDisposition::Duplicate { submission_id, .. } if submission_id == legacy + )); + assert!(matches!( + runtime.submit_intent(occurred([0xb2; 32])), + Err(RuntimeError::ContinuumOccurrenceConflict { existing_submission_id, .. }) + if existing_submission_id == legacy + )); + } + /// Owner conditions for #890: an inherited occurrence keeps resolving to /// its decision at the fork basis however the parent advances; a parent /// decision made after the fork never leaks into the strand, even for a diff --git a/crates/warp-core/src/head_inbox.rs b/crates/warp-core/src/head_inbox.rs index 0d981d0b..8a1f6c5e 100644 --- a/crates/warp-core/src/head_inbox.rs +++ b/crates/warp-core/src/head_inbox.rs @@ -26,6 +26,13 @@ const RETAINED_INGRESS_ENVELOPE_MAGIC_V1: &[u8; 8] = b"EINGR001"; const RETAINED_INGRESS_ENVELOPE_MAGIC_V2: &[u8; 8] = b"EINGR002"; /// Retained envelope format that also carries a Continuum occurrence. const RETAINED_INGRESS_ENVELOPE_MAGIC_V3: &[u8; 8] = b"EINGR003"; +/// Retained envelope format for causal intents identified by the separated +/// causal recipe ([`IngressIdentityVersion::SeparatedCausal`]). +const RETAINED_INGRESS_ENVELOPE_MAGIC_V4: &[u8; 8] = b"EINGR004"; +/// Ingress id domain of the separated causal recipe. +/// +/// Like the occurrence domain, it must not begin with `"ingress:"`. +const SEPARATED_CAUSAL_INGRESS_DOMAIN: &[u8] = b"echo.ingress.causal.v3\0"; /// Ingress id domain for a Continuum occurrence of content-addressed intent. /// /// It must not begin with `"ingress:"`; see [`compute_ingress_id`]. @@ -283,6 +290,32 @@ fn lower_hex(bytes: &Hash) -> [u8; 64] { out } +// ============================================================================= +// IngressIdentityVersion +// ============================================================================= + +/// Which identity recipe an envelope's ingress id uses. +/// +/// The original recipes hash a parentless intent as `"ingress:" || kind || +/// bytes` and a causal intent as `"ingress:causal:v2\0" || kind || ...`. The +/// second begins with the first and the kind is any 32 bytes, so a plain intent +/// can alias a causal intent's hash input exactly (#885). New causal intents +/// therefore use a separated recipe whose domain cannot alias plain input. +/// +/// The version is retained with the envelope so stored requests keep their +/// exact identity. It only matters for causal intents: an envelope without +/// causal parents uses the unchanged plain recipe and is always `Original`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum IngressIdentityVersion { + /// The recipes in use before #885, kept for retained requests. Retention + /// formats `EINGR001`, `EINGR002` and `EINGR003` decode to this version. + Original, + /// The separated causal recipe, `"echo.ingress.causal.v3\0" || ...`, + /// used by every newly constructed causal intent and retained as + /// `EINGR004`. + SeparatedCausal, +} + // ============================================================================= // IngressEnvelope // ============================================================================= @@ -307,6 +340,8 @@ pub struct IngressEnvelope { payload: IngressPayload, /// Continuum occurrence this envelope admits, if it came from Continuum. occurrence: Option, + /// Identity recipe of `ingress_id`; `Original` when there are no parents. + identity_version: IngressIdentityVersion, } impl IngressEnvelope { @@ -332,7 +367,14 @@ impl IngressEnvelope { intent_bytes: Vec, causal_parents: Vec, ) -> Self { - Self::build(target, intent_kind, intent_bytes, causal_parents, None) + Self::build( + target, + intent_kind, + intent_bytes, + causal_parents, + None, + IngressIdentityVersion::SeparatedCausal, + ) } /// Creates an intent that admits one Continuum occurrence. @@ -357,6 +399,7 @@ impl IngressEnvelope { intent_bytes, causal_parents, Some(occurrence), + IngressIdentityVersion::SeparatedCausal, ) } @@ -366,11 +409,22 @@ impl IngressEnvelope { intent_bytes: Vec, mut causal_parents: Vec, occurrence: Option, + identity_version: IngressIdentityVersion, ) -> Self { causal_parents.sort_unstable(); causal_parents.dedup(); - let ingress_id = - compute_ingress_id(&intent_kind, &intent_bytes, &causal_parents, occurrence); + let identity_version = if causal_parents.is_empty() { + IngressIdentityVersion::Original + } else { + identity_version + }; + let ingress_id = compute_ingress_id( + &intent_kind, + &intent_bytes, + &causal_parents, + occurrence, + identity_version, + ); Self { ingress_id, target, @@ -380,9 +434,52 @@ impl IngressEnvelope { intent_bytes, }, occurrence, + identity_version, } } + /// Returns the identity recipe of this envelope's ingress id. + #[must_use] + pub fn identity_version(&self) -> IngressIdentityVersion { + self.identity_version + } + + /// Returns this request as identified by the original recipe, when it is + /// a causal request identified by the separated recipe. + /// + /// Echo uses it to recognise a resend of a request accepted before #885; + /// a match must still compare the retained request, because the original + /// recipe can be aliased. + #[must_use] + pub(crate) fn original_identity_form(&self) -> Option { + if self.identity_version != IngressIdentityVersion::SeparatedCausal { + return None; + } + let IngressPayload::LocalIntent { + intent_kind, + intent_bytes, + } = &self.payload; + Some(Self::build( + self.target.clone(), + *intent_kind, + intent_bytes.clone(), + self.causal_parents.clone(), + self.occurrence, + IngressIdentityVersion::Original, + )) + } + + /// Whether two envelopes state the same request: kind, bytes, causal + /// parents, occurrence (both digests) and identity version. The routing + /// target is compared by the caller after resolution. + #[must_use] + pub(crate) fn same_request_as(&self, other: &Self) -> bool { + self.payload == other.payload + && self.causal_parents == other.causal_parents + && self.occurrence == other.occurrence + && self.identity_version == other.identity_version + } + /// Returns the canonical content address of this envelope. #[must_use] pub fn ingress_id(&self) -> Hash { @@ -404,7 +501,13 @@ impl IngressEnvelope { IngressPayload::LocalIntent { intent_kind, intent_bytes, - } => compute_ingress_id(intent_kind, intent_bytes, &self.causal_parents, None), + } => compute_ingress_id( + intent_kind, + intent_bytes, + &self.causal_parents, + None, + self.identity_version, + ), } } @@ -452,8 +555,11 @@ impl IngressEnvelope { /// applied digest. #[must_use] pub fn to_retained_bytes(&self) -> Vec { + let separated = self.identity_version == IngressIdentityVersion::SeparatedCausal; let mut out = Vec::new(); - out.extend_from_slice(if self.occurrence.is_some() { + out.extend_from_slice(if separated { + RETAINED_INGRESS_ENVELOPE_MAGIC_V4 + } else if self.occurrence.is_some() { RETAINED_INGRESS_ENVELOPE_MAGIC_V3 } else { RETAINED_INGRESS_ENVELOPE_MAGIC_V2 @@ -502,6 +608,9 @@ impl IngressEnvelope { out.extend_from_slice(intent_bytes); } } + if separated { + out.push(u8::from(self.occurrence.is_some())); + } if let Some(occurrence) = &self.occurrence { out.extend_from_slice(&occurrence.occurrence_key_digest); out.extend_from_slice(&occurrence.applied_digest); @@ -519,6 +628,9 @@ impl IngressEnvelope { let magic = bytes .get(..RETAINED_INGRESS_ENVELOPE_MAGIC_V2.len()) .ok_or(IngressEnvelopeDecodeError::UnexpectedEof)?; + if magic == RETAINED_INGRESS_ENVELOPE_MAGIC_V4 { + return Self::from_retained_bytes_v2_or_v3(bytes, RETAINED_INGRESS_ENVELOPE_MAGIC_V4); + } if magic == RETAINED_INGRESS_ENVELOPE_MAGIC_V3 { return Self::from_retained_bytes_v2_or_v3(bytes, RETAINED_INGRESS_ENVELOPE_MAGIC_V3); } @@ -531,7 +643,8 @@ impl IngressEnvelope { Err(IngressEnvelopeDecodeError::InvalidMagic) } - /// Decodes `EINGR002`, or `EINGR003`, which appends a Continuum occurrence. + /// Decodes `EINGR002`; `EINGR003`, which appends a Continuum occurrence; + /// or `EINGR004`, a separated-causal request with an occurrence flag. fn from_retained_bytes_v2_or_v3( bytes: &[u8], magic: &[u8; 8], @@ -594,7 +707,21 @@ impl IngressEnvelope { let intent_kind = IntentKind::from_hash(cursor.read_hash()?); let intent_len = cursor.read_len()?; let intent_bytes = cursor.read_exact(intent_len)?.to_vec(); - let occurrence = if magic == RETAINED_INGRESS_ENVELOPE_MAGIC_V3 { + let (has_occurrence, identity_version) = + if magic == RETAINED_INGRESS_ENVELOPE_MAGIC_V4 { + let has_occurrence = match cursor.read_u8()? { + 0 => false, + 1 => true, + _ => return Err(IngressEnvelopeDecodeError::NonCanonical), + }; + (has_occurrence, IngressIdentityVersion::SeparatedCausal) + } else { + ( + magic == RETAINED_INGRESS_ENVELOPE_MAGIC_V3, + IngressIdentityVersion::Original, + ) + }; + let occurrence = if has_occurrence { Some(ContinuumOccurrence::new( cursor.read_hash()?, cursor.read_hash()?, @@ -608,6 +735,7 @@ impl IngressEnvelope { intent_bytes, causal_parents, occurrence, + identity_version, ) } tag => return Err(IngressEnvelopeDecodeError::UnknownPayloadTag(tag)), @@ -700,6 +828,7 @@ impl IngressEnvelope { intent_bytes, &self.causal_parents, self.occurrence, + self.identity_version, ), } } @@ -809,24 +938,31 @@ impl<'a> RetainedIngressCursor<'a> { /// Computes the content address of a local intent. /// -/// Parentless intents preserve the original -/// `BLAKE3("ingress:" || kind_hash || bytes)` identity. Causal intents use a -/// separate versioned domain with explicit lengths so the same contract bytes -/// cited against different receipt evidence cannot collapse as duplicates. -/// A Continuum occurrence wraps that content identity: -/// `BLAKE3("echo.ingress.continuum-occurrence.v1\0" || occurrence_key || content_id)`, -/// so distinct occurrences of identical content are distinct ingress. The -/// occurrence domain differs from `"ingress:"` at its first byte: local ingress -/// hashes `"ingress:" || kind || bytes` with an arbitrary 32-byte kind, so any -/// domain beginning with `"ingress:"` could be reproduced by a local intent. +/// Every recipe's hash input begins with a domain no other recipe's input can +/// begin with, and every later field has a fixed length or a length prefix: +/// +/// - plain intent: `"ingress:" || kind || bytes` (unchanged since the first +/// release); +/// - causal intent, [`IngressIdentityVersion::SeparatedCausal`]: +/// `"echo.ingress.causal.v3\0" || kind || len || bytes || count || +/// (tag || receipt_ref)*`, with one-byte role tags and fixed-length refs; +/// - causal intent, [`IngressIdentityVersion::Original`] (retained requests +/// only): `"ingress:causal:v2\0" || ...`, which a plain intent can alias; +/// - Continuum occurrence: `"echo.ingress.continuum-occurrence.v1\0" || +/// occurrence_key || content_id`, where `content_id` is the request's own +/// id under the same identity version. +/// +/// The `echo.` domains differ from `"ingress:"` (and from the engine's +/// `"intent:"`) at the first byte, and from each other at byte 14. fn compute_ingress_id( kind: &IntentKind, bytes: &[u8], causal_parents: &[IngressCausalParent], occurrence: Option, + identity_version: IngressIdentityVersion, ) -> Hash { if let Some(occurrence) = occurrence { - let content_id = compute_ingress_id(kind, bytes, causal_parents, None); + let content_id = compute_ingress_id(kind, bytes, causal_parents, None, identity_version); let mut hasher = blake3::Hasher::new(); hasher.update(CONTINUUM_OCCURRENCE_INGRESS_DOMAIN); hasher.update(&occurrence.occurrence_key_digest); @@ -835,20 +971,41 @@ fn compute_ingress_id( } let mut hasher = blake3::Hasher::new(); if !causal_parents.is_empty() { - hasher.update(b"ingress:causal:v2\0"); - hasher.update(kind.as_hash()); - hasher.update(&(bytes.len() as u64).to_le_bytes()); - hasher.update(bytes); - hasher.update(&(causal_parents.len() as u64).to_le_bytes()); - for parent in causal_parents { - match parent { - IngressCausalParent::TickReceipt { receipt_ref } => { - hasher.update(b"tick-receipt\0"); + match identity_version { + IngressIdentityVersion::SeparatedCausal => { + hasher.update(SEPARATED_CAUSAL_INGRESS_DOMAIN); + hasher.update(kind.as_hash()); + hasher.update(&(bytes.len() as u64).to_le_bytes()); + hasher.update(bytes); + hasher.update(&(causal_parents.len() as u64).to_le_bytes()); + for parent in causal_parents { + let (tag, receipt_ref) = match parent { + IngressCausalParent::TickReceipt { receipt_ref } => (1u8, receipt_ref), + IngressCausalParent::ContractInverseTarget { receipt_ref } => { + (2u8, receipt_ref) + } + }; + hasher.update(&[tag]); hasher.update(&receipt_ref.to_canonical_bytes()); } - IngressCausalParent::ContractInverseTarget { receipt_ref } => { - hasher.update(b"contract-inverse-target\0"); - hasher.update(&receipt_ref.to_canonical_bytes()); + } + IngressIdentityVersion::Original => { + hasher.update(b"ingress:causal:v2\0"); + hasher.update(kind.as_hash()); + hasher.update(&(bytes.len() as u64).to_le_bytes()); + hasher.update(bytes); + hasher.update(&(causal_parents.len() as u64).to_le_bytes()); + for parent in causal_parents { + match parent { + IngressCausalParent::TickReceipt { receipt_ref } => { + hasher.update(b"tick-receipt\0"); + hasher.update(&receipt_ref.to_canonical_bytes()); + } + IngressCausalParent::ContractInverseTarget { receipt_ref } => { + hasher.update(b"contract-inverse-target\0"); + hasher.update(&receipt_ref.to_canonical_bytes()); + } + } } } } diff --git a/crates/warp-core/src/lib.rs b/crates/warp-core/src/lib.rs index 38cfeee4..a6c8288b 100644 --- a/crates/warp-core/src/lib.rs +++ b/crates/warp-core/src/lib.rs @@ -555,8 +555,8 @@ pub use head::{ /// [`WorldlineRuntime::ingest`] with these types. pub use head_inbox::{ make_intent_kind, ContinuumOccurrence, ContinuumOccurrenceError, HeadInbox, InboxAddress, - InboxPolicy, IngressCausalParent, IngressEnvelope, IngressEnvelopeDecodeError, IngressPayload, - IngressTarget, IntentKind, + InboxPolicy, IngressCausalParent, IngressEnvelope, IngressEnvelopeDecodeError, + IngressIdentityVersion, IngressPayload, IngressTarget, IntentKind, }; pub use worldline_registry::WorldlineRegistry; pub use worldline_state::{WorldlineFrontier, WorldlineState, WorldlineStateError}; diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index e8e248c7..d59ce69a 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -6909,7 +6909,7 @@ impl TrustedRuntimeApp<'_> { } if handle.duplicate && runtime_wal - .has_recorded_submission_acceptance(handle.submission_id, envelope.ingress_id()) + .has_recorded_submission_acceptance(handle.submission_id, handle.ingress_id) { self.host.track_pending_echo_operation_action_v1( handle.submission_id, @@ -6917,6 +6917,27 @@ impl TrustedRuntimeApp<'_> { ); return Ok(handle); } + // A resend that resolved to a submission accepted under the original + // identity recipe is recorded with that submission's retained + // envelope, so the WAL's identity matches the handle on recovery. + let envelope = if handle.ingress_id == envelope.ingress_id() { + envelope + } else { + match self + .host + .runtime + .witnessed_submission_envelope(&handle.submission_id) + .cloned() + { + Some(retained) => retained, + None => { + return Err(RuntimeError::WitnessedSubmissionEnvelopeUnavailable( + handle.submission_id, + ) + .into()) + } + } + }; if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle, sync) { if matches!(sync, CommitSync::Deferred) { // Earlier records of this batch are appended and not synced. diff --git a/crates/warp-core/tests/fixtures/legacy_causal_wal/README.md b/crates/warp-core/tests/fixtures/legacy_causal_wal/README.md new file mode 100644 index 00000000..51da1d55 --- /dev/null +++ b/crates/warp-core/tests/fixtures/legacy_causal_wal/README.md @@ -0,0 +1,43 @@ + + + +# Frozen legacy causal WAL + +A filesystem WAL written by Echo's own writer **before the separated causal +identity recipe existed** (#885). Its causal requests are retained as +`EINGR002` and `EINGR003` and identified by the original recipe, +`"ingress:causal:v2\0" || ...`. Readers must keep accepting these bytes and +their identities exactly as written, and a resend of either request in the +separated form must resolve to its original submission. + +**Do not regenerate these files with a newer writer.** They exist to prove that +a reader accepts what an earlier writer actually wrote; bytes produced by the +current encoder cannot show that. + +| File | Bytes | SHA-256 | +| --- | --- | --- | +| `segments/segment-00000000000000000001.ecwal` | 7594 | `95a762b44e674e7612ea74aff516034cdab192c8700c6c75d0a76b6386894457` | +| `writer-epochs.ecwal` | 231 | `bdef26980af969e059f52c3a212ade94cc26329083020b8bd1119313877efc61` | + +## Provenance + +Written on 2026-10-16 by `main` at `0512e9d2` (after #884, before #885), in the +guarded container, by a throwaway test added to +`crates/warp-core/tests/trusted_runtime_host_loop_tests.rs` and not committed: + +- the reference host (`runtime()`, `empty_engine()`) with a filesystem WAL and + the reference contract package; +- one WAL-acknowledged causal EINT intent, `causal_eint_envelope(lane, + [causal_receipt_ref(lane, "legacy-parent", 41)])`, staged with + `admission_ticket(61)` and decided by `run_until_idle(4)`: submission + `2348e63b4cd2b630e659d9bcac2e5456d56c35c7a5f04f717d07820a113888f7`, ingress + `811015cf74795d042fc637d41d506024f538206586707d533fede984a2dddcbf`; +- one WAL-acknowledged causal Continuum occurrence, `pack_intent_v1(MUTATION_OP_ID, + b"amount=9")` with the same parent and `ContinuumOccurrence::new([7; 32], + [0xa7; 32])`, left pending: submission + `8879b31036123e5b70962a068958f51e379750627e13abecf5b585d43fdd3f63`, ingress + `4af4cf66ff6f479386420b7d1df87e762ee5964323d402b3f378d007381acf8e`. + +The runtime lock file `writer-epoch.lock` is not retained; the WAL recreates it. + +Used by `causal_requests_accepted_before_the_separated_recipe_survive_restart_and_resend`. diff --git a/crates/warp-core/tests/fixtures/legacy_causal_wal/segments/segment-00000000000000000001.ecwal b/crates/warp-core/tests/fixtures/legacy_causal_wal/segments/segment-00000000000000000001.ecwal new file mode 100644 index 00000000..5570ef57 Binary files /dev/null and b/crates/warp-core/tests/fixtures/legacy_causal_wal/segments/segment-00000000000000000001.ecwal differ diff --git a/crates/warp-core/tests/fixtures/legacy_causal_wal/writer-epochs.ecwal b/crates/warp-core/tests/fixtures/legacy_causal_wal/writer-epochs.ecwal new file mode 100644 index 00000000..29120708 Binary files /dev/null and b/crates/warp-core/tests/fixtures/legacy_causal_wal/writer-epochs.ecwal differ diff --git a/crates/warp-core/tests/ingress_retention_codec_tests.rs b/crates/warp-core/tests/ingress_retention_codec_tests.rs index f0b7503c..bb385d51 100644 --- a/crates/warp-core/tests/ingress_retention_codec_tests.rs +++ b/crates/warp-core/tests/ingress_retention_codec_tests.rs @@ -6,8 +6,8 @@ use warp_core::{ make_head_id, make_intent_kind, CausalTickReceiptRef, ContinuumOccurrence, ContinuumOccurrenceError, GlobalTick, InboxAddress, IngressCausalParent, IngressEnvelope, - IngressEnvelopeDecodeError, IngressTarget, WorldlineId, WorldlineTick, WriterHeadKey, - CAUSAL_TICK_RECEIPT_REF_LEN, + IngressEnvelopeDecodeError, IngressIdentityVersion, IngressTarget, IntentKind, WorldlineId, + WorldlineTick, WriterHeadKey, CAUSAL_TICK_RECEIPT_REF_LEN, }; const PARENT_COUNT_OFFSET: usize = 8 + 1 + 32; @@ -359,3 +359,129 @@ fn a_local_intent_cannot_take_an_occurrence_ingress_id() { ); assert_ne!(crafted.ingress_id(), occurred.ingress_id()); } + +/// The original-recipe form of a separated causal envelope, as a request +/// retained before #885 decodes: `EINGR004` is the `EINGR002` layout plus a +/// trailing occurrence flag, so dropping the flag and relabelling gives the +/// bytes the earlier writer produced. +fn original_form(separated: &IngressEnvelope) -> IngressEnvelope { + let mut bytes = separated.to_retained_bytes(); + assert_eq!(&bytes[..8], b"EINGR004"); + assert_eq!(bytes.pop(), Some(0), "no occurrence"); + bytes[..8].copy_from_slice(b"EINGR002"); + IngressEnvelope::from_retained_bytes(&bytes).expect("an EINGR002 causal request decodes") +} + +/// #885 alias demonstration. Under the original recipe a plain intent +/// reproduces a causal intent's hash input byte for byte: the plain domain +/// `"ingress:"` prefixes the causal domain `"ingress:causal:v2\0"`, and the +/// plain kind is any 32 bytes. The construction is direct; it costs one +/// envelope, no search. The separated recipe has no such alias. +#[test] +fn a_plain_intent_aliases_the_original_causal_recipe_but_not_the_separated_one() { + let worldline_id = WorldlineId::from_bytes(digest("worldline")); + let parent = receipt_ref(worldline_id, "alias-parent"); + let kind = make_intent_kind("fixture.intent/causal"); + let payload = b"causal payload".to_vec(); + let separated = IngressEnvelope::local_intent_with_causal_parents( + IngressTarget::DefaultWriter { worldline_id }, + kind, + payload.clone(), + vec![IngressCausalParent::TickReceipt { + receipt_ref: parent, + }], + ); + assert_eq!( + separated.identity_version(), + IngressIdentityVersion::SeparatedCausal + ); + let original = original_form(&separated); + assert_eq!( + original.identity_version(), + IngressIdentityVersion::Original + ); + assert_ne!(original.ingress_id(), separated.ingress_id()); + + // "ingress:" || alias_kind || alias_bytes == + // "ingress:causal:v2\0" || kind || len || bytes || count || "tick-receipt\0" || ref + let mut alias_kind = [0u8; 32]; + alias_kind[..10].copy_from_slice(b"causal:v2\0"); + alias_kind[10..].copy_from_slice(&kind.as_hash()[..22]); + let mut alias_bytes = kind.as_hash()[22..].to_vec(); + alias_bytes.extend_from_slice(&(payload.len() as u64).to_le_bytes()); + alias_bytes.extend_from_slice(&payload); + alias_bytes.extend_from_slice(&1u64.to_le_bytes()); + alias_bytes.extend_from_slice(b"tick-receipt\0"); + alias_bytes.extend_from_slice(&parent.to_canonical_bytes()); + let alias = IngressEnvelope::local_intent( + IngressTarget::DefaultWriter { worldline_id }, + IntentKind::from_hash(alias_kind), + alias_bytes, + ); + assert_eq!( + alias.ingress_id(), + original.ingress_id(), + "the original recipe is aliased" + ); + assert_ne!( + alias.ingress_id(), + separated.ingress_id(), + "the separated recipe is not" + ); +} + +#[test] +fn retention_formats_keep_their_identity_versions() { + let worldline_id = WorldlineId::from_bytes(digest("worldline")); + let parents = vec![IngressCausalParent::TickReceipt { + receipt_ref: receipt_ref(worldline_id, "retained-parent"), + }]; + let target = IngressTarget::DefaultWriter { worldline_id }; + let kind = make_intent_kind("fixture.intent/causal"); + + // Plain intents keep EINGR002 and the original recipe. + let plain = IngressEnvelope::local_intent(target.clone(), kind, b"plain".to_vec()); + assert_eq!(plain.identity_version(), IngressIdentityVersion::Original); + assert_eq!(&plain.to_retained_bytes()[..8], b"EINGR002"); + + // New causal requests are EINGR004 and round trip, with and without an + // occurrence. + let causal = IngressEnvelope::local_intent_with_causal_parents( + target.clone(), + kind, + b"c".to_vec(), + parents.clone(), + ); + let occurred = IngressEnvelope::continuum_occurred_intent( + target, + kind, + b"c".to_vec(), + parents, + continuum_occurrence("causal"), + ); + for envelope in [causal.clone(), occurred] { + let retained = envelope.to_retained_bytes(); + assert_eq!(&retained[..8], b"EINGR004"); + round_trip(envelope); + } + + // The original form keeps its legacy bytes and identity exactly. + let original = original_form(&causal); + assert_eq!(&original.to_retained_bytes()[..8], b"EINGR002"); + round_trip(original); + + // An EINGR004 record must carry parents and a valid occurrence flag. + let mut bad_flag = causal.to_retained_bytes(); + *bad_flag.last_mut().expect("flag byte") = 2; + assert_eq!( + IngressEnvelope::from_retained_bytes(&bad_flag), + Err(IngressEnvelopeDecodeError::NonCanonical) + ); + let mut parentless = plain.to_retained_bytes(); + parentless[..8].copy_from_slice(b"EINGR004"); + parentless.push(0); + assert_eq!( + IngressEnvelope::from_retained_bytes(&parentless), + Err(IngressEnvelopeDecodeError::NonCanonical) + ); +} diff --git a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs index c0fc6a69..208a407c 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -4211,3 +4211,216 @@ fn a_strand_inherits_its_fork_basis_occurrences_across_restart() { drop(restored); fs::remove_dir_all(&root).expect("WAL fixture cleanup succeeds"); } + +const LEGACY_CAUSAL_SEGMENT: &[u8] = + include_bytes!("fixtures/legacy_causal_wal/segments/segment-00000000000000000001.ecwal"); +const LEGACY_CAUSAL_EPOCHS: &[u8] = + include_bytes!("fixtures/legacy_causal_wal/writer-epochs.ecwal"); +/// Decided causal submission in the frozen WAL, and its original ingress id. +const LEGACY_DECIDED: &str = "2348e63b4cd2b630e659d9bcac2e5456d56c35c7a5f04f717d07820a113888f7"; +const LEGACY_DECIDED_INGRESS: &str = + "811015cf74795d042fc637d41d506024f538206586707d533fede984a2dddcbf"; +/// Pending causal Continuum occurrence in the frozen WAL. +const LEGACY_PENDING: &str = "8879b31036123e5b70962a068958f51e379750627e13abecf5b585d43fdd3f63"; + +fn hex_hash(value: &str) -> Hash { + hex::decode(value) + .expect("hex") + .try_into() + .expect("32 bytes") +} + +/// A fresh copy of the frozen legacy causal WAL, checked against its digests. +fn legacy_causal_wal() -> PathBuf { + use sha2::{Digest, Sha256}; + assert_eq!( + hex::encode(Sha256::digest(LEGACY_CAUSAL_SEGMENT)), + "95a762b44e674e7612ea74aff516034cdab192c8700c6c75d0a76b6386894457" + ); + assert_eq!( + hex::encode(Sha256::digest(LEGACY_CAUSAL_EPOCHS)), + "bdef26980af969e059f52c3a212ade94cc26329083020b8bd1119313877efc61" + ); + let root = temp_runtime_wal_dir("legacy-causal"); + fs::create_dir_all(root.join("segments")).expect("segments directory"); + fs::write( + root.join("segments") + .join("segment-00000000000000000001.ecwal"), + LEGACY_CAUSAL_SEGMENT, + ) + .expect("frozen segment copies"); + fs::write(root.join("writer-epochs.ecwal"), LEGACY_CAUSAL_EPOCHS) + .expect("frozen epoch ledger copies"); + root +} + +/// Issue #885: restart, then resend. Requests a writer accepted before the +/// separated causal recipe resolve to their original submissions when resent +/// in the new form: no acceptance is written, the decided one does not run +/// again, and the pending one is staged from its retained envelope and is +/// decided once. A request with a changed parent is a new act. +#[test] +fn causal_requests_accepted_before_the_separated_recipe_survive_restart_and_resend() { + let root = legacy_causal_wal(); + let (rt, lane) = runtime(); + let mut host = TrustedRuntimeHost::new(rt, empty_engine()).expect("host"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&root)) + .expect("the frozen WAL recovers"); + host.register_contract_package(package()).expect("package"); + let decided = hex_hash(LEGACY_DECIDED); + let pending = hex_hash(LEGACY_PENDING); + assert_eq!( + host.runtime() + .witnessed_submission_envelope(&decided) + .expect("retained") + .identity_version(), + warp_core::IngressIdentityVersion::Original + ); + let acceptances = host + .runtime_wal() + .expect("wal") + .submission_acceptance_count(); + let ticks = host.provenance().len(lane).expect("history"); + + let parent = causal_receipt_ref(lane, "legacy-parent", 41); + let resend = host + .app() + .submit_intent_with_runtime_wal_ack(causal_eint_envelope(lane, vec![parent])) + .expect("a resend of the decided request"); + assert!(resend.duplicate); + assert_eq!(resend.submission_id, decided); + assert_eq!(resend.ingress_id, hex_hash(LEGACY_DECIDED_INGRESS)); + let pending_resend = host + .app() + .submit_intent_with_runtime_wal_ack(IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id: lane }, + make_intent_kind("echo.intent/eint-v1"), + echo_wasm_abi::pack_intent_v1(MUTATION_OP_ID, b"amount=9").expect("EINT"), + vec![IngressCausalParent::TickReceipt { + receipt_ref: parent, + }], + ContinuumOccurrence::new([7; 32], [0xa7; 32]), + )) + .expect("a resend of the pending occurrence"); + assert!(pending_resend.duplicate); + assert_eq!(pending_resend.submission_id, pending); + assert_eq!( + host.runtime_wal() + .expect("wal") + .submission_acceptance_count(), + acceptances, + "resends write no acceptance" + ); + host.run_until_idle(4).expect("idle"); + assert_eq!( + host.provenance().len(lane).expect("history"), + ticks, + "the decided request does not run again" + ); + + host.stage_installed_contract_submission(pending, &admission_ticket(62)) + .expect("the pending request stages from its retained envelope"); + host.run_until_idle(4).expect("commit"); + assert_eq!(host.provenance().len(lane).expect("history"), ticks + 1); + // Decided once: the reference rule does not match `amount=9`. + let outcome = host.app().observe_intent_outcome(&pending); + assert!( + matches!(outcome, IntentOutcome::NoMatchingRule { submission_id, .. } if submission_id == pending), + "{outcome:?}" + ); + host.run_until_idle(4).expect("idle"); + assert_eq!(host.provenance().len(lane).expect("history"), ticks + 1); + + let changed = host + .app() + .submit_intent_with_runtime_wal_ack(causal_eint_envelope( + lane, + vec![causal_receipt_ref(lane, "another-parent", 42)], + )) + .expect("a changed parent is a new request"); + assert!(!changed.duplicate); + assert_eq!( + host.runtime_wal() + .expect("wal") + .submission_acceptance_count(), + acceptances + 1 + ); + drop(host); + fs::remove_dir_all(&root).expect("WAL fixture cleanup succeeds"); +} + +/// PR #900 review: a runtime may hold accepted legacy causal material that a +/// newly enabled WAL does not. A resend in the separated form resolves to that +/// submission, and the WAL must record the retained (original-form) envelope +/// under it, so the log still recovers. +#[test] +fn a_legacy_resolved_resend_records_the_retained_envelope() { + let legacy_root = legacy_causal_wal(); + let (rt, lane) = runtime(); + let mut recovered = TrustedRuntimeHost::new(rt, empty_engine()).expect("host"); + recovered + .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&legacy_root)) + .expect("the frozen WAL recovers"); + let pending = hex_hash(LEGACY_PENDING); + let snapshot = recovered + .runtime() + .witnessed_submission_persistence_snapshot() + .expect("snapshot"); + let pending_only = warp_core::WitnessedSubmissionPersistenceSnapshot::new( + snapshot + .records() + .iter() + .filter(|record| record.submission.submission_id == pending) + .cloned() + .collect(), + ); + drop(recovered); + fs::remove_dir_all(&legacy_root).expect("fixture cleanup"); + + // A host whose WAL has never seen the accepted legacy request, given that + // request through host-level reconfiguration of its runtime. (Enabling a + // WAL on a runtime that already holds it is refused.) + let fresh_root = temp_runtime_wal_dir("legacy-resolved-resend"); + let (rt, _) = runtime(); + let mut host = TrustedRuntimeHost::new(rt, empty_engine()).expect("host"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&fresh_root)) + .expect("a fresh WAL"); + let mut parts = host.into_parts(); + parts + .runtime_mut() + .restore_witnessed_submission_persistence(pending_only) + .expect("legacy material restores"); + let mut host = TrustedRuntimeHost::from_parts(parts); + let parent = causal_receipt_ref(lane, "legacy-parent", 41); + let resend = host + .app() + .submit_intent_with_runtime_wal_ack(IngressEnvelope::continuum_occurred_intent( + IngressTarget::DefaultWriter { worldline_id: lane }, + make_intent_kind("echo.intent/eint-v1"), + echo_wasm_abi::pack_intent_v1(MUTATION_OP_ID, b"amount=9").expect("EINT"), + vec![IngressCausalParent::TickReceipt { + receipt_ref: parent, + }], + ContinuumOccurrence::new([7; 32], [0xa7; 32]), + )) + .expect("the resend resolves to the legacy submission"); + assert!(resend.duplicate); + assert_eq!(resend.submission_id, pending); + drop(host); + + let (rt, _) = runtime(); + let mut restarted = TrustedRuntimeHost::new(rt, empty_engine()).expect("host"); + restarted + .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&fresh_root)) + .expect("the WAL recorded the retained envelope and recovers"); + assert_eq!( + restarted + .runtime() + .witnessed_submission_envelope(&pending) + .expect("retained") + .identity_version(), + warp_core::IngressIdentityVersion::Original + ); + drop(restarted); + fs::remove_dir_all(&fresh_root).expect("WAL cleanup"); +} diff --git a/docs/spec/canonical-inbox-sequencing.md b/docs/spec/canonical-inbox-sequencing.md index b54dd6d0..e02654ac 100644 --- a/docs/spec/canonical-inbox-sequencing.md +++ b/docs/spec/canonical-inbox-sequencing.md @@ -28,6 +28,8 @@ Two intents are the same when their canonical bytes are byte-identical: `intent_ Refined by Decision 6 for intents admitted from Continuum: a declared Continuum occurrence is not transport metadata, and it is part of ingress identity. Local intents keep this rule unchanged. +Refined by Decision 7 for causal intents: the ingress id of an intent that cites causal parents uses a versioned recipe that a plain intent cannot alias. + ## Decision 2: Ingress is idempotent per resolved writer head The pending inbox is keyed by `ingress_id` for the resolved writer head. Re-ingesting an already pending or already committed intent must not create a second live entry. @@ -142,3 +144,56 @@ however the parent advances, and changes no retained receipt on either lane. Rec branches is a separate merge question. Decision record: [#880](https://github.com/flyingrobots/echo/issues/880). + +## Decision 7: Identity recipes are versioned and unambiguous + +The original recipes hash a plain intent as `"ingress:" || kind || bytes` and a +causal intent as `"ingress:causal:v2\0" || kind || len || bytes || count || +parents`. The causal domain begins with the plain one and a kind is any 32 +bytes, so a plain intent can reproduce a causal intent's hash input byte for +byte. That is an **encoding alias**, not a hash collision: the construction is +direct and costs nothing (`a_plain_intent_aliases_the_original_causal_recipe_but_not_the_separated_one`). +The planted plain intent then takes the causal request's ingress and submission +IDs, and the genuine request is treated as its duplicate. + +Every envelope now records its `IngressIdentityVersion`: + +| Recipe | Version | Domain | Retained as | +| --- | --- | --- | --- | +| Plain intent | `Original` (unchanged) | `ingress:` | `EINGR002` (`EINGR003` with an occurrence) | +| Causal intent, accepted before #885 | `Original` | `ingress:causal:v2\0` | `EINGR002` / `EINGR003` | +| Causal intent, new | `SeparatedCausal` | `echo.ingress.causal.v3\0` | `EINGR004` | +| Continuum occurrence | as its content | `echo.ingress.continuum-occurrence.v1\0` | per content | + +`EINGR001` to `EINGR003` are retention formats whose identity interpretation is +frozen: they decode to `Original`, including ordinary plain intents, and keep +their bytes and IDs exactly. `EINGR004` is the `EINGR002` layout followed by an +occurrence flag and, when set, the two occurrence digests. Every recipe's input +begins with a domain no other input can begin with, and every later field has a +fixed length or a length prefix; the separated recipe tags parent roles with one +byte over fixed-length receipt references. + +Every newly constructed causal intent uses the separated recipe. A causal +request in the original recipe can only name an existing submission; admitting +it as new fails with `RuntimeError::OriginalCausalIdentityNotAdmissible`. +Historical decoding and new admission policy are separate: retained records are +never rejected because a new admission rule would refuse them. + +**Transition rule.** A resend of a causal request accepted under the original +recipe arrives in the separated form, with a different ingress id. Exact legacy +matches take precedence over creating a separated submission, so a resend never +becomes a second act: + +| Found for the request's original-recipe id on its head | Echo's answer | +| --- | --- | +| A retained submission whose envelope states the same kind, bytes, causal parents, occurrence (both digests) and resolved head | That submission (`Duplicate`, with its original ingress id); staging uses its retained envelope | +| A retained submission whose envelope states something else (an alias) | Not a retry and no reservation: ordinary separated admission | +| A submission without retained envelope material, or committed ingress without a submission | `RuntimeError::OriginalIngressIdentityUnverifiable` | +| Both a legacy and a separated submission of the same request | `RuntimeError::IngressIdentityAmbiguous` | + +The same comparison applies to occurrence retries and to occurrences a strand +inherits. The recipe, the retention layout and this lookup are Echo's own; any +identity version or outcome exposed through Continuum must first be authored in +Continuum's contracts. + +Decision record: [#885](https://github.com/flyingrobots/echo/issues/885).