Fault Tolerance

fn read_record(file: &mut File, head: AnchorHead) -> Result<CommitRecord, RecoveryError> {    let end = recovery_bounds::checked_end(head.offset, head.length)        .ok_or(RecoveryError::InvalidImage)?;    if head.length == 0 || end > file.metadata().map_err(|_| RecoveryError::Io)?.len() {        return Err(RecoveryError::InvalidImage);    }    let size = usize::try_from(head.length).map_err(|_| RecoveryError::InvalidImage)?;    let mut bytes = Vec::new();    bytes        .try_reserve_exact(size)        .map_err(|_| RecoveryError::InvalidImage)?;    bytes.resize(size, 0);    file.seek(SeekFrom::Start(head.offset))        .map_err(|_| RecoveryError::Io)?;    file.read_exact(&mut bytes).map_err(|_| RecoveryError::Io)?;    let digest: [u8; 32] = Sha256::digest(&bytes).into();    if digest != head.digest {        return Err(RecoveryError::InvalidImage);    }    let record: CommitRecord =        serde_json::from_slice(&bytes).map_err(|_| RecoveryError::InvalidImage)?;    if record.profile != head.profile        || record.sequence != head.sequence        || encode(&record)? != bytes    {        return Err(RecoveryError::InvalidImage);    }    match record.previous {        None if head.sequence == 0 => (),        Some(previous)            if recovery_bounds::valid_predecessor(                previous.profile == head.profile,                previous.sequence,                previous.offset,                previous.length,                head.sequence,                head.offset,            ) => {}        _ => return Err(RecoveryError::InvalidImage),    }    Ok(record)}

fn append_record(    file: &mut File,    record: &CommitRecord,    flush: impl FnOnce(&File) -> io::Result<()>,) -> Result<AnchorHead, RecoveryError> {    let bytes = encode(record)?;    let offset = file.seek(SeekFrom::End(0)).map_err(|_| RecoveryError::Io)?;    file.write_all(&bytes).map_err(|_| RecoveryError::Io)?;    flush(file).map_err(|_| RecoveryError::Io)?;    Ok(AnchorHead {        profile: record.profile,        sequence: record.sequence,        offset,        length: u64::try_from(bytes.len()).map_err(|_| RecoveryError::Io)?,        digest: Sha256::digest(&bytes).into(),    })}

fn confirm_authority_record(    file: &mut File,    anchor: &impl AuthenticatedHead,    expected: AnchorHead,    record: &CommitRecord,    checkpoint: CheckpointState,) -> Result<(), RecoveryError> {    let (head, floor) = match checkpoint {        CheckpointState::Direct => (anchor.current().map_err(|_| RecoveryError::Hold)?, None),        CheckpointState::Independent(observed) => {            let (head, current) = anchor.checkpoint_readback().map_err(|_| RecoveryError::Hold)?;            if current.writer_generation != observed.writer_generation                || current.writer_id != observed.writer_id                || current.protected_head != Some(expected)            {                return Err(RecoveryError::Hold);            }            (head, Some(current.head.ok_or(RecoveryError::Hold)?))        }    };    if head != Some(expected) {        return Err(RecoveryError::Hold);    }    let readback = exact_image(file, expected)?;    if readback != *record {        return Err(RecoveryError::InvalidImage);    }    if let Some(floor) = floor {        validate_lineage(file, expected, &readback, floor)?;        anchor.verify_independent_journal(file, expected).map_err(|_| RecoveryError::Hold)?;    }    Ok(())}

