Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,15 @@

### Added

- 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`)
naming the original submission, its head, the fork basis and its receipt
when ticketed; nothing is admitted and no acceptance is written.
`IntentSubmissionHandle` reports it as a duplicate with `inherited_through`
set. Reusing an inherited occurrence key for other material is an
occurrence conflict. Occurrences undecided at the fork basis, or decided
after it, are not inherited. Nested forks and restart preserve this.
- Restoring committed history without the witnessed submissions it accounts
for no longer reopens admission (#887). `restore_causal_runtime_history`
keeps the restored worldline readable, but `submit_intent`, `ingest` and
Expand Down
864 changes: 845 additions & 19 deletions crates/warp-core/src/coordinator.rs

Large diffs are not rendered by default.

16 changes: 8 additions & 8 deletions crates/warp-core/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -534,14 +534,14 @@ pub use worldline::{
/// Prefer this coordinator/runtime API for new stepping and routing code.
pub use coordinator::{
ForkStrandReceipt, ForkStrandRequest, IngressDisposition, IngressSubmissionGeneration,
IntentOutcome, IntentOutcomeDecision, IntentOutcomeObservation, IntentOutcomeReceipt,
IntentSubmissionDisposition, IntentSubmissionHandle, IntentSubmissionRecord,
ReceiptCorrelationPersistenceRecord, ReceiptCorrelationRecord, RuntimeError,
SchedulerCoordinator, SchedulerFaultGeneration, SchedulerFaultId, SchedulerFaultRecord,
SchedulerFaultRecoveryAuthority, SchedulerFaultScope, SchedulerFaultStatus, SchedulerRunId,
StepRecord, TicketedRuntimeIngressAuthority, TicketedRuntimeIngressDisposition,
TicketedRuntimeIngressRecord, WitnessedSubmissionPersistenceRecord,
WitnessedSubmissionPersistenceSnapshot, WorldlineRuntime,
InheritedSubmission, InheritedThrough, IntentOutcome, IntentOutcomeDecision,
IntentOutcomeObservation, IntentOutcomeReceipt, IntentSubmissionDisposition,
IntentSubmissionHandle, IntentSubmissionRecord, ReceiptCorrelationPersistenceRecord,
ReceiptCorrelationRecord, RuntimeError, SchedulerCoordinator, SchedulerFaultGeneration,
SchedulerFaultId, SchedulerFaultRecord, SchedulerFaultRecoveryAuthority, SchedulerFaultScope,
SchedulerFaultStatus, SchedulerRunId, StepRecord, TicketedRuntimeIngressAuthority,
TicketedRuntimeIngressDisposition, TicketedRuntimeIngressRecord,
WitnessedSubmissionPersistenceRecord, WitnessedSubmissionPersistenceSnapshot, WorldlineRuntime,
};
/// Writer-head registry and routing primitives used by the runtime-owned ingress path.
pub use head::{
Expand Down
16 changes: 16 additions & 0 deletions crates/warp-core/src/strand.rs
Original file line number Diff line number Diff line change
Expand Up @@ -658,6 +658,15 @@ pub enum StrandError {
#[error("strand already exists: {0:?}")]
AlreadyExists(StrandId),

/// The child worldline already belongs to another strand.
#[error("worldline {child:?} is already the child of strand {existing:?}")]
ChildWorldlineAlreadyRelated {
/// Child worldline named by the rejected strand.
child: WorldlineId,
/// Strand the worldline already belongs to.
existing: StrandId,
},

/// The strand was not found in the registry.
#[error("strand not found: {0:?}")]
NotFound(StrandId),
Expand Down Expand Up @@ -838,6 +847,13 @@ impl StrandRegistry {
if self.strands.contains_key(&strand.strand_id) {
return Err(StrandError::AlreadyExists(strand.strand_id));
}
// A child worldline has exactly one causal past: one strand relation.
if let Some(existing) = self.find_by_child_worldline(&strand.child_worldline_id) {
return Err(StrandError::ChildWorldlineAlreadyRelated {
child: strand.child_worldline_id,
existing: existing.strand_id,
});
}
validate_local_strand_invariants::<DynamicPosture>(
strand.strand_id,
&strand.fork_basis_ref,
Expand Down
9 changes: 8 additions & 1 deletion crates/warp-core/src/trusted_runtime_host.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6755,7 +6755,7 @@ impl TrustedRuntimeApp<'_> {
return Err(RuntimeError::EchoOperationActionRequiresRuntimeWalAck);
}
let handle = self.host.runtime.submit_app_intent(envelope)?;
if self.host.runtime_wal.is_none() {
if self.host.runtime_wal.is_none() && handle.inherited_through.is_none() {
self.host.track_pending_echo_operation_action_v1(
handle.submission_id,
is_echo_operation_action,
Expand Down Expand Up @@ -6902,6 +6902,11 @@ impl TrustedRuntimeApp<'_> {
.rollback_intent_submission(submission_rollback, &handle);
return Err(TrustedRuntimeHostError::RuntimeWalUnavailable);
};
// An occurrence already in inherited history admits nothing here, so
// there is no acceptance to record.
if handle.inherited_through.is_some() {
return Ok(handle);
}
if handle.duplicate
&& runtime_wal
.has_recorded_submission_acceptance(handle.submission_id, envelope.ingress_id())
Expand Down Expand Up @@ -8270,6 +8275,7 @@ mod tests {
submission_id: [12; 32],
submission_generation: IngressSubmissionGeneration::from_raw(1),
duplicate: false,
inherited_through: None,
};
let record = SubmissionAcceptanceRecord {
submission_id: handle.submission_id,
Expand Down Expand Up @@ -8383,6 +8389,7 @@ mod tests {
submission_id: [21; 32],
submission_generation: IngressSubmissionGeneration::from_raw(1),
duplicate: false,
inherited_through: None,
};
let acceptance = SubmissionAcceptanceRecord {
submission_id: handle.submission_id,
Expand Down
21 changes: 14 additions & 7 deletions crates/warp-core/tests/inbox.rs
Original file line number Diff line number Diff line change
Expand Up @@ -249,7 +249,8 @@ fn ticketed_invocation_ingests_runtime_envelope_without_ticking() {
);
let accepted = match runtime.submit_intent(envelope.clone()).unwrap() {
IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id),
IntentSubmissionDisposition::Duplicate { .. } => None,
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None,
};
assert!(accepted.is_some(), "first submission should be accepted");
let Some(submission) = accepted else {
Expand Down Expand Up @@ -327,7 +328,8 @@ fn ticketed_ingress_preserves_submission_and_ticket_identity() {
);
let accepted = match runtime.submit_intent(envelope.clone()).unwrap() {
IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id),
IntentSubmissionDisposition::Duplicate { .. } => None,
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None,
};
assert!(accepted.is_some(), "first submission should be accepted");
let Some(submission) = accepted else {
Expand Down Expand Up @@ -395,7 +397,8 @@ fn receipt_correlation_does_not_exist_before_scheduler_tick() {
);
let accepted = match runtime.submit_intent(envelope.clone()).unwrap() {
IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id),
IntentSubmissionDisposition::Duplicate { .. } => None,
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None,
};
assert!(accepted.is_some(), "first submission should be accepted");
let Some(submission) = accepted else {
Expand Down Expand Up @@ -463,7 +466,8 @@ fn witnessed_submission_outcome_observation_is_pending_without_tick() {
submission_generation,
..
} => Some((submission_id, submission_generation)),
IntentSubmissionDisposition::Duplicate { .. } => None,
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None,
};
assert!(accepted.is_some(), "first submission should be accepted");
let Some((submission, generation)) = accepted else {
Expand Down Expand Up @@ -497,7 +501,8 @@ fn ticketed_ingress_correlates_tick_receipt_after_scheduler_owned_tick() {
);
let accepted = match runtime.submit_intent(envelope.clone()).unwrap() {
IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id),
IntentSubmissionDisposition::Duplicate { .. } => None,
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None,
};
assert!(accepted.is_some(), "first submission should be accepted");
let Some(submission) = accepted else {
Expand Down Expand Up @@ -579,7 +584,8 @@ fn ticketed_submission_outcome_observation_is_decided_after_scheduler_tick() {
);
let accepted = match runtime.submit_intent(envelope.clone()).unwrap() {
IntentSubmissionDisposition::Accepted { submission_id, .. } => Some(submission_id),
IntentSubmissionDisposition::Duplicate { .. } => None,
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => None,
};
assert!(accepted.is_some(), "first submission should be accepted");
let Some(submission) = accepted else {
Expand Down Expand Up @@ -659,7 +665,8 @@ fn ticketed_ingress_rejects_legacy_pending_duplicate_without_correlation() {

let accepted = match runtime.ingest(envelope.clone()).unwrap() {
IngressDisposition::Accepted { submission_id, .. } => Some(submission_id),
IngressDisposition::Duplicate { .. } => None,
IngressDisposition::Duplicate { .. }
| IngressDisposition::AlreadyPresentInInheritedHistory { .. } => None,
};
assert!(
accepted.is_some(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -480,7 +480,8 @@ fn installed_contract_mutation_dispatches_only_through_ticketed_scheduler_tick()
.expect("submission should be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("first submission should not be duplicate")
}
};
Expand Down Expand Up @@ -570,7 +571,8 @@ fn unsupported_installed_contract_mutation_cannot_enter_ticketed_runtime_ingress
.expect("unsupported submission should still be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("first submission should not be duplicate")
}
};
Expand Down Expand Up @@ -620,7 +622,8 @@ fn relabeled_eint_cannot_enter_installed_contract_runtime_ingress() {
.expect("relabeled submission is witnessed before contract admission")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("first relabeled submission should not be duplicate")
}
};
Expand Down Expand Up @@ -663,7 +666,8 @@ fn footprint_conflict_is_final_without_hidden_retry() {
.expect("first submission should be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("first submission must not be duplicate")
}
};
Expand All @@ -672,7 +676,8 @@ fn footprint_conflict_is_final_without_hidden_retry() {
.expect("second submission should be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("second submission must not be duplicate")
}
};
Expand Down Expand Up @@ -864,7 +869,8 @@ fn witnessed_submission_replay_restores_pending_history_without_runtime_ingress(
.expect("first submission should be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("first submission must not be duplicate")
}
};
Expand All @@ -873,7 +879,8 @@ fn witnessed_submission_replay_restores_pending_history_without_runtime_ingress(
.expect("second submission should be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("second submission must not be duplicate")
}
};
Expand Down Expand Up @@ -964,7 +971,8 @@ fn external_contract_fixture_proves_mutation_query_retention_and_replay() {
.expect("external fixture submission should be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("first external fixture submission must not be duplicate")
}
};
Expand Down Expand Up @@ -1156,7 +1164,8 @@ fn installed_contract_pipeline_replays_to_same_receipt_and_outcome() {
.expect("submission should be witnessed")
{
IntentSubmissionDisposition::Accepted { submission_id, .. } => submission_id,
IntentSubmissionDisposition::Duplicate { .. } => {
IntentSubmissionDisposition::Duplicate { .. }
| IntentSubmissionDisposition::AlreadyPresentInInheritedHistory { .. } => {
panic!("first submission must not be duplicate")
}
};
Expand Down
94 changes: 94 additions & 0 deletions crates/warp-core/tests/trusted_runtime_host_loop_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4117,3 +4117,97 @@ fn continuum_occurrence_identity_survives_host_restart_and_redelivery() {
drop(restarted);
fs::remove_dir_all(&wal_root).expect("WAL fixture cleanup succeeds");
}

/// Issue #886: after restart, a strand still inherits the occurrences of its
/// fork basis. Redelivering one to the child names the original submission and
/// writes no acceptance; reusing its key for other material is a conflict.
#[test]
fn a_strand_inherits_its_fork_basis_occurrences_across_restart() {
let root = temp_runtime_wal_dir("strand-occurrence-inheritance");
let (rt, lane) = runtime();
let head = *rt.heads().iter().next().expect("head").0;
let mut host = TrustedRuntimeHost::new(rt, empty_engine()).expect("host");
host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&root))
.expect("wal");
host.register_contract_package(package()).expect("package");
let applied = [0xa1; 32];
let original = host
.app()
.submit_intent_with_runtime_wal_ack(occurred_eint_envelope(lane, 1, applied, MUTATION_VARS))
.expect("submit");
host.stage_installed_contract_submission(original.submission_id, &admission_ticket(31))
.expect("stage");
host.run_until_idle(4).expect("commit");
let child = host
.fork_local_operation_strand_v1(head, "occurrence-child")
.expect("native fork");
drop(host);

let (rt, _) = runtime();
let mut restored = TrustedRuntimeHost::new(rt, empty_engine()).expect("host");
restored
.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&root))
.expect("recovery");
restored
.register_contract_package(package())
.expect("package");
let to_child = |key: u8, vars: &[u8]| {
IngressEnvelope::continuum_occurred_intent(
IngressTarget::ExactHead { key: child },
make_intent_kind("echo.intent/eint-v1"),
echo_wasm_abi::pack_intent_v1(MUTATION_OP_ID, vars).expect("EINT should pack"),
Vec::new(),
ContinuumOccurrence::new([key; 32], applied),
)
};
let acceptances = restored
.runtime_wal()
.expect("wal")
.submission_acceptance_count();
let witnessed = restored.runtime().witnessed_submission_count();
let inherited = restored
.app()
.submit_intent_with_runtime_wal_ack(to_child(1, MUTATION_VARS))
.expect("an inherited occurrence is an idempotency result");
assert!(inherited.duplicate);
assert_eq!(inherited.submission_id, original.submission_id);
assert_eq!(inherited.head_key, original.head_key);
let fork = restored
.runtime()
.strands()
.find_by_child_worldline(&child.worldline_id)
.expect("retained strand")
.fork_basis_ref();
assert_eq!(inherited.inherited_through, Some(fork.into()));
assert_eq!(
restored
.runtime_wal()
.expect("wal")
.submission_acceptance_count(),
acceptances,
"an inherited occurrence writes no acceptance"
);
assert_eq!(restored.runtime().witnessed_submission_count(), witnessed);

let conflict = restored
.app()
.submit_intent_with_runtime_wal_ack(to_child(1, b"amount=8"))
.expect_err("reusing an inherited occurrence key for other content");
assert!(matches!(
conflict,
TrustedRuntimeHostError::Runtime(error) if matches!(
*error,
RuntimeError::ContinuumOccurrenceConflict { existing_submission_id, .. }
if existing_submission_id == original.submission_id
)
));

let fresh = restored
.app()
.submit_intent_with_runtime_wal_ack(to_child(2, MUTATION_VARS))
.expect("an occurrence absent from the inherited past is admitted");
assert!(!fresh.duplicate && fresh.inherited_through.is_none());
assert_eq!(fresh.head_key, child);
drop(restored);
fs::remove_dir_all(&root).expect("WAL fixture cleanup succeeds");
}
4 changes: 4 additions & 0 deletions crates/warp-wasm/src/warp_kernel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1014,6 +1014,10 @@ impl KernelPort for WarpKernel {
Some(submission_id.to_vec()),
Some(submission_generation.as_u64()),
),
IngressDisposition::AlreadyPresentInInheritedHistory { inherited, .. } => (
Some(inherited.submission_id.to_vec()),
Some(inherited.submission_generation.as_u64()),
),
};
self.refresh_scheduler_status();
Ok(DispatchResponse {
Expand Down
Loading
Loading