flush-before-cas-is-not-committed	prepare	deployment=7;profile=99;old_head=0;candidate=1;flushed=true;cas=false	head=0;new_reply_exposed=false	staged_commit_head_still_identifies_the_previous_image,new_reply_is_invisible_during_prepare_or_flushhead-confirms-the-exact-record	reopen	deployment=7;format=1;profile=99;old_head=0;candidate=1;flushed=true;cas=true	head=1;reopened=true	exact_new_head_reopens_the_records_only_product,independently_anchored_records_construct_a_fresh_endpointold-valid-record-replay	reopen	deployment=7;profile=99;trusted_head=1;supplied_sequence=0;old_record_valid=true	reopened=false;genesis_fallback=false	replaying_old_valid_bytes_cannot_replace_the_latest_image,an_old_valid_record_at_the_current_address_is_rejectedcurrent-sequence-altered-body	reopen	deployment=7;profile=99;trusted_head=1;supplied_sequence=1;body_changed=true	reopened=false	a_changed_body_cannot_match_the_anchored_digestmissing-anchored-record	reopen	deployment=7;profile=99;trusted_head=1;record_present=false	reopened=false;genesis_fallback=false	a_missing_anchor_record_does_not_restart_from_genesiswrong-authority-profile	reopen	deployment=7;trusted_profile=99;attachment_profile=98;head=1	reopened=false	independently_anchored_records_construct_a_fresh_endpoint,reopen_from_authenticated_head_defuncertain-commit-confirmed-by-head	commit	deployment=7;profile=99;old_head=0;candidate=1;durable=true;cas_applied=true;cas_reply_lost=true	head=1;reply_exposed_after_verified_head=true;reexecution=false	exact_committed_head_releases_the_prepared_response,exact_new_head_reopens_the_records_only_productcompeting-compare-and-set	commit	deployment=7;profile=99;expected_head=0;current_head=1;candidate=1;previous_reply_lost=true	head_unchanged=true;unconfirmed_reply_exposed=false	losing_a_compare_and_set_does_not_rewrite_the_current_head,new_reply_is_invisible_during_prepare_or_flush

#[cfg(target_os = "linux")]#[derive(Eq, PartialEq)]struct TrustIdentity {    paths: [PathBuf; 3],    digests: [[u8; 32]; 3],}#[cfg(target_os = "linux")]impl TrustIdentity {    fn read(configured: &[String; 3]) -> Result<Self, HeadError> {        let mut paths = [PathBuf::new(), PathBuf::new(), PathBuf::new()];        let mut digests = [[0; 32]; 3];        for ((path, digest), configured) in paths.iter_mut().zip(&mut digests).zip(configured) {            *path = fs::canonicalize(configured).map_err(|_| HeadError)?;            let mut file = fs::File::open(&path).map_err(|_| HeadError)?;            if !file.metadata().map_err(|_| HeadError)?.is_file() {                return Err(HeadError);            }            let mut hash = Sha256::new();            let bytes = std::io::copy(&mut file, &mut hash).map_err(|_| HeadError)?;            if bytes == 0 {                return Err(HeadError);            }            *digest = hash.finalize().into();        }        Ok(Self { paths, digests })    }}

/// Read one current, linearizable control value. Missing keys fail closed.pub fn current(&self) -> Result<ControlValue, HeadError> {    let stdout = self.run(&["get", "--consistency=l", "--", &self.key], None)?;    range_value(        &stdout,        self.cluster_id,        self.key.as_bytes(),        self.max_control_bytes,    )}/// Compare the exact previous value and per-key MOD in one transaction./// A false result means the server reported a definite comparison failure./// Any command or confirmation error may be ambiguous.pub fn replace(&self, expected: &ControlValue, candidate: &[u8]) -> Result<bool, HeadError> {    if expected.revision == 0        || expected.revision > i64::MAX as u64        || expected.bytes.len() > self.max_control_bytes        || candidate.len() > self.max_control_bytes    {        return Err(HeadError);    }    // Bind the configured endpoint before writing. The dedicated TLS    // identity also applies to the separate connection used by the CAS.    self.current()?;    let input = compare_input(&self.key, expected, candidate);    let stdout = self.run(&["txn"], Some(input))?;    let Some(revision) = txn_revision(&stdout, self.cluster_id, expected.revision)? else {        return Ok(false);    };    require_readback(self.current()?, revision, candidate)?;    Ok(true)}

fn compare_input(key: &str, expected: &ControlValue, candidate: &[u8]) -> Vec<u8> {    let key = quoted(key.as_bytes());    let revision = quoted(expected.revision.to_string().as_bytes());    format!(        "mod({key}) = {revision}\nvalue({key}) = {}\n\nput {key} {}\n\n\n",        quoted(&expected.bytes),        quoted(candidate),    )    .into_bytes()}

fn txn_revision(bytes: &[u8], cluster: u64, expected: u64) -> Result<Option<u64>, HeadError> {    let response = TxnResponse::decode(bytes).map_err(|_| HeadError)?;    let revision = verified_revision(response.header, cluster)?;    match (response.succeeded, response.responses.as_slice()) {        (false, []) => Ok(None),        (            true,            [                ResponseOp {                    range: None,                    put: Some(put),                    delete: None,                    txn: None,                },            ],        ) if revision > expected && put.prev_kv.is_none() => {            if let Some(header) = &put.header {                // etcd may omit the nested header or populate only its revision.                if (header.cluster_id != 0 && header.cluster_id != cluster)                    || (header.revision != 0 && header.revision as u64 != revision)                {                    return Err(HeadError);                }            }            Ok(Some(revision))        }        _ => Err(HeadError),    }}

/// Supplied by the host's authenticated direct-mode attachment policy./// A profile requiring an independent witness cannot use this verifier.pub trait DirectAttachmentVerifier {    type Proof;    fn verify_direct_attachment(        &self,        profile: RecoveryProfile,        connected: bool,        proof: &Self::Proof,    ) -> bool;}/// The raw transport can also be used by a separate independent witness./// This attachment wrapper serves only an explicitly authorized direct profile.pub struct EtcdControlStore<A> {    transport: EtcdTransport,    profile: RecoveryProfile,    attachment: A,}impl<A: DirectAttachmentVerifier> EtcdControlStore<A> {    pub fn new(transport: EtcdTransport, profile: RecoveryProfile, attachment: A) -> Self {        Self {            transport,            profile,            attachment,        }    }}

/// Retain only the prefix selected by this exact head. A staged or/// unanchored suffix after that prefix is never included.pub fn retain(&self, source: &mut File, head: AnchorHead) -> Result<ArchiveRef, RecoveryError> {    let prefix_bytes = self.prefix_length(head)?;    if source.metadata().map_err(io_error)?.len() < prefix_bytes {        return Err(RecoveryError::InvalidImage);    }    sync_directory(&self.directory)?;    source.sync_all().map_err(io_error)?;    let mut staged = Staged::create(&self.directory)?;    let prefix_sha256 = copy_prefix(source, prefix_bytes, &mut staged.file)?;    let archive = ArchiveRef {        head,        prefix_bytes,        prefix_sha256,    };    self.verify(&mut staged.file, archive)?;    staged.file.sync_all().map_err(io_error)?;    let destination = self.archive_path(archive);    match fs::hard_link(&staged.path, &destination) {        Ok(()) => {}        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {            // A concurrent or retried identical put must validate the            // existing bytes and reestablish the directory barrier.            self.open_verified(archive)?;        }        Err(_) => return Err(RecoveryError::Io),    }    sync_directory(&self.directory)?;    Ok(archive)}

#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]struct Selection {    profile: RecoveryProfile,    authority: [u8; 32],    witness: [u8; 32],    control: Vec<u8>,    archive: ArchiveRef,    floor: AnchorHead,    writer: WitnessWriter,    previous: Option<Mark>,    repair_nonce: [u8; 32],    authorization: Option<[u8; 32]>,}

fn publish_independent(    &self,    file: &mut File,    candidate: AnchorHead,    admission: &OrderedPublication,    checkpoint: CheckpointFence,) -> Result<Result<(), PublicationRefusal>, HeadError> {    let store = &self.ordered.store;    let previous = store.current_selection()?;    if previous.control() != admission.cut.stored || previous.checkpoint() != checkpoint {        return Ok(Err(PublicationRefusal::CheckpointChanged));    }    let Some(state) = proposed_publication::<WitnessStore<A>>(candidate, admission)? else {        return Ok(Err(PublicationRefusal::CheckpointChanged));    };    let archive = store        .archive        .retain(file, candidate)        .map_err(|_| HeadError)?;    let mut staged = store        .archive        .open_verified(archive)        .map_err(|_| HeadError)?;    store        .archive        .verify_journal_prefix(&mut staged, previous.envelope.archive)        .map_err(|_| HeadError)?;    let proposal = store.propose(&previous, state.encode()?, archive, store.writer);    store.select(&previous, &proposal).map(|selected| {        if selected {            Ok(())        } else {            Err(PublicationRefusal::CheckpointChanged)        }    })}

/// Construct a fresh service without an earlier product or snapshot.pub fn open(    file: File,    anchor: H,    attachment: Attachment<S, H::AttachmentProof>,) -> Result<Self, RecoveryError> {    let mut service = Self {        file,        anchor,        profile: attachment.profile,        proof: attachment.proof,        security: Arc::new(attachment.security),        source_connected: attachment.source_connected,        independent_checkpoint: None,    };    service.transact(|_| ())?;    Ok(service)}

fn transact_guarded<T>(    &mut self,    act: impl FnOnce(&H, AnchorHead, &mut ProductAuthority<S>) -> std::ops::ControlFlow<T, T>,    wire: bool,    before_publish: impl FnOnce() -> Result<Option<PublicationGuard>, RecoveryError>,) -> Result<T, RecoveryError> {    Self::check_attachment(        &self.anchor,        self.profile,        self.source_connected,        &self.proof,    )?;    let locked = LockedFile::acquire(&mut self.file)?;    let head = self        .anchor        .current()        .map_err(|_| RecoveryError::Hold)?        .ok_or(RecoveryError::MissingHead)?;    if head.profile != self.profile {        return Err(RecoveryError::AttachmentRejected);    }    let checkpoint = Self::read_configured_checkpoint(        &self.anchor, self.profile, self.source_connected, &self.proof, self.independent_checkpoint,    )?;    let independent = matches!(checkpoint, CheckpointState::Independent(_));    let before = exact_image(locked.0, head)?;    if !before.image.accepts_security(self.security.as_ref()) {        return Err(RecoveryError::AttachmentRejected);    }    match checkpoint {        CheckpointState::Independent(witness) => {            if witness.protected_head != Some(head) {                return Err(RecoveryError::Hold);            }            let floor = witness.head.ok_or(RecoveryError::Hold)?;            validate_lineage(locked.0, head, &before, floor)?;            self.anchor.verify_independent_journal(locked.0, head).map_err(|_| RecoveryError::Hold)?;        }        CheckpointState::Direct => {}    }

original-signed-debit-retry	retry	actor=7;epoch=9;source=0;receiver=2;asset=17;event=17;amount=5;source_initial=10;exact_original_request=true;cut_advanced=true;reopened=true	reply=retained;new_effects=0;source_balance=5;pending=5	an_admitted_exact_retry_checks_the_retained_result_before_fresh_conditions,prior_call_on_any_route_cannot_repeat_an_authority_effect,source_debit_puts_five_into_pendingcompleted-busy-survives-reopen	retry	actor=7;epoch=9;namespace=7;sequence=1;outcome=busy;anchor_committed=true;reopened=true	reply=busy;new_effects=0	retained_request_and_busy_outcome_survive_fresh_reconstruction,an_admitted_exact_retry_checks_the_retained_result_before_fresh_conditionsadmitted-unavailable-survives-reopen	retry	actor=7;epoch=9;namespace=7;sequence=1;authority_reachable=true;source_connected=false;anchor_committed=true	reply=unavailable;new_effects=0	disconnected_admitted_call_retains_unavailable_without_source_effect,product_round_trip_preserves_the_durable_projection

cache-revalidates-on-cold-open	current-read	provider=1;epoch=9;asset=17;query=application_value;same_authority=true;same_cut=true;same_value=true;reopened=true	reply=current;value_preserved=true	a_valid_same_authority_cache_preserves_the_actual_current_response,authenticated_records_only_recovery_is_a_forward_simulationcache-claimed-value-must-be-actual	current-read	provider=1;epoch=9;asset=17;query=application_value;same_authority=true;same_cut=true;same_value=false;reopened=true	reply=busy;forged_value_exposed=false	wrong_cache_value_cannot_be_returned_as_current

fn source_probe_at<S: Security>(    product: &mut ProductAuthority<S>,    maintenance: &S::MaintenanceToken,    origin: OriginKey,    evidence: Option<TerminalEvidence>,) -> Result<SourceAction, RecoveryError> {    if product.maintenance(maintenance).is_none() {        return Err(RecoveryError::Endpoint(EndpointError::Rejected));    }    let authority = &product.reads.authority;    let Some(binding) = authority.state.primary.ledger.open_source_binding(origin) else {        return Ok(SourceAction::None);    };    if !authority.connected {        return Ok(SourceAction::Hold(binding));    }    match evidence {        None => Ok(SourceAction::Query(binding)),        Some(evidence)            if evidence.binding == binding                && product.security.verify_recovery_evidence(evidence) =>        {            match evidence.kind {                TerminalKind::Confirmed => Ok(SourceAction::ConsiderConfirmed(binding)),                TerminalKind::Reversed => Ok(SourceAction::ConsiderReversed(binding)),            }        }        Some(_) => Ok(SourceAction::Hold(binding)),    }}

dispatched-pending-needs-query	source-probe	source=0;receiver=2;asset=17;event=17;initial=10;pending=5;reopened=true;source_connected=true;evidence=none	next=query_exact;event=17;pending=5;source_balance=5	recovery_queries_the_same_origin_and_holds_mismatched_evidence,recovered_world_keeps_the_actual_pending_reservationwrong-origin-certificate-holds	source-probe	source=0;receiver=2;asset=17;event=17;pending=5;evidence_event=22;reopened=true	next=hold;event=17;pending=5;refund=0	recovery_queries_the_same_origin_and_holds_mismatched_evidence,mismatched_source_evidence_only_holds_the_exact_pending_bindingno-reversal-no-source-return	source-return	source=0;receiver=2;asset=17;event=17;pending=5;source_balance=5;exact_reversal=false;reopened=true	source_balance=5;pending=5;refund=0	missing_or_wrong_reversal_cannot_move_the_source_balance,pending_source_rejects_cancel_and_fence

/// Wait for an ordered worker's reply until the caller's monotonic deadline./// The worker keeps its authority to finish. A timed-out or disconnected/// waiter must resolve the original signed request before reissuing the work.pub fn wait_for_ordered_wire_reply<T>(    reply: &Receiver<Result<T, WireError>>,    deadline: Instant,) -> Result<T, WireError> {    let now = Instant::now();    if now >= deadline {        return Err(WireError::RequestUnresolved);    }    let result = reply.recv_timeout(deadline - now)        .map_err(|_| WireError::RequestUnresolved)?;    if Instant::now() >= deadline {        return Err(WireError::RequestUnresolved);    }    result}

fn error(error: WireError) -> Self {    use ResponseCode::*;    Self::plain(match error {        WireError::InvalidFormat => InvalidFormat,        WireError::UnsupportedVersion => UnsupportedVersion,        WireError::InvalidSource => InvalidSource,        WireError::InvalidSignature => InvalidSignature,        WireError::InvalidThresholdSignature => InvalidThresholdSignature,        WireError::TtlInvalid => TtlInvalid,        WireError::ReplayConflict => ReplayConflict,        WireError::ProductRejected        | WireError::Recovery(RecoveryError::Endpoint(EndpointError::Rejected)) => {            ProductRejected        }        WireError::TtlExpired => TtlExpired,        WireError::Cancelled => Cancelled,        WireError::DeadlineExpired => DeadlineExpired,        WireError::ResponseLost => ResponseLost,        WireError::RequestUnresolved => Unresolved,        WireError::Recovery(RecoveryError::Endpoint(EndpointError::Unavailable)) => {            DeliveryUnavailable        }        WireError::Recovery(_) => Unresolved,    })}

#[test]fn reference_response_distinguishes_delivery_and_authentication() {    use response::ResponseCode;    let path = JournalPath::create();    let mut service = memory_endpoint(&path);    let channel = channel(0);    let original = signed(debit_core(2));    let before = service.anchor.current().unwrap().unwrap();    let unresolved = service.resolve_wire_debit_response(&channel, &original, &wire_auth(), false).unwrap();    assert_response(&original, &unresolved, 2, ResponseCode::Unresolved, None);    // A frame whose own signature does not verify receives no signed answer.    let mut bad_signature = original.clone();    *bad_signature.last_mut().unwrap() ^= 1;    assert_eq!(service.receive_wire_response(        &channel, Entry::Adapter, &bad_signature, &wire_auth(), DeliveryWindow::default(),    ), Err(WireError::InvalidSignature));    assert_eq!(        service.resolve_wire_debit_response(&channel, &bad_signature, &wire_auth(), false),        Err(WireError::InvalidSignature),    );    assert_eq!(service.receive_wire_response(        &channel, Entry::Adapter, &[0], &wire_auth(), DeliveryWindow::default(),    ), Err(WireError::InvalidFormat));    assert_eq!(service.anchor.current().unwrap(), Some(before));    let lost = service.receive_wire_response(        &channel, Entry::Adapter, &original, &wire_auth(),        DeliveryWindow { response_lost: true, ..DeliveryWindow::default() },    ).unwrap();    assert_response(&original, &lost, 2, ResponseCode::ResponseLost, None);    let after = service.anchor.current().unwrap().unwrap();    assert_eq!(after.sequence, before.sequence + 1);    let other = &super::super::super::vector_tests::SUCCESSOR;    let refusal = service.resolve_wire_debit_response(        &channel, &original, &WireCredentials { token: other, ..wire_auth() }, false,    ).unwrap();    assert_response(&original, &refusal, 3, ResponseCode::ProductRejected, None);    let receipt = service.resolve_wire_debit_response(&channel, &original, &wire_auth(), false).unwrap();    assert_response(&original, &receipt, 1, ResponseCode::Debited, Some(2));    assert_eq!(service.anchor.current().unwrap(), Some(after));}