From bf6e6c7bd4febae5354766d34b74b8a353000741 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 4 Sep 2026 12:01:05 +0100 Subject: [PATCH 1/6] Defer recovery restart until commit Request host restart only from the committed JOINING state hook so aborted recovery transactions cannot trigger a restart. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 1 + src/node/recovery_decision_protocol.cpp | 13 +++++++++++-- src/node/recovery_decision_protocol.h | 1 + 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d48ef750fe5..5198e2a58ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Ledger chunk metadata and snapshot scheduling are no longer restored by a transaction whose writes a concurrent view change has already discarded. Both are now updated under the same lock as the rollback, and skipped when the transaction's rollback epoch or view no longer holds (#8243). - A transaction whose view changed while it was committing could apply its writes to the local key-value store and then fail to replicate, leaving state that never reached consensus. The transaction's view is now validated atomically with the allocation of its version, so it is rejected before any map is modified, and `ccf::kv::CommitResult::FAIL_NO_REPLICATE` no longer implies a locally applied write (#8242). - A transaction in a JavaScript application endpoint which conflicts with compaction is now re-executed, rather than returning `500 Internal Server Error` (#8289). +- Recovery-decision-protocol nodes now request host restart only after the `JOINING` state transaction commits, preventing restart for an aborted transaction. (#8282) ### Changed diff --git a/src/node/recovery_decision_protocol.cpp b/src/node/recovery_decision_protocol.cpp index 31a1785985b..4f78167200d 100644 --- a/src/node/recovery_decision_protocol.cpp +++ b/src/node/recovery_decision_protocol.cpp @@ -25,6 +25,11 @@ namespace ccf node_state(node_state_) {} + void RecoveryDecisionProtocolSubsystem::restart_after_commit() + { + RINGBUFFER_WRITE_MESSAGE(AdminMessage::restart, node_state->to_host); + } + void RecoveryDecisionProtocolSubsystem::reset_state(ccf::kv::Tx& tx) { // Clear any previous state @@ -90,6 +95,12 @@ namespace ccf start_message_retry_timers(); start_failover_timers(); } + else if ( + w.has_value() && + w.value() == recovery_decision_protocol::StateMachine::JOINING) + { + restart_after_commit(); + } })); } @@ -245,8 +256,6 @@ namespace ccf auto service_cert = ccf::crypto::cert_der_to_pem(node_config->service_cert_der); LOG_INFO_FMT("{}", service_cert.str()); - - RINGBUFFER_WRITE_MESSAGE(AdminMessage::restart, node_state->to_host); } case recovery_decision_protocol::StateMachine::OPENING: { diff --git a/src/node/recovery_decision_protocol.h b/src/node/recovery_decision_protocol.h index 6c761ad6584..6833c9ea772 100644 --- a/src/node/recovery_decision_protocol.h +++ b/src/node/recovery_decision_protocol.h @@ -72,6 +72,7 @@ namespace ccf // Stop periodic tasks void stop_timers(); + void restart_after_commit(); // Steady state operations recovery_decision_protocol::RequestNodeInfo& get_node_info( From 5d2cfb46aeed8dc6e00558159ec661423d81056b Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 4 Sep 2026 12:01:43 +0100 Subject: [PATCH 2/6] Add commit-aware recovery tracing Gate versioned RDP_TRACE records behind CCF_RECOVERY_TRACE and publish receive, timeout, effect, and retry-send events only at their required commit boundaries. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CMakeLists.txt | 9 + .../ccf/service/tables/self_healing_open.h | 26 ++ src/node/recovery_decision_protocol.cpp | 416 ++++++++++++++++-- src/node/recovery_decision_protocol.h | 68 ++- src/node/rpc/self_healing_open_handlers.h | 69 ++- 5 files changed, 548 insertions(+), 40 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index dee27f1f816..d641e52c49a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -454,6 +454,15 @@ if(CCF_RAFT_TRACING) add_compile_definitions(CCF_RAFT_TRACING) endif() +option( + CCF_RECOVERY_TRACE + "Enable committed recovery-decision-protocol tracing" + OFF +) +if(CCF_RECOVERY_TRACE) + add_compile_definitions(CCF_RECOVERY_TRACE) +endif() + # Build common library for CCF enclaves set( CCF_IMPL_SOURCE diff --git a/include/ccf/service/tables/self_healing_open.h b/include/ccf/service/tables/self_healing_open.h index 20066451e8e..21f6dfed856 100644 --- a/include/ccf/service/tables/self_healing_open.h +++ b/include/ccf/service/tables/self_healing_open.h @@ -94,6 +94,28 @@ namespace ccf using TimeoutSMState = ServiceValue; using OpenKind = ServiceValue; + +#ifdef CCF_RECOVERY_TRACE + struct TraceEvent + { + std::string kind; + std::optional message_id = std::nullopt; + std::optional caused_by = std::nullopt; + std::optional source = std::nullopt; + std::optional view = std::nullopt; + std::optional seqno = std::nullopt; + std::string pre; + std::string post; + std::optional open_kind = std::nullopt; + std::optional send = std::nullopt; + }; + DECLARE_JSON_TYPE_WITH_OPTIONAL_FIELDS(TraceEvent); + DECLARE_JSON_REQUIRED_FIELDS(TraceEvent, kind, pre, post); + DECLARE_JSON_OPTIONAL_FIELDS( + TraceEvent, message_id, caused_by, source, view, seqno, open_kind, send); + + using TraceEvents = ServiceMap; +#endif } namespace Tables @@ -112,5 +134,9 @@ namespace ccf "public:ccf.gov.recovery_decision_protocol.timeout_sm_state"; static constexpr auto RECOVERY_DECISION_PROTOCOL_OPEN_KIND = "public:ccf.gov.recovery_decision_protocol.open_kind"; +#ifdef CCF_RECOVERY_TRACE + static constexpr auto RECOVERY_DECISION_PROTOCOL_TRACE_EVENTS = + "public:ccf.internal.recovery_decision_protocol.trace_events"; +#endif } } diff --git a/src/node/recovery_decision_protocol.cpp b/src/node/recovery_decision_protocol.cpp index 4f78167200d..17d2d111e7a 100644 --- a/src/node/recovery_decision_protocol.cpp +++ b/src/node/recovery_decision_protocol.cpp @@ -19,6 +19,45 @@ namespace ccf { +#ifdef CCF_RECOVERY_TRACE + static constexpr auto RECOVERY_TRACE_VERSION = + "ccf.recovery_decision_protocol.trace/1"; + static constexpr auto RECOVERY_TRACE_MARKER = "RDP_TRACE"; + + static std::string trace_state_name( + recovery_decision_protocol::StateMachine state) + { + switch (state) + { + case recovery_decision_protocol::StateMachine::GOSSIPING: + return "GOSSIPING"; + case recovery_decision_protocol::StateMachine::VOTING: + return "VOTING"; + case recovery_decision_protocol::StateMachine::OPENING: + return "OPENING"; + case recovery_decision_protocol::StateMachine::JOINING: + return "JOINING"; + case recovery_decision_protocol::StateMachine::OPEN: + return "OPEN"; + default: + throw std::logic_error("Unknown recovery-decision-protocol state"); + } + } + + static std::string trace_open_kind_name( + recovery_decision_protocol::OpenKinds kind) + { + switch (kind) + { + case recovery_decision_protocol::OpenKinds::QUORUM: + return "QUORUM"; + case recovery_decision_protocol::OpenKinds::FAILOVER: + return "FAILOVER"; + default: + throw std::logic_error("Unknown recovery-decision-protocol open kind"); + } + } +#endif RecoveryDecisionProtocolSubsystem::RecoveryDecisionProtocolSubsystem( NodeState* node_state_) : @@ -30,6 +69,235 @@ namespace ccf RINGBUFFER_WRITE_MESSAGE(AdminMessage::restart, node_state->to_host); } +#ifdef CCF_RECOVERY_TRACE + void RecoveryDecisionProtocolSubsystem::initialise_trace(ccf::kv::Tx& tx) + { + const auto previous_service_cert = + tx.ro(node_state->network.previous_service_identity)->get(); + if (!previous_service_cert.has_value()) + { + throw std::logic_error( + "Previous service identity not found while initialising " + "recovery-decision-protocol tracing"); + } + + { + std::lock_guard guard(trace_lock); + next_trace_record_id = 0; + next_trace_sequence = 0; + next_trace_message_number = 0; + trace_instance_id = + recovery_decision_protocol::service_fingerprint_from_pem( + previous_service_cert.value()); + trace_node = get_location().name; + trace_committed_state = "GOSSIPING"; + trace_expected_locations.clear(); + for (const auto& location : get_config().expected_locations) + { + trace_expected_locations.push_back(location.name); + } + } + + node_state->network.tables->set_global_hook( + Tables::RECOVERY_DECISION_PROTOCOL_TRACE_EVENTS, + recovery_decision_protocol::TraceEvents::wrap_commit_hook( + [this]( + ccf::kv::Version, + const recovery_decision_protocol::TraceEvents::Write& writes) { + for (const auto& [_, event] : writes) + { + if (event.has_value()) + { + emit_trace_event(event.value()); + if (event->kind == "join_restart") + { + restart_after_commit(); + } + } + } + })); + } + + void RecoveryDecisionProtocolSubsystem::emit_trace_event( + recovery_decision_protocol::TraceEvent event) + { + std::lock_guard guard(trace_lock); + emit_trace_event_unsafe(std::move(event)); + } + + void RecoveryDecisionProtocolSubsystem::emit_trace_event_unsafe( + recovery_decision_protocol::TraceEvent event) + { + if (event.kind == "send") + { + event.pre = trace_committed_state; + event.post = trace_committed_state; + } + else + { + trace_committed_state = event.post; + } + nlohmann::json trace = event; + trace["version"] = RECOVERY_TRACE_VERSION; + trace["instance"] = trace_instance_id; + trace["expected_locations"] = trace_expected_locations; + trace["node"] = trace_node; + trace["sequence"] = next_trace_sequence++; + LOG_INFO_FMT("{} {}", RECOVERY_TRACE_MARKER, trace.dump()); + } + + void RecoveryDecisionProtocolSubsystem::record_trace_event( + ccf::kv::Tx& tx, recovery_decision_protocol::TraceEvent event) + { + uint64_t record_id = 0; + { + std::lock_guard guard(trace_lock); + record_id = next_trace_record_id++; + } + tx.rw( + Tables::RECOVERY_DECISION_PROTOCOL_TRACE_EVENTS) + ->put(record_id, std::move(event)); + } + + std::string RecoveryDecisionProtocolSubsystem::new_trace_message_id() + { + std::lock_guard guard(trace_lock); + return new_trace_message_id_unsafe(); + } + + std::string RecoveryDecisionProtocolSubsystem::new_trace_message_id_unsafe() + { + return fmt::format( + "{}:{}:{}", trace_instance_id, trace_node, next_trace_message_number++); + } + + recovery_decision_protocol::StateMachine RecoveryDecisionProtocolSubsystem:: + get_trace_state(kv::ReadOnlyTx& tx) + { + const auto state = tx.ro( + Tables::RECOVERY_DECISION_PROTOCOL_SM_STATE) + ->get(); + if (!state.has_value()) + { + throw std::logic_error( + "Recovery-decision-protocol state not set while tracing"); + } + return state.value(); + } + + void RecoveryDecisionProtocolSubsystem::emit_trace_send_unsafe( + const std::string& message_id, + const std::string& description, + const std::optional& txid) + { + recovery_decision_protocol::TraceEvent event{ + .kind = "send", + .message_id = message_id, + .pre = "", + .post = "", + .send = description, + }; + if (txid.has_value()) + { + event.view = txid->view; + event.seqno = txid->seqno; + } + emit_trace_event_unsafe(std::move(event)); + } + + void RecoveryDecisionProtocolSubsystem::record_trace_effects( + ccf::kv::Tx& tx, + recovery_decision_protocol::StateMachine pre, + recovery_decision_protocol::StateMachine post) + { + if ( + post == recovery_decision_protocol::StateMachine::OPENING && + pre != recovery_decision_protocol::StateMachine::OPENING) + { + const auto open_kind = tx.ro( + Tables::RECOVERY_DECISION_PROTOCOL_OPEN_KIND) + ->get(); + if (!open_kind.has_value()) + { + throw std::logic_error( + "Recovery-decision-protocol open kind not set while tracing"); + } + record_trace_event( + tx, + { + .kind = "open", + .pre = "OPENING", + .post = "OPENING", + .open_kind = trace_open_kind_name(open_kind.value()), + }); + } + + if (post == recovery_decision_protocol::StateMachine::JOINING) + { + record_trace_event( + tx, + { + .kind = "join_restart", + .pre = "JOINING", + .post = "JOINING", + }); + } + + if ( + pre == recovery_decision_protocol::StateMachine::OPENING && + post == recovery_decision_protocol::StateMachine::OPEN) + { + record_trace_event( + tx, + { + .kind = "complete", + .pre = "OPEN", + .post = "OPEN", + }); + } + } + + void RecoveryDecisionProtocolSubsystem::record_trace_receive( + ccf::kv::Tx& tx, + const std::string& kind, + const std::optional& caused_by, + const std::string& source, + const std::optional& txid, + recovery_decision_protocol::StateMachine pre) + { + const auto post = get_trace_state(tx); + recovery_decision_protocol::TraceEvent event{ + .kind = kind, + .message_id = new_trace_message_id(), + .caused_by = caused_by, + .source = source, + .pre = trace_state_name(pre), + .post = trace_state_name(post), + }; + if (txid.has_value()) + { + event.view = txid->view; + event.seqno = txid->seqno; + } + record_trace_event(tx, std::move(event)); + record_trace_effects(tx, pre, post); + } + + void RecoveryDecisionProtocolSubsystem::record_trace_timeout( + ccf::kv::Tx& tx, recovery_decision_protocol::StateMachine pre) + { + const auto post = get_trace_state(tx); + record_trace_event( + tx, + { + .kind = "timeout", + .pre = trace_state_name(pre), + .post = trace_state_name(post), + }); + record_trace_effects(tx, pre, post); + } +#endif + void RecoveryDecisionProtocolSubsystem::reset_state(ccf::kv::Tx& tx) { // Clear any previous state @@ -54,6 +322,11 @@ namespace ccf tx.rw( Tables::RECOVERY_DECISION_PROTOCOL_OPEN_KIND) ->clear(); +#ifdef CCF_RECOVERY_TRACE + tx.rw( + Tables::RECOVERY_DECISION_PROTOCOL_TRACE_EVENTS) + ->clear(); +#endif } void RecoveryDecisionProtocolSubsystem::try_start( @@ -74,6 +347,10 @@ namespace ccf LOG_INFO_FMT("Starting recovery-decision-protocol"); +#ifdef CCF_RECOVERY_TRACE + initialise_trace(tx); +#endif + tx.rw( Tables::RECOVERY_DECISION_PROTOCOL_SM_STATE) ->put(recovery_decision_protocol::StateMachine::GOSSIPING); @@ -92,6 +369,13 @@ namespace ccf w.has_value() && w.value() == recovery_decision_protocol::StateMachine::GOSSIPING) { +#ifdef CCF_RECOVERY_TRACE + emit_trace_event({ + .kind = "start", + .pre = "GOSSIPING", + .post = "GOSSIPING", + }); +#endif start_message_retry_timers(); start_failover_timers(); } @@ -99,7 +383,11 @@ namespace ccf w.has_value() && w.value() == recovery_decision_protocol::StateMachine::JOINING) { +#ifndef CCF_RECOVERY_TRACE restart_after_commit(); +#else + // The trace-event commit hook emits join_restart before restarting. +#endif } })); } @@ -342,10 +630,21 @@ namespace ccf return; } + std::optional + gossip_request = std::nullopt; + std::optional + vote_request = std::nullopt; + std::optional chosen_node_info = + std::nullopt; + std::optional + iamopen_request = std::nullopt; + switch (sm_state) { case recovery_decision_protocol::StateMachine::GOSSIPING: - send_gossip_unsafe(tx); + gossip_request = recovery_decision_protocol::GossipRequest{}; + gossip_request->info = get_node_info(tx); + gossip_request->txid = get_last_recovered_signed_txid(); break; case recovery_decision_protocol::StateMachine::VOTING: { @@ -360,7 +659,7 @@ namespace ccf throw std::logic_error( "Recovery-decision-protocol chosen node not set, cannot vote"); } - auto chosen_node_info = + chosen_node_info = node_info_handle->get(chosen_replica_handle->get().value()); if (!chosen_node_info.has_value()) { @@ -368,13 +667,15 @@ namespace ccf "Recovery-decision-protocol chosen node {} not found", chosen_replica_handle->get().value())); } - send_vote_unsafe(tx, chosen_node_info.value()); - // keep gossiping to allow lagging nodes to eventually vote - send_gossip_unsafe(tx); + vote_request = recovery_decision_protocol::TaggedWithNodeInfo{ + .info = get_node_info(tx)}; + gossip_request = recovery_decision_protocol::GossipRequest{}; + gossip_request->info = vote_request->info; + gossip_request->txid = get_last_recovered_signed_txid(); break; } case recovery_decision_protocol::StateMachine::OPENING: - send_iamopen_unsafe(tx); + iamopen_request = get_iamopen_request(tx); break; case recovery_decision_protocol::StateMachine::JOINING: case recovery_decision_protocol::StateMachine::OPEN: @@ -385,6 +686,47 @@ namespace ccf "Unknown recovery-decision-protocol state: {}", static_cast(sm_state))); } + + const auto self_signed_node_cert = + node_state->get_self_signed_certificate(); + const auto node_private_key = + node_state->node_sign_kp->private_key_pem(); + +#ifdef CCF_RECOVERY_TRACE + std::lock_guard trace_guard(trace_lock); + if (trace_committed_state != trace_state_name(sm_state)) + { + return; + } +#endif + + switch (sm_state) + { + case recovery_decision_protocol::StateMachine::GOSSIPING: + send_gossip_unsafe( + gossip_request.value(), self_signed_node_cert, node_private_key); + break; + case recovery_decision_protocol::StateMachine::VOTING: + send_vote_unsafe( + vote_request.value(), + chosen_node_info.value(), + self_signed_node_cert, + node_private_key); + // Keep gossiping to allow lagging nodes to eventually vote. + send_gossip_unsafe( + gossip_request.value(), self_signed_node_cert, node_private_key); + break; + case recovery_decision_protocol::StateMachine::OPENING: + send_iamopen_unsafe( + iamopen_request.value(), self_signed_node_cert, node_private_key); + break; + case recovery_decision_protocol::StateMachine::JOINING: + case recovery_decision_protocol::StateMachine::OPEN: + default: + throw std::logic_error(fmt::format( + "Unexpected prepared recovery-decision-protocol state: {}", + static_cast(sm_state))); + } }, "RecoveryDecisionProtocolRetry"); @@ -590,23 +932,28 @@ namespace ccf return node_info_cache.value(); } - void RecoveryDecisionProtocolSubsystem::send_gossip_unsafe(kv::ReadOnlyTx& tx) + void RecoveryDecisionProtocolSubsystem::send_gossip_unsafe( + recovery_decision_protocol::GossipRequest request, + const crypto::Pem& self_signed_node_cert, + const crypto::Pem& node_private_key) { auto& config = get_config(); LOG_TRACE_FMT("Broadcasting recovery-decision-protocol gossip"); - recovery_decision_protocol::GossipRequest request; - request.info = get_node_info(tx); - request.txid = get_last_recovered_signed_txid(); - nlohmann::json request_json = request; - const auto self_signed_node_cert = - node_state->get_self_signed_certificate(); - const auto node_private_key = node_state->node_sign_kp->private_key_pem(); - for (auto& target : config.expected_locations) { auto target_address = target.address; +#ifdef CCF_RECOVERY_TRACE + request.trace_message_id = new_trace_message_id_unsafe(); +#endif + nlohmann::json request_json = request; +#ifdef CCF_RECOVERY_TRACE + emit_trace_send_unsafe( + request.trace_message_id.value(), + fmt::format("gossip:{}", target.name), + request.txid); +#endif dispatch_authenticated_message( request_json, target_address, @@ -617,25 +964,31 @@ namespace ccf } void RecoveryDecisionProtocolSubsystem::send_vote_unsafe( - kv::ReadOnlyTx& tx, const recovery_decision_protocol::NodeInfo& node_info) + recovery_decision_protocol::TaggedWithNodeInfo request, + const recovery_decision_protocol::NodeInfo& node_info, + const crypto::Pem& self_signed_node_cert, + const crypto::Pem& node_private_key) { LOG_TRACE_FMT( "Sending recovery-decision-protocol vote to {} at {}", node_info.location.name, node_info.location.address); - recovery_decision_protocol::TaggedWithNodeInfo request{ - .info = get_node_info(tx)}; +#ifdef CCF_RECOVERY_TRACE + request.trace_message_id = new_trace_message_id_unsafe(); +#endif nlohmann::json request_json = request; - const auto self_signed_node_cert = - node_state->get_self_signed_certificate(); - +#ifdef CCF_RECOVERY_TRACE + emit_trace_send_unsafe( + request.trace_message_id.value(), + fmt::format("vote:{}", node_info.location.name)); +#endif dispatch_authenticated_message( request_json, node_info.location.address, "vote", self_signed_node_cert, - node_state->node_sign_kp->private_key_pem()); + node_private_key); } recovery_decision_protocol::IAmOpenRequest& @@ -676,18 +1029,14 @@ namespace ccf } void RecoveryDecisionProtocolSubsystem::send_iamopen_unsafe( - ccf::kv::ReadOnlyTx& tx) + recovery_decision_protocol::IAmOpenRequest request, + const crypto::Pem& self_signed_node_cert, + const crypto::Pem& node_private_key) { auto& config = get_config(); auto& location = get_location(); LOG_TRACE_FMT("Sending recovery-decision-protocol iamopen"); - - nlohmann::json request_json = get_iamopen_request(tx); - const auto self_signed_node_cert = - node_state->get_self_signed_certificate(); - const auto node_private_key = node_state->node_sign_kp->private_key_pem(); - for (auto& target : config.expected_locations) { if (target.name == location.name) @@ -695,6 +1044,15 @@ namespace ccf // Don't send to self continue; } +#ifdef CCF_RECOVERY_TRACE + request.trace_message_id = new_trace_message_id_unsafe(); +#endif + nlohmann::json request_json = request; +#ifdef CCF_RECOVERY_TRACE + emit_trace_send_unsafe( + request.trace_message_id.value(), + fmt::format("iamopen:{}", target.name)); +#endif dispatch_authenticated_message( request_json, target.address, diff --git a/src/node/recovery_decision_protocol.h b/src/node/recovery_decision_protocol.h index 6833c9ea772..c822d8bcee2 100644 --- a/src/node/recovery_decision_protocol.h +++ b/src/node/recovery_decision_protocol.h @@ -16,9 +16,18 @@ namespace ccf::recovery_decision_protocol { public: RequestNodeInfo info; +#ifdef CCF_RECOVERY_TRACE + std::optional trace_message_id = std::nullopt; +#endif }; +#ifdef CCF_RECOVERY_TRACE + DECLARE_JSON_TYPE_WITH_OPTIONAL_FIELDS(TaggedWithNodeInfo); + DECLARE_JSON_REQUIRED_FIELDS(TaggedWithNodeInfo, info); + DECLARE_JSON_OPTIONAL_FIELDS(TaggedWithNodeInfo, trace_message_id); +#else DECLARE_JSON_TYPE(TaggedWithNodeInfo); DECLARE_JSON_REQUIRED_FIELDS(TaggedWithNodeInfo, info); +#endif struct GossipRequest : public TaggedWithNodeInfo { @@ -56,6 +65,17 @@ namespace ccf std::optional iamopen_request_cache; +#ifdef CCF_RECOVERY_TRACE + ds::Mutex trace_lock; + uint64_t next_trace_record_id = 0; + uint64_t next_trace_sequence = 0; + uint64_t next_trace_message_number = 0; + std::string trace_instance_id; + std::vector trace_expected_locations; + std::string trace_node; + std::string trace_committed_state; +#endif + public: RecoveryDecisionProtocolSubsystem(NodeState* node_state); void reset_state(ccf::kv::Tx& tx); @@ -65,6 +85,20 @@ namespace ccf recovery_decision_protocol::IAmOpenRequest& get_iamopen_request( kv::ReadOnlyTx& tx); +#ifdef CCF_RECOVERY_TRACE + recovery_decision_protocol::StateMachine get_trace_state( + kv::ReadOnlyTx& tx); + void record_trace_receive( + ccf::kv::Tx& tx, + const std::string& kind, + const std::optional& caused_by, + const std::string& source, + const std::optional& txid, + recovery_decision_protocol::StateMachine pre); + void record_trace_timeout( + ccf::kv::Tx& tx, recovery_decision_protocol::StateMachine pre); +#endif + private: // Start path void start_message_retry_timers(); @@ -77,14 +111,40 @@ namespace ccf // Steady state operations recovery_decision_protocol::RequestNodeInfo& get_node_info( kv::ReadOnlyTx& tx); - void send_gossip_unsafe(kv::ReadOnlyTx& tx); + void send_gossip_unsafe( + recovery_decision_protocol::GossipRequest request, + const crypto::Pem& self_signed_node_cert, + const crypto::Pem& node_private_key); void send_vote_unsafe( - kv::ReadOnlyTx& tx, - const recovery_decision_protocol::NodeInfo& node_info); - void send_iamopen_unsafe(kv::ReadOnlyTx& tx); + recovery_decision_protocol::TaggedWithNodeInfo request, + const recovery_decision_protocol::NodeInfo& node_info, + const crypto::Pem& self_signed_node_cert, + const crypto::Pem& node_private_key); + void send_iamopen_unsafe( + recovery_decision_protocol::IAmOpenRequest request, + const crypto::Pem& self_signed_node_cert, + const crypto::Pem& node_private_key); RecoveryDecisionProtocolConfig& get_config(); sealing_recovery::Location& get_location(); ccf::TxID get_last_recovered_signed_txid(); + +#ifdef CCF_RECOVERY_TRACE + void initialise_trace(ccf::kv::Tx& tx); + void record_trace_event( + ccf::kv::Tx& tx, recovery_decision_protocol::TraceEvent event); + void record_trace_effects( + ccf::kv::Tx& tx, + recovery_decision_protocol::StateMachine pre, + recovery_decision_protocol::StateMachine post); + void emit_trace_event(recovery_decision_protocol::TraceEvent event); + void emit_trace_event_unsafe(recovery_decision_protocol::TraceEvent event); + std::string new_trace_message_id(); + std::string new_trace_message_id_unsafe(); + void emit_trace_send_unsafe( + const std::string& message_id, + const std::string& description, + const std::optional& txid = std::nullopt); +#endif }; } diff --git a/src/node/rpc/self_healing_open_handlers.h b/src/node/rpc/self_healing_open_handlers.h index e61dfc65936..6cbf1be6d57 100644 --- a/src/node/rpc/self_healing_open_handlers.h +++ b/src/node/rpc/self_healing_open_handlers.h @@ -15,6 +15,8 @@ #include "node/recovery_decision_protocol.h" #include "node/rpc/node_frontend_utils.h" +#include + namespace ccf::node { template @@ -25,9 +27,10 @@ namespace ccf::node template static HandlerJsonParamsAndForward wrap_recovery_decision_protocol( RecoveryDecisionProtocolHandler cb, - ccf::AbstractNodeContext& node_context) + ccf::AbstractNodeContext& node_context, + const std::string& trace_kind) { - return [cb = std::move(cb), &node_context]( + return [cb = std::move(cb), &node_context, trace_kind]( endpoints::EndpointContext& args, const nlohmann::json& params) { auto config = node_context.get_subsystem(); auto node_operation = node_context.get_subsystem(); @@ -54,6 +57,16 @@ namespace ccf::node auto in = params.get(); recovery_decision_protocol::RequestNodeInfo info = in.info; +#ifdef CCF_RECOVERY_TRACE + if (!in.trace_message_id.has_value()) + { + return make_error( + HTTP_STATUS_BAD_REQUEST, + ccf::errors::InvalidInput, + "Recovery trace message ID is required in trace-enabled builds"); + } +#endif + // ---- Validate the quote against our store and store the node info ---- auto cert_der = ccf::crypto::public_key_der_from_cert( @@ -112,6 +125,22 @@ namespace ccf::node node_info_handle->put(info.location.name, src_info); } +#ifdef CCF_RECOVERY_TRACE + const auto trace_pre = args.tx + .ro( + Tables::RECOVERY_DECISION_PROTOCOL_SM_STATE) + ->get(); + if (!trace_pre.has_value()) + { + return make_error( + HTTP_STATUS_INTERNAL_SERVER_ERROR, + ccf::errors::InternalError, + "Recovery-decision-protocol state not set while tracing"); + } +#else + (void)trace_kind; +#endif + // ---- Run callback ---- auto ret = cb(args, in); @@ -125,7 +154,24 @@ namespace ccf::node try { - node_operation->recovery_decision_protocol().advance(args.tx, false); + auto& protocol = node_operation->recovery_decision_protocol(); + protocol.advance(args.tx, false); +#ifdef CCF_RECOVERY_TRACE + std::optional trace_txid = std::nullopt; + if constexpr (std::is_same_v< + Input, + recovery_decision_protocol::GossipRequest>) + { + trace_txid = in.txid; + } + protocol.record_trace_receive( + args.tx, + trace_kind, + in.trace_message_id, + info.location.name, + trace_txid, + trace_pre.value()); +#endif } catch (const std::logic_error& e) { @@ -186,7 +232,7 @@ namespace ccf::node HTTP_PUT, json_adapter(wrap_recovery_decision_protocol< recovery_decision_protocol::GossipRequest>( - recovery_decision_protocol_gossip, node_context)), + recovery_decision_protocol_gossip, node_context, "gossip_accepted")), no_auth_required) .set_forwarding_required(endpoints::ForwardingRequired::Never) .set_openapi_hidden(true) @@ -212,7 +258,7 @@ namespace ccf::node HTTP_PUT, json_adapter(wrap_recovery_decision_protocol< recovery_decision_protocol::TaggedWithNodeInfo>( - recovery_decision_protocol_vote, node_context)), + recovery_decision_protocol_vote, node_context, "vote_accepted")), no_auth_required) .set_forwarding_required(endpoints::ForwardingRequired::Never) .set_openapi_hidden(true) @@ -284,7 +330,9 @@ namespace ccf::node HTTP_PUT, json_adapter(wrap_recovery_decision_protocol< recovery_decision_protocol::IAmOpenRequest>( - recovery_decision_protocol_iamopen, node_context)), + recovery_decision_protocol_iamopen, + node_context, + "iamopen_accepted")), no_auth_required) .set_forwarding_required(endpoints::ForwardingRequired::Never) .set_openapi_hidden(true) @@ -343,7 +391,14 @@ namespace ccf::node try { - node_operation->recovery_decision_protocol().advance(args.tx, true); + auto& protocol = node_operation->recovery_decision_protocol(); +#ifdef CCF_RECOVERY_TRACE + const auto trace_pre = protocol.get_trace_state(args.tx); +#endif + protocol.advance(args.tx, true); +#ifdef CCF_RECOVERY_TRACE + protocol.record_trace_timeout(args.tx, trace_pre); +#endif } catch (const std::logic_error& e) { From f8528b551de1b5bf4546ac74eb92667d31f60b91 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 4 Sep 2026 12:07:39 +0100 Subject: [PATCH 3/6] Add isolated Lean trace validator Replay strict version 1 recovery traces against the canonical model through a local package dependency, with focused rejection tests, no-sorry checks, documentation, and a shallow workflow. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../lean-disaster-recovery-trace.yml | 55 +++ lean/disaster-recovery-trace/.gitignore | 1 + lean/disaster-recovery-trace/AxiomChecks.lean | 21 + .../DisasterRecoveryTrace.lean | 1 + .../DisasterRecoveryTrace/Protocol/Trace.lean | 2 + .../Protocol/Trace/Format.lean | 143 ++++++ .../Protocol/Trace/Replay.lean | 454 ++++++++++++++++++ lean/disaster-recovery-trace/README.md | 33 ++ .../TRACE_FORMAT_V1.md | 143 ++++++ lean/disaster-recovery-trace/TraceMain.lean | 23 + lean/disaster-recovery-trace/TraceTests.lean | 227 +++++++++ .../lake-manifest.json | 102 ++++ lean/disaster-recovery-trace/lakefile.toml | 28 ++ lean/disaster-recovery-trace/lean-toolchain | 1 + 14 files changed, 1234 insertions(+) create mode 100644 .github/workflows/lean-disaster-recovery-trace.yml create mode 100644 lean/disaster-recovery-trace/.gitignore create mode 100644 lean/disaster-recovery-trace/AxiomChecks.lean create mode 100644 lean/disaster-recovery-trace/DisasterRecoveryTrace.lean create mode 100644 lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace.lean create mode 100644 lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Format.lean create mode 100644 lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Replay.lean create mode 100644 lean/disaster-recovery-trace/README.md create mode 100644 lean/disaster-recovery-trace/TRACE_FORMAT_V1.md create mode 100644 lean/disaster-recovery-trace/TraceMain.lean create mode 100644 lean/disaster-recovery-trace/TraceTests.lean create mode 100644 lean/disaster-recovery-trace/lake-manifest.json create mode 100644 lean/disaster-recovery-trace/lakefile.toml create mode 100644 lean/disaster-recovery-trace/lean-toolchain diff --git a/.github/workflows/lean-disaster-recovery-trace.yml b/.github/workflows/lean-disaster-recovery-trace.yml new file mode 100644 index 00000000000..3c1605c8f2b --- /dev/null +++ b/.github/workflows/lean-disaster-recovery-trace.yml @@ -0,0 +1,55 @@ +name: "Lean Disaster Recovery Trace" + +on: + pull_request: + paths: + - "lean/disaster-recovery-trace/**" + - "lean/disaster-recovery/**" + - "include/ccf/service/tables/self_healing_open.h" + - "src/node/recovery_decision_protocol.cpp" + - "src/node/recovery_decision_protocol.h" + - "src/node/rpc/self_healing_open_handlers.h" + - "tests/e2e_operations.py" + - "tests/infra/recovery_trace.py" + - "CMakeLists.txt" + - ".github/workflows/ci.yml" + - ".github/workflows/lean-disaster-recovery-trace.yml" + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +permissions: read-all + +jobs: + trace-validator: + name: Trace Validator + runs-on: ubuntu-latest + timeout-minutes: 20 + + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Install Lean + shell: bash + run: | + set -euo pipefail + sudo apt-get update + sudo apt-get install -y elan + elan toolchain install "$(cat lean/disaster-recovery-trace/lean-toolchain)" + + - name: Restore Mathlib cache + working-directory: lean/disaster-recovery-trace + shell: bash + run: | + set -euo pipefail + lake exe cache get + + - name: Build and test validator + working-directory: lean/disaster-recovery-trace + shell: bash + run: | + set -euo pipefail + lake build + lake env lean -DwarningAsError=true AxiomChecks.lean + lake exe trace-checks diff --git a/lean/disaster-recovery-trace/.gitignore b/lean/disaster-recovery-trace/.gitignore new file mode 100644 index 00000000000..4080d07dfc3 --- /dev/null +++ b/lean/disaster-recovery-trace/.gitignore @@ -0,0 +1 @@ +/.lake/ diff --git a/lean/disaster-recovery-trace/AxiomChecks.lean b/lean/disaster-recovery-trace/AxiomChecks.lean new file mode 100644 index 00000000000..4ce748b74a9 --- /dev/null +++ b/lean/disaster-recovery-trace/AxiomChecks.lean @@ -0,0 +1,21 @@ +import DisasterRecoveryTrace +import Lean.Elab.Command +import Lean.Util.CollectAxioms + +open Lean Elab Command + +elab "#assert_no_trace_sorries" : command => do + let env <- getEnv + let mut offenders : Array Name := #[] + for (name, _) in env.constants.toList do + if name.toString.startsWith "DisasterRecoveryTrace" then + let axioms <- liftCoreM <| Lean.collectAxioms name + if axioms.contains (Name.mkSimple "sorryAx") then + offenders := offenders.push name + unless offenders.isEmpty do + throwError "trace declarations contain sorryAx: {offenders}" + +#assert_no_trace_sorries + +def main : IO Unit := + pure () diff --git a/lean/disaster-recovery-trace/DisasterRecoveryTrace.lean b/lean/disaster-recovery-trace/DisasterRecoveryTrace.lean new file mode 100644 index 00000000000..0e19cc5345f --- /dev/null +++ b/lean/disaster-recovery-trace/DisasterRecoveryTrace.lean @@ -0,0 +1 @@ +import DisasterRecoveryTrace.Protocol.Trace diff --git a/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace.lean b/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace.lean new file mode 100644 index 00000000000..cf9314260d5 --- /dev/null +++ b/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace.lean @@ -0,0 +1,2 @@ +import DisasterRecoveryTrace.Protocol.Trace.Format +import DisasterRecoveryTrace.Protocol.Trace.Replay diff --git a/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Format.lean b/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Format.lean new file mode 100644 index 00000000000..3eeee9ae895 --- /dev/null +++ b/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Format.lean @@ -0,0 +1,143 @@ +import DisasterRecovery.Protocol.Model +import Lean.Data.Json +import Lean.Data.Json.FromToJson + +namespace DisasterRecoveryTrace.Protocol.Trace + +open DisasterRecovery.Protocol + +open Lean + +def contractVersion : String := + "ccf.recovery_decision_protocol.trace/1" + +inductive Kind where + | start + | gossipAccepted + | voteAccepted + | iAmOpenAccepted + | timeout + | send + | open + | joinRestart + | complete +deriving Repr, BEq, Inhabited + +structure TraceEvent where + version : String + instanceId : String + expectedLocations : List Location + node : Location + sequence : Nat + kind : Kind + messageId : Option String + causedBy : Option String + source : Option Location + txid : Option TxID + pre : Option Phase + post : Option Phase + openKind : Option OpenKind + send : Option String +deriving Repr, BEq, Inhabited + +private def parseKind : String -> Except String Kind + | "start" => pure .start + | "gossip_accepted" => pure .gossipAccepted + | "vote_accepted" => pure .voteAccepted + | "iamopen_accepted" => pure .iAmOpenAccepted + | "timeout" => pure .timeout + | "send" => pure .send + | "open" => pure .open + | "join_restart" => pure .joinRestart + | "complete" => pure .complete + | value => throw s!"unknown kind '{value}'" + +private def parsePhase : String -> Except String Phase + | "GOSSIPING" => pure .gossiping + | "VOTING" => pure .voting + | "OPENING" => pure .opening + | "JOINING" => pure .joining + | "OPEN" => pure .open + | value => throw s!"unknown phase '{value}'" + +private def parseOpenKind : String -> Except String OpenKind + | "QUORUM" => pure .quorum + | "FAILOVER" => pure .failover + | value => throw s!"unknown open kind '{value}'" + +private def optionalString (json : Json) (key : String) : + Except String (Option String) := + match json.getObjVal? key with + | .error _ | .ok .null => pure none + | .ok value => some <$> value.getStr? + +private def optionalNat (json : Json) (key : String) : + Except String (Option Nat) := + match json.getObjVal? key with + | .error _ | .ok .null => pure none + | .ok value => some <$> value.getNat? + +private def optionalParsed + (json : Json) + (key : String) + (parse : String -> Except String α) : + Except String (Option α) := do + match <- optionalString json key with + | none => pure none + | some value => some <$> parse value + +def parseEvent (line : String) : Except String TraceEvent := do + let json <- Json.parse line + let version <- json.getObjValAs? String "version" + if version != contractVersion then + throw s!"unsupported version '{version}'" + + let view <- optionalNat json "view" + let seqno <- optionalNat json "seqno" + if view.isSome != seqno.isSome then + throw "view and seqno must appear together" + + let instanceId <- json.getObjValAs? String "instance" + let expectedLocations <- + json.getObjValAs? (List String) "expected_locations" + let node <- json.getObjValAs? String "node" + let sequence <- json.getObjValAs? Nat "sequence" + let kindName <- json.getObjValAs? String "kind" + let kind <- parseKind kindName + let messageId <- optionalString json "message_id" + let causedBy <- optionalString json "caused_by" + let source <- optionalString json "source" + let pre <- optionalParsed json "pre" parsePhase + let post <- optionalParsed json "post" parsePhase + let openKind <- optionalParsed json "open_kind" parseOpenKind + let send <- optionalString json "send" + pure { + version + instanceId + expectedLocations + node + sequence + kind + messageId + causedBy + source + txid := match view, seqno with + | some view, some seqno => some { view, seqno } + | _, _ => none + pre + post + openKind + send + } + +def parseNDJSON (input : String) : Except String (List TraceEvent) := do + let lines := (input.splitOn "\n").filter + (fun line => !line.trimAscii.isEmpty) + let mut events := [] + for (line, index) in lines.zipIdx do + match parseEvent line with + | .ok event => events := event :: events + | .error message => throw s!"line {index + 1}: {message}" + pure events.reverse + +end DisasterRecoveryTrace.Protocol.Trace diff --git a/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Replay.lean b/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Replay.lean new file mode 100644 index 00000000000..58fdf106a7c --- /dev/null +++ b/lean/disaster-recovery-trace/DisasterRecoveryTrace/Protocol/Trace/Replay.lean @@ -0,0 +1,454 @@ +import DisasterRecoveryTrace.Protocol.Trace.Format + +namespace DisasterRecoveryTrace.Protocol.Trace + +open DisasterRecovery.Protocol + +structure Failure where + prefixLength : Nat + message : String + expected : List String +deriving Repr, BEq + +structure ObservedSend where + messageId : String + source : Location + description : String + txid : Option TxID +deriving Repr, BEq + +structure PendingEffect where + node : Location + effect : Effect +deriving Repr, BEq + +structure PendingSendBatch where + node : Location + remaining : List String +deriving Repr, BEq + +structure ActiveReplay where + config : Config + system : SystemState + startedNodes : List Location + sends : List ObservedSend := [] + consumedSendIds : List String := [] + pendingEffects : List PendingEffect := [] + pendingSendBatches : List PendingSendBatch := [] + terminalNodes : List Location := [] + completedNodes : List Location := [] +deriving Repr, BEq + +structure ReplayState where + active : Option ActiveReplay := none + nextSequence : List (Prod Location Nat) := [] + seenMessageIds : List String := [] +deriving Repr, BEq, Inhabited + +private def nodeState (system : SystemState) (node : Location) : Option NodeState := + (system.nodes.find? fun entry => entry.1 == node).map Prod.snd + +private def phaseMatches (expected : Option Phase) (actual : Phase) : Bool := + expected == some actual + +private def isReceive : Kind -> Bool + | .gossipAccepted | .voteAccepted | .iAmOpenAccepted => true + | _ => false + +private def shapeError (event : TraceEvent) : Option String := + if event.pre.isNone || event.post.isNone then + some "pre and post are required" + else if isReceive event.kind && + (event.messageId.isNone || event.causedBy.isNone || event.source.isNone) then + some "message_id, caused_by, and source are required for receives" + else if event.kind == .gossipAccepted && event.txid.isNone then + some "view and seqno are required for gossip" + else if event.kind == .send && + (event.messageId.isNone || event.send.isNone) then + some "message_id and send are required for sends" + else if event.kind == .send && + (event.send.getD "").startsWith "gossip:" && event.txid.isNone then + some "view and seqno are required for gossip sends" + else if event.kind == .open && event.openKind.isNone then + some "open_kind is required for open" + else if !isReceive event.kind && event.causedBy.isSome then + some "caused_by is only valid on receive events" + else if event.messageId.map String.isEmpty |>.getD false then + some "message_id must not be empty" + else if event.causedBy.map String.isEmpty |>.getD false then + some "caused_by must not be empty" + else if event.source.map String.isEmpty |>.getD false then + some "source must not be empty" + else + none + +private def configError (config : Config) : Option String := + if config.instanceId.isEmpty then + some "instance must not be empty" + else if config.expectedLocations.isEmpty then + some "expected_locations must not be empty" + else if config.expectedLocations.any String.isEmpty then + some "expected_locations must not contain an empty name" + else if config.expectedLocations.eraseDups.length != + config.expectedLocations.length then + some "expected_locations must not contain duplicates" + else + none + +private def expectedSequence (state : ReplayState) (node : Location) : Nat := + (state.nextSequence.find? fun entry => entry.1 == node).map Prod.snd |>.getD 0 + +private def setSequence + (sequences : List (Prod Location Nat)) + (node : Location) + (next : Nat) : + List (Prod Location Nat) := + if sequences.any (fun entry => entry.1 == node) then + sequences.map fun entry => if entry.1 == node then (node, next) else entry + else + (node, next) :: sequences + +private def effectName : Effect -> Option String + | .sendGossip destination => some s!"gossip:{destination}" + | .sendVote destination => some s!"vote:{destination}" + | .sendIAmOpen destination => some s!"iamopen:{destination}" + | _ => none + +private def sendBatch (config : Config) (state : NodeState) : List String := + (step config state .retry).effects.filterMap effectName + +private def setPendingSendBatch + (node : Location) + (remaining : List String) + (batches : List PendingSendBatch) : + List PendingSendBatch := + let others := batches.filter (fun batch => batch.node != node) + if remaining.isEmpty then + others + else + { node, remaining } :: others + +private def receiveDescription (event : TraceEvent) : Option String := + match event.kind with + | .gossipAccepted => some s!"gossip:{event.node}" + | .voteAccepted => some s!"vote:{event.node}" + | .iAmOpenAccepted => some s!"iamopen:{event.node}" + | _ => none + +private def eventInput (event : TraceEvent) : Option Event := + match event.kind, event.source, event.txid with + | .gossipAccepted, some source, some txid => + some (.receiveGossip source txid .accepted) + | .voteAccepted, some source, _ => + some (.receiveVote source .accepted) + | .iAmOpenAccepted, some source, _ => + some (.receiveIAmOpen source .accepted) + | .timeout, _, _ => some .timeout + | _, _, _ => none + +private def isOneShotEffect : Effect -> Bool + | .opening _ | .restart _ | .completed => true + | _ => false + +private def addEffects + (node : Location) + (effects : List Effect) + (pending : List PendingEffect) : + List PendingEffect := + pending ++ (effects.filter isOneShotEffect).map fun effect => + ({ node := node, effect := effect } : PendingEffect) + +private def removeEffect (node : Location) (target : Effect) : + List PendingEffect -> Option (List PendingEffect) + | [] => none + | pending :: rest => + if pending.node == node && pending.effect == target then + some rest + else + (removeEffect node target rest).map (fun remaining => + pending :: remaining) + +private def removeRestart (node : Location) : + List PendingEffect -> Option (List PendingEffect) + | [] => none + | pending :: rest => + if pending.node == node then + match pending.effect with + | .restart _ => some rest + | _ => (removeRestart node rest).map (fun remaining => + pending :: remaining) + else + (removeRestart node rest).map (fun remaining => pending :: remaining) + +private def consumeCause + (active : ActiveReplay) + (event : TraceEvent) : Except String ActiveReplay := do + let cause := event.causedBy.getD "" + if active.consumedSendIds.contains cause then + throw s!"caused_by '{cause}' was already consumed" + let send <- match active.sends.find? (fun send => send.messageId == cause) with + | none => throw s!"caused_by '{cause}' has no prior send" + | some send => pure send + let source := event.source.getD "" + let description := receiveDescription event |>.getD "" + if send.source != source || send.description != description then + throw s!"caused_by '{cause}' has the wrong source, class, or destination" + if event.kind == .gossipAccepted && send.txid != event.txid then + throw s!"caused_by '{cause}' has the wrong gossip TxID" + pure { + active with + consumedSendIds := cause :: active.consumedSendIds + } + +private def applyTransition + (active : ActiveReplay) + (event : TraceEvent) : Except String ActiveReplay := do + let before <- match nodeState active.system event.node with + | none => throw s!"unknown node {event.node}" + | some state => pure state + if !phaseMatches event.pre before.phase then + throw s!"pre phase does not match {phaseName before.phase}" + let input <- match eventInput event with + | none => throw "event is not a protocol transition" + | some input => pure input + let (system, output) <- match + systemStep active.config active.system event.node input with + | none => throw s!"unknown node {event.node}" + | some result => pure result + if !output.accepted then + throw "protocol transition was rejected" + if !phaseMatches event.post output.state.phase then + throw s!"post phase does not match {phaseName output.state.phase}" + pure { + active with + system + pendingEffects := + addEffects event.node output.effects active.pendingEffects + } + +private def applyReceive + (active : ActiveReplay) + (event : TraceEvent) : Except String ActiveReplay := do + applyTransition (← consumeCause active event) event + +private def applySend + (active : ActiveReplay) + (event : TraceEvent) : Except String ActiveReplay := do + let state <- match nodeState active.system event.node with + | none => throw s!"unknown node {event.node}" + | some state => pure state + if !phaseMatches event.pre state.phase || !phaseMatches event.post state.phase then + throw s!"send phase does not match {phaseName state.phase}" + let description := event.send.getD "" + let batch := (active.pendingSendBatches.find? + (fun batch => batch.node == event.node)).map (fun batch => batch.remaining) + |>.getD (sendBatch active.config state) + let expected <- match batch with + | [] => throw "no retry send batch is enabled" + | expected :: _ => pure expected + if description != expected then + throw s!"expected send '{expected}', got '{description}'" + pure { + active with + sends := { + messageId := event.messageId.getD "" + source := event.node + description + txid := event.txid + } :: active.sends + pendingSendBatches := + setPendingSendBatch event.node batch.tail active.pendingSendBatches + } + +private def applyObservation + (active : ActiveReplay) + (event : TraceEvent) : Except String ActiveReplay := do + let state <- match nodeState active.system event.node with + | none => throw s!"unknown node {event.node}" + | some state => pure state + if !phaseMatches event.pre state.phase || !phaseMatches event.post state.phase then + throw s!"observation phase does not match {phaseName state.phase}" + match event.kind with + | .open => + if state.phase != .opening || event.openKind != state.openKind then + throw "open observation does not match state" + let pendingEffects <- match + removeEffect event.node (.opening event.openKind.get!) active.pendingEffects with + | none => throw "open observation has no pending opening effect" + | some pending => pure pending + pure { active with pendingEffects } + | .joinRestart => + if state.phase != .joining || !state.restartRequested then + throw "join_restart observation does not match state" + let pendingEffects <- match removeRestart event.node active.pendingEffects with + | none => throw "join_restart has no pending restart effect" + | some pending => pure pending + pure { + active with + pendingEffects + terminalNodes := event.node :: active.terminalNodes + } + | .complete => + if state.phase != .open then + throw "complete observation does not match state" + let pendingEffects <- match + removeEffect event.node .completed active.pendingEffects with + | none => throw "complete has no pending completion effect" + | some pending => pure pending + pure { + active with + pendingEffects + terminalNodes := event.node :: active.terminalNodes + completedNodes := event.node :: active.completedNodes + } + | _ => throw "event is not a protocol observation" + +private def expectedEvents (active : ActiveReplay) (node : Location) : + List String := + let phase := nodeState active.system node |>.map + (fun state => phaseName state.phase) |>.getD "UNKNOWN" + [s!"state={phase}", "send", "gossip_accepted", "vote_accepted", + "iamopen_accepted", "timeout", "open", "join_restart", "complete"] + +private def start + (active : Option ActiveReplay) + (config : Config) + (event : TraceEvent) : Except String ActiveReplay := do + if !config.expectedLocations.contains event.node then + throw s!"start node {event.node} is not expected" + let current := active.getD { + config + system := initialSystem config + startedNodes := [] + } + if current.config != config then + throw "instance or expected_locations changed" + if current.startedNodes.contains event.node then + throw s!"duplicate start event for node {event.node}" + let state <- match nodeState current.system event.node with + | none => throw s!"unknown node {event.node}" + | some state => pure state + if !phaseMatches event.pre state.phase || !phaseMatches event.post state.phase then + throw "start pre/post phase does not match GOSSIPING" + pure { + current with + startedNodes := event.node :: current.startedNodes + } + +private def processActive + (active : ActiveReplay) + (event : TraceEvent) : Except String ActiveReplay := do + if active.config.instanceId != event.instanceId || + active.config.expectedLocations != event.expectedLocations then + throw "instance or expected_locations changed" + if !active.startedNodes.contains event.node then + throw s!"node {event.node} has no start event" + if event.kind != .send && + active.pendingSendBatches.any (fun batch => batch.node == event.node) then + throw s!"node {event.node} has an incomplete retry send batch" + match event.kind with + | .gossipAccepted | .voteAccepted | .iAmOpenAccepted => + applyReceive active event + | .timeout => applyTransition active event + | .send => applySend active event + | .open | .joinRestart | .complete => applyObservation active event + | .start => throw "unexpected start event" + +private def fail + (index : Nat) + (message : String) + (expected : List String := []) : + Except Failure α := + throw { prefixLength := index + 1, message, expected } + +private def process + (index : Nat) + (state : ReplayState) + (event : TraceEvent) : Except Failure ReplayState := do + if let some message := shapeError event then + fail index message + let config : Config := { + instanceId := event.instanceId + expectedLocations := event.expectedLocations + } + if let some message := configError config then + fail index message + let expectedSeq := expectedSequence state event.node + if event.sequence != expectedSeq then + fail index s!"node {event.node} sequence {event.sequence}, expected {expectedSeq}" + if let some messageId := event.messageId then + if state.seenMessageIds.contains messageId then + fail index s!"message_id '{messageId}' was already used" + if event.causedBy == some messageId then + fail index "message_id and caused_by must identify distinct observations" + + let nextActive <- match event.kind with + | .start => + match start state.active config event with + | .ok active => pure active + | .error message => fail index message + | _ => + match state.active with + | none => fail index "trace must begin with start" ["start"] + | some active => + match processActive active event with + | .ok next => pure next + | .error message => fail index message (expectedEvents active event.node) + + pure { + active := some nextActive + nextSequence := + setSequence state.nextSequence event.node (expectedSeq + 1) + seenMessageIds := event.messageId.toList ++ state.seenMessageIds + } + +def validate (events : List TraceEvent) : Except Failure Unit := do + if events.isEmpty then + throw { + prefixLength := 0 + message := "empty trace" + expected := ["start"] + } + let mut state : ReplayState := {} + for (event, index) in events.zipIdx do + state <- process index state event + let active <- match state.active with + | none => + throw { + prefixLength := events.length + message := "trace has no start event" + expected := ["start"] + } + | some active => pure active + if !active.pendingEffects.isEmpty then + throw { + prefixLength := events.length + message := "trace ended with unobserved committed effects" + expected := ["open", "join_restart", "complete"] + } + if !active.pendingSendBatches.isEmpty then + throw { + prefixLength := events.length + message := "trace ended with incomplete retry send batches" + expected := ["send"] + } + if !active.startedNodes.all (fun node => active.terminalNodes.contains node) then + throw { + prefixLength := events.length + message := "trace ended before every participating node terminated" + expected := ["join_restart", "complete"] + } + if active.completedNodes.isEmpty then + throw { + prefixLength := events.length + message := "trace has no completed opener" + expected := ["complete"] + } + +def renderFailure (failure : Failure) : String := + let expected := + if failure.expected.isEmpty then "" + else s!"\nexpected compatible events:\n {String.intercalate "\n " failure.expected}" + s!"shortest failing prefix: {failure.prefixLength}\n{failure.message}{expected}" + +end DisasterRecoveryTrace.Protocol.Trace diff --git a/lean/disaster-recovery-trace/README.md b/lean/disaster-recovery-trace/README.md new file mode 100644 index 00000000000..6f8875dd4c3 --- /dev/null +++ b/lean/disaster-recovery-trace/README.md @@ -0,0 +1,33 @@ +# Disaster recovery trace validation + +This package validates version 1 implementation traces from CCF's C++ recovery +decision protocol against the permanent model in `../disaster-recovery`. It is +deliberately separate from the canonical model and depends only on +`DisasterRecovery.Protocol.Model`. + +`DisasterRecoveryTrace.Protocol.Trace.Format` parses the strict versioned +NDJSON contract. `DisasterRecoveryTrace.Protocol.Trace.Replay` replays each +event against the canonical transition system. The validator rejects the first +incompatible event and reports the shortest failing prefix. + +## Build and test + +```sh +lake exe cache get +lake build +lake exe trace-checks +lake exe axiom-checks +``` + +Run the validator with: + +```sh +lake exe trace-validator -- TRACE.recovery.ndjson +``` + +`TraceTests.lean` contains small in-memory parser and replay tests. These tests +exercise rejection behavior; they are not implementation conformance evidence. +Conformance evidence is produced only from real C++ SNP recovery runs and is +uploaded by the Milan and Genoa jobs. + +See [TRACE_FORMAT_V1.md](TRACE_FORMAT_V1.md) for the complete contract. diff --git a/lean/disaster-recovery-trace/TRACE_FORMAT_V1.md b/lean/disaster-recovery-trace/TRACE_FORMAT_V1.md new file mode 100644 index 00000000000..0eb8ef10e3c --- /dev/null +++ b/lean/disaster-recovery-trace/TRACE_FORMAT_V1.md @@ -0,0 +1,143 @@ +# Recovery decision protocol trace format, version 1 + +The media type is newline-delimited JSON. Each nonempty line is one committed +semantic observation. The version string is: + +```text +ccf.recovery_decision_protocol.trace/1 +``` + +## Record + +Every record is a JSON object with these required fields: + +| Field | Type | Meaning | +| -------------------- | ---------------- | ----------------------------------- | +| `version` | string | Exactly the version above | +| `instance` | string | Stable recovery instance identifier | +| `expected_locations` | array of strings | Stable configured location names | +| `node` | string | Observed node/location name | +| `sequence` | natural number | Per-node sequence, starting at zero | +| `kind` | string | Event kind from the table below | + +These fields are optional unless the event requires them: + +| Field | Type | Meaning | +| ------------ | ---------------- | ------------------------------------------------------------------------- | +| `message_id` | string | Globally unique ID for an observed send or receive | +| `caused_by` | string | `message_id` of the send that caused a receive | +| `source` | string | Sender location name | +| `view` | natural number | Gossip TxID view | +| `seqno` | natural number | Gossip TxID sequence number | +| `pre` | phase string | Observable phase before the event | +| `post` | phase string | Observable phase after the event | +| `open_kind` | open-kind string | `QUORUM` or `FAILOVER` for an `open` observation | +| `send` | string | Send class and destination: `gossip:NAME`, `vote:NAME`, or `iamopen:NAME` | + +Phase strings are `GOSSIPING`, `VOTING`, `OPENING`, `JOINING`, and `OPEN`. +Unknown fields are ignored for forward-compatible instrumentation metadata. +All integers must be nonnegative Lean `Nat` values. + +## Event kinds + +| Kind | Required event fields | Canonical boundary | +| ------------------ | ----------------------------------------------------------------------------------------------------- | ------------------------------------------- | +| `start` | `pre`, `post` | Protocol state initialized | +| `gossip_accepted` | `message_id`, `caused_by`, `source`, `view`, `seqno`, `pre`, `post` | Validated gossip callback committed | +| `vote_accepted` | `message_id`, `caused_by`, `source`, `pre`, `post` | Validated vote callback committed | +| `iamopen_accepted` | `message_id`, `caused_by`, `source`, `pre`, `post` | IAmOpen selected peer and Joining committed | +| `timeout` | `pre`, `post` | Timeout transaction committed | +| `send` | `send` in `class:destination` form, `message_id`, `pre`, `post`; gossip also requires `view`, `seqno` | Transport send observed | +| `open` | `open_kind`, `pre`, `post` | Service-open transition committed | +| `join_restart` | `pre`, `post` | Joining/restart side effect committed | +| `complete` | `pre`, `post` | Opening-to-Open completion committed | + +Every receive uses `caused_by` to identify an earlier `send`. The validator +checks the sender, destination, message class, gossip TxID payload, and single +consumption of that send. Message IDs, causal IDs, and source names must be +nonempty, and message IDs cannot be reused. Non-receive events must omit +`caused_by`. + +Each participating configured node has one `start` event at sequence zero. The +first creates the replay system; later starts activate other configured nodes +without resetting it. Non-start events for a node +before its start are rejected. Configured but unavailable nodes may have no +start event. Subsequent records must preserve `instance` and +`expected_locations`, refer to a configured node, and increment that node's +sequence exactly. Empty instance IDs, empty configurations, empty location +names, and duplicate configured names are rejected. + +The NDJSON record order is a topological linearization of the distributed +trace. Per-node `sequence` and `caused_by` edges define the ordering; wall-clock +timestamps do not. + +## Strict replay + +Version 1 is a complete successful-execution trace: every transport send, +accepted receive, committed timeout, and one-shot effect is explicit. +`DisasterRecoveryTrace/Protocol/Trace/Replay.lean` folds these events over one deterministic `SystemState`. +It retains only observed sends, consumed causal IDs, per-node sequences, and +pending ordered retry-send batches and `open`, `join_restart`, or `complete` +effects. + +The validator rejects the first event that is not enabled by the canonical +model or whose recorded pre/post state, cause, or effect does not match. It +reports this shortest failing prefix with the current phase and expected event +classes. + +Rejected HTTP/validation inputs do not mutate the modeled state and are not +part of version 1. A future need to validate rejection behavior or incomplete +traces should use a new contract version rather than adding implicit behavior +to this deterministic replay. + +## C++ instrumentation + +Configure CCF with `-DCCF_RECOVERY_TRACE=ON` to enable implementation tracing. +Accepted receive and timeout events are written to +`public:ccf.internal.recovery_decision_protocol.trace_events` in the same +transaction as the modeled state change. A global commit hook emits them only +after commit, followed by any `open`, `join_restart`, or `complete` effect from +that transition. Aborted transactions therefore emit nothing. +In trace-enabled builds the joiner restart request is issued by the trace hook +after the committed receive and `join_restart` records are emitted. Default +builds issue the restart from the committed state hook. Both modes therefore +wait for global commit before requesting restart. + +The committed start hook emits `start` before scheduling retry and failover +tasks. Transport sends are emitted immediately before dispatch and propagate +their generated `message_id` in the internal request as `trace_message_id`; +the committed receive records it as `caused_by`. +If a retry observes a locally committed phase that is not yet globally visible +to the trace hook, tracing defers that retry invocation. Once phases match, the +trace lock serializes the complete send batch against later commit publication. + +Each log record contains `RDP_TRACE ` followed by the event object. +`../../tests/infra/recovery_trace.py` extracts records from all participating node +logs, topologically orders them by per-node sequence and causal send edges, +writes NDJSON, and invokes the Lean validator. The quorum, failover, and +multiple-timeout SNP e2e scenarios call this helper. +Each generated `*.recovery.ndjson` file is retained with the SNP job's uploaded +logs, so a failed replay can be reproduced locally. + +The e2e helper additionally requires scenario-specific terminal evidence before +accepting the trace: the expected open kind, at least one completed opener, and +a `complete` or `join_restart` event for every participating node. + +## Example + +```json +{ + "version": "ccf.recovery_decision_protocol.trace/1", + "instance": "example", + "expected_locations": ["node0"], + "node": "node0", + "sequence": 0, + "kind": "start", + "pre": "GOSSIPING", + "post": "GOSSIPING" +} +``` + +No recovery-decision-protocol traces are checked into the repository. Every +NDJSON trace passed to the validator in CI is captured from the running C++ +implementation. diff --git a/lean/disaster-recovery-trace/TraceMain.lean b/lean/disaster-recovery-trace/TraceMain.lean new file mode 100644 index 00000000000..1d792a75134 --- /dev/null +++ b/lean/disaster-recovery-trace/TraceMain.lean @@ -0,0 +1,23 @@ +import DisasterRecoveryTrace.Protocol.Trace + +open DisasterRecoveryTrace.Protocol.Trace + +def main (args : List String) : IO UInt32 := do + match args with + | [path] => + let input <- IO.FS.readFile path + match parseNDJSON input with + | .error message => + IO.eprintln message + pure 1 + | .ok events => + match validate events with + | .error failure => + IO.eprintln (renderFailure failure) + pure 1 + | .ok () => + IO.println s!"trace accepted: {events.length} events" + pure 0 + | _ => + IO.eprintln "usage: trace-validator TRACE.ndjson" + pure 2 diff --git a/lean/disaster-recovery-trace/TraceTests.lean b/lean/disaster-recovery-trace/TraceTests.lean new file mode 100644 index 00000000000..25ac131896c --- /dev/null +++ b/lean/disaster-recovery-trace/TraceTests.lean @@ -0,0 +1,227 @@ +import DisasterRecoveryTrace.Protocol.Trace + +open DisasterRecovery.Protocol +open DisasterRecoveryTrace.Protocol.Trace + +private def expect (condition : Bool) (message : String) : IO Unit := + unless condition do throw (IO.userError message) + +private def baseEvent + (locations : List Location) + (node : Location) + (sequence : Nat) + (kind : Kind) : TraceEvent := { + version := contractVersion + instanceId := "trace-tests" + expectedLocations := locations + node + sequence + kind + messageId := none + causedBy := none + source := none + txid := none + pre := none + post := none + openKind := none + send := none +} + +private def startEvent + (locations : List Location) + (node : Location) : TraceEvent := { + baseEvent locations node 0 .start with + pre := some .gossiping + post := some .gossiping +} + +private def sendEvent + (locations : List Location) + (sequence : Nat) + (messageId description : String) + (phase : Phase) : TraceEvent := { + baseEvent locations "A" sequence .send with + messageId := some messageId + pre := some phase + post := some phase + txid := if description.startsWith "gossip:" then + some { view := 1, seqno := 1 } + else + none + send := some description +} + +private def gossipEvent + (locations : List Location) + (sequence : Nat) + (messageId cause : String) + (post : Phase) : TraceEvent := { + baseEvent locations "A" sequence .gossipAccepted with + messageId := some messageId + causedBy := some cause + source := some "A" + txid := some { view := 1, seqno := 1 } + pre := some .gossiping + post := some post +} + +private def voteEvent + (locations : List Location) + (sequence : Nat) + (messageId cause : String) + (post : Phase) : TraceEvent := { + baseEvent locations "A" sequence .voteAccepted with + messageId := some messageId + causedBy := some cause + source := some "A" + pre := some .voting + post := some post +} + +private def timeoutEvent + (locations : List Location) + (sequence : Nat) + (pre post : Phase) : TraceEvent := { + baseEvent locations "A" sequence .timeout with + pre := some pre + post := some post +} + +private def openEvent + (locations : List Location) + (sequence : Nat) + (kind : OpenKind) : TraceEvent := { + baseEvent locations "A" sequence .open with + pre := some .opening + post := some .opening + openKind := some kind +} + +private def completeEvent + (locations : List Location) + (sequence : Nat) : TraceEvent := { + baseEvent locations "A" sequence .complete with + pre := some .open + post := some .open +} + +private def validationSucceeds (events : List TraceEvent) : Bool := + match validate events with + | .ok () => true + | .error _ => false + +private def failedAt (events : List TraceEvent) (expectedPrefix : Nat) : Bool := + match validate events with + | .error failure => failure.prefixLength == expectedPrefix + | .ok () => false + +private def parseFails (value : String) : Bool := + match parseEvent value with + | .error _ => true + | .ok _ => false + +private def quorumTrace : List TraceEvent := + let locations := ["A"] + [ + startEvent locations "A", + sendEvent locations 1 "send-gossip" "gossip:A" .gossiping, + gossipEvent locations 2 "receive-gossip" "send-gossip" .voting, + sendEvent locations 3 "send-vote" "vote:A" .voting, + sendEvent locations 4 "send-voting-gossip" "gossip:A" .voting, + voteEvent locations 5 "receive-vote" "send-vote" .opening, + openEvent locations 6 .quorum, + timeoutEvent locations 7 .opening .opening, + timeoutEvent locations 8 .opening .opening, + timeoutEvent locations 9 .opening .open, + completeEvent locations 10 + ] + +def main : IO UInt32 := do + expect (validationSucceeds quorumTrace) "complete quorum trace was rejected" + + let locations := ["A", "B"] + expect + (failedAt [startEvent locations "A", startEvent locations "B"] 2) + "incomplete multi-node trace was accepted" + expect + (failedAt [startEvent locations "A"] 1) + "incomplete single-node trace was accepted" + expect + (failedAt [startEvent locations "A", startEvent locations "A"] 2) + "duplicate start was accepted" + expect + (failedAt [startEvent ["A", "A"] "A"] 1) + "duplicate expected locations were accepted" + expect + (failedAt [{ startEvent ["A"] "A" with instanceId := "" }] 1) + "empty recovery instance was accepted" + + let single := ["A"] + let start := startEvent single "A" + let gossipSend := sendEvent single 1 "send-gossip" "gossip:A" .gossiping + let missingCause := { + gossipEvent single 2 "receive-gossip" "missing" .voting with + causedBy := none + } + expect + (failedAt [start, gossipSend, missingCause] 3) + "receive without caused_by was accepted" + + let wrongClass := { + voteEvent single 2 "receive-vote" "send-gossip" .voting with + pre := some .gossiping + } + expect + (failedAt [start, gossipSend, wrongClass] 3) + "vote consumed a gossip send" + + let wrongTxid := { + gossipEvent single 2 "receive-gossip" "send-gossip" .voting with + txid := some { view := 9, seqno := 9 } + } + expect + (failedAt [start, gossipSend, wrongTxid] 3) + "gossip received a different TxID than its send" + + let wrongPost := gossipEvent single 2 "receive-gossip" "send-gossip" .open + expect + (failedAt [start, gossipSend, wrongPost] 3) + "invalid gossip post-state was accepted" + + let received := gossipEvent single 2 "receive-gossip" "send-gossip" .voting + let reusedCause := { + voteEvent single 3 "receive-vote" "send-gossip" .voting with + pre := some .voting + } + expect + (failedAt [start, gossipSend, received, reusedCause] 4) + "one send caused multiple receives" + + let badSend := sendEvent single 1 "send-vote" "vote:A" .gossiping + expect + (failedAt [start, badSend] 2) + "Voting send was accepted while Gossiping" + + let abortedTimeout := timeoutEvent single 1 .gossiping .gossiping + expect + (failedAt [start, abortedTimeout] 2) + "aborted empty-gossip timeout was accepted" + + let throughOpen := quorumTrace.take 7 + expect + (failedAt (throughOpen ++ [openEvent single 7 .quorum]) 8) + "one opening transition produced multiple open observations" + expect + (failedAt (quorumTrace.take 6) 6) + "trace with an unobserved opening effect was accepted" + + let rejectedJson := + "{\"version\":\"ccf.recovery_decision_protocol.trace/1\"," + ++ "\"instance\":\"x\",\"expected_locations\":[\"A\"]," + ++ "\"node\":\"A\",\"sequence\":0,\"kind\":\"gossip_rejected\"," + ++ "\"pre\":\"GOSSIPING\",\"post\":\"GOSSIPING\"}" + expect (parseFails rejectedJson) + "unused rejection event remains in the strict v1 format" + + IO.println "all strict trace replay checks passed" + pure 0 diff --git a/lean/disaster-recovery-trace/lake-manifest.json b/lean/disaster-recovery-trace/lake-manifest.json new file mode 100644 index 00000000000..569faa3bb97 --- /dev/null +++ b/lean/disaster-recovery-trace/lake-manifest.json @@ -0,0 +1,102 @@ +{"version": "1.1.0", + "packagesDir": ".lake/packages", + "packages": + [{"type": "path", + "scope": "", + "name": "disaster_recovery", + "manifestFile": "lake-manifest.json", + "inherited": false, + "dir": "../disaster-recovery", + "configFile": "lakefile.toml"}, + {"url": "https://github.com/leanprover-community/mathlib4.git", + "type": "git", + "subDir": null, + "scope": "", + "rev": "8f9d9cff6bd728b17a24e163c9402775d9e6a365", + "name": "mathlib", + "manifestFile": "lake-manifest.json", + "inputRev": "v4.28.0", + "inherited": true, + "configFile": "lakefile.lean"}, + {"url": "https://github.com/leanprover-community/plausible", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "55c8532eb21ec9f6d565d51d96b8ca50bd1fbef3", + "name": "plausible", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml"}, + {"url": "https://github.com/leanprover-community/LeanSearchClient", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "c5d5b8fe6e5158def25cd28eb94e4141ad97c843", + "name": "LeanSearchClient", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml"}, + {"url": "https://github.com/leanprover-community/import-graph", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "85b59af46828c029a9168f2f9c35119bd0721e6e", + "name": "importGraph", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml"}, + {"url": "https://github.com/leanprover-community/ProofWidgets4", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "be3b2e63b1bbf496c478cef98b86972a37c1417d", + "name": "proofwidgets", + "manifestFile": "lake-manifest.json", + "inputRev": "v0.0.87", + "inherited": true, + "configFile": "lakefile.lean"}, + {"url": "https://github.com/leanprover-community/aesop", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "f642a64c76df8ba9cb53dba3b919425a0c2aeaf1", + "name": "aesop", + "manifestFile": "lake-manifest.json", + "inputRev": "master", + "inherited": true, + "configFile": "lakefile.toml"}, + {"url": "https://github.com/leanprover-community/quote4", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "b8f98e9087e02c8553945a2c5abf07cec8e798c3", + "name": "Qq", + "manifestFile": "lake-manifest.json", + "inputRev": "master", + "inherited": true, + "configFile": "lakefile.toml"}, + {"url": "https://github.com/leanprover-community/batteries", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "495c008c3e3f4fb4256ff5582ddb3abf3198026f", + "name": "batteries", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml"}, + {"url": "https://github.com/leanprover/lean4-cli", + "type": "git", + "subDir": null, + "scope": "leanprover", + "rev": "4f10f47646cb7d5748d6f423f4a07f98f7bbcc9e", + "name": "Cli", + "manifestFile": "lake-manifest.json", + "inputRev": "v4.28.0", + "inherited": true, + "configFile": "lakefile.toml"}], + "name": "disaster_recovery_trace", + "lakeDir": ".lake"} diff --git a/lean/disaster-recovery-trace/lakefile.toml b/lean/disaster-recovery-trace/lakefile.toml new file mode 100644 index 00000000000..09d013b891a --- /dev/null +++ b/lean/disaster-recovery-trace/lakefile.toml @@ -0,0 +1,28 @@ +name = "disaster_recovery_trace" +version = "0.1.0" +moreLeanArgs = ["-DwarningAsError=true"] +defaultTargets = [ + "DisasterRecoveryTrace", + "trace-checks", + "trace-validator", + "axiom-checks", +] + +[[require]] +name = "disaster_recovery" +path = "../disaster-recovery" + +[[lean_lib]] +name = "DisasterRecoveryTrace" + +[[lean_exe]] +name = "trace-checks" +root = "TraceTests" + +[[lean_exe]] +name = "trace-validator" +root = "TraceMain" + +[[lean_exe]] +name = "axiom-checks" +root = "AxiomChecks" diff --git a/lean/disaster-recovery-trace/lean-toolchain b/lean/disaster-recovery-trace/lean-toolchain new file mode 100644 index 00000000000..4c685fa085f --- /dev/null +++ b/lean/disaster-recovery-trace/lean-toolchain @@ -0,0 +1 @@ +leanprover/lean4:v4.28.0 From 0c87e1bf361b8479120b10112954599d9d89c9ad Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 4 Sep 2026 12:08:30 +0100 Subject: [PATCH 4/6] Add deterministic recovery trace extraction Validate trace identity, per-node sequences, message IDs, and causal edges before producing a deterministic NDJSON linearization for the isolated Lean validator. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- tests/infra/recovery_trace.py | 268 +++++++++++++++++++++++++++++ tests/infra/recovery_trace_test.py | 210 ++++++++++++++++++++++ 2 files changed, 478 insertions(+) create mode 100644 tests/infra/recovery_trace.py create mode 100644 tests/infra/recovery_trace_test.py diff --git a/tests/infra/recovery_trace.py b/tests/infra/recovery_trace.py new file mode 100644 index 00000000000..521a8a216ba --- /dev/null +++ b/tests/infra/recovery_trace.py @@ -0,0 +1,268 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the Apache 2.0 License. + +import heapq +import itertools +import json +import logging +import os +import pathlib +import subprocess +import time + +TRACE_MARKER = "RDP_TRACE " +TRACE_VALIDATOR_ENV = "CCF_LEAN_TRACE_VALIDATOR" +TRACE_VERSION = "ccf.recovery_decision_protocol.trace/1" +LOG = logging.getLogger(__name__) + + +def _event_from_log_line(line, path, line_number): + message = line + try: + outer = json.loads(line) + if isinstance(outer, dict) and isinstance(outer.get("msg"), str): + message = outer["msg"] + except json.JSONDecodeError: + pass + + marker = message.find(TRACE_MARKER) + if marker < 0: + return None + + payload = message[marker + len(TRACE_MARKER) :].lstrip() + try: + event, _ = json.JSONDecoder().raw_decode(payload) + except json.JSONDecodeError as error: + raise ValueError( + f"{path}:{line_number}: invalid recovery trace JSON: {error}" + ) from error + if not isinstance(event, dict): + raise TypeError(f"{path}:{line_number}: recovery trace is not an object") + return event + + +def extract_events(nodes): + events = [] + for node in nodes: + out_path, _ = node.get_logs() + if out_path is None or not os.path.isfile(out_path): + continue + with open(out_path, encoding="utf-8", errors="replace") as log: + for line_number, line in enumerate(log, 1): + event = _event_from_log_line(line, out_path, line_number) + if event is not None: + events.append(event) + if not events: + raise ValueError("no recovery-decision-protocol trace events found") + return events + + +def linearize(events): + successors = [set() for _ in events] + indegree = [0 for _ in events] + + def add_edge(source, destination): + if destination not in successors[source]: + successors[source].add(destination) + indegree[destination] += 1 + + by_node = {} + message_ids = {} + identity = None + for index, event in enumerate(events): + try: + version = event["version"] + instance = event["instance"] + expected_locations = event["expected_locations"] + node = event["node"] + sequence = event["sequence"] + kind = event["kind"] + except KeyError as error: + raise ValueError( + f"trace event {index} is missing {error.args[0]}" + ) from error + if version != TRACE_VERSION: + raise ValueError(f"trace event {index} has unsupported version {version}") + if not isinstance(instance, str) or not instance: + raise ValueError(f"trace event {index} has an invalid instance") + if ( + not isinstance(expected_locations, list) + or not expected_locations + or any( + not isinstance(location, str) or not location + for location in expected_locations + ) + or len(set(expected_locations)) != len(expected_locations) + ): + raise ValueError( + f"trace event {index} has invalid expected_locations" + ) + event_identity = (instance, tuple(expected_locations)) + if identity is None: + identity = event_identity + elif event_identity != identity: + raise ValueError(f"trace event {index} changes recovery identity") + if ( + not isinstance(node, str) + or not node + or node not in expected_locations + or type(sequence) is not int + or sequence < 0 + or not isinstance(kind, str) + ): + raise TypeError(f"trace event {index} has an invalid node or sequence") + by_node.setdefault(node, []).append((sequence, index)) + + message_id = event.get("message_id") + if message_id is not None: + if not isinstance(message_id, str) or not message_id: + raise ValueError(f"trace event {index} has an invalid message_id") + if message_id in message_ids: + raise ValueError(f"duplicate trace message_id {message_id}") + message_ids[message_id] = (index, kind) + + for node, node_events in by_node.items(): + node_events.sort() + sequences = [sequence for sequence, _ in node_events] + if sequences != list(range(len(node_events))): + raise ValueError( + f"node {node} trace sequence is not contiguous from zero: {sequences}" + ) + for (_, previous), (_, current) in itertools.pairwise(node_events): + add_edge(previous, current) + + for index, event in enumerate(events): + caused_by = event.get("caused_by") + if caused_by is None: + continue + if not isinstance(caused_by, str) or not caused_by: + raise ValueError(f"trace event {index} has an invalid caused_by") + if caused_by not in message_ids: + raise ValueError(f"caused_by {caused_by} has no matching send event") + source, kind = message_ids[caused_by] + if kind != "send": + raise ValueError(f"caused_by {caused_by} does not identify a send event") + add_edge(source, index) + + ready = [] + for index, degree in enumerate(indegree): + if degree == 0: + event = events[index] + heapq.heappush( + ready, (event["node"], event["sequence"], event["kind"], index) + ) + + ordered = [] + while ready: + _, _, _, index = heapq.heappop(ready) + ordered.append(events[index]) + for successor in successors[index]: + indegree[successor] -= 1 + if indegree[successor] == 0: + event = events[successor] + heapq.heappush( + ready, + (event["node"], event["sequence"], event["kind"], successor), + ) + + if len(ordered) != len(events): + raise ValueError("recovery trace contains a causal cycle") + return ordered + + +def _validator_path(): + configured = os.getenv(TRACE_VALIDATOR_ENV) + if configured: + return pathlib.Path(configured) + repository = pathlib.Path(__file__).resolve().parents[2] + return ( + repository + / "lean" + / "disaster-recovery-trace" + / ".lake" + / "build" + / "bin" + / "trace-validator" + ) + + +def _participating_node_count(nodes): + return sum(node.remote is not None for node in nodes) + + +def wait_for_terminal_events(network, expected_open_kind, timeout): + expected_node_count = _participating_node_count(network.nodes) + end_time = time.time() + timeout + events = [] + while time.time() < end_time: + try: + events = extract_events(network.nodes) + except ValueError: + time.sleep(0.1) + continue + + started = {event["node"] for event in events if event["kind"] == "start"} + completed = {event["node"] for event in events if event["kind"] == "complete"} + terminal = completed | { + event["node"] for event in events if event["kind"] == "join_restart" + } + opened = [event for event in events if event["kind"] == "open"] + if ( + len(started) == expected_node_count + and started <= terminal + and completed + and opened + and all(event.get("open_kind") == expected_open_kind for event in opened) + ): + return events + time.sleep(0.1) + + raise TimeoutError( + "timed out waiting for terminal recovery trace events: " + f"expected_node_count={expected_node_count}, " + f"started={sorted(started) if events else []}, " + f"expected_open_kind={expected_open_kind}, events={events}" + ) + + +def validate_recovery_trace(network, label, expected_open_kind=None, timeout=20): + if expected_open_kind is None: + events = extract_events(network.nodes) + else: + events = wait_for_terminal_events(network, expected_open_kind, timeout) + events = linearize(events) + trace_path = pathlib.Path(network.common_dir) / f"{label}.recovery.ndjson" + with open(trace_path, "w", encoding="utf-8") as trace: + for event in events: + trace.write(json.dumps(event, separators=(",", ":"), sort_keys=True)) + trace.write("\n") + + validator = _validator_path() + if not validator.is_file(): + raise FileNotFoundError( + f"Lean trace validator not found at {validator}; set {TRACE_VALIDATOR_ENV}" + ) + result = subprocess.run( + [validator, trace_path], + text=True, + capture_output=True, + check=False, + ) + if result.returncode != 0: + raise AssertionError( + f"Lean recovery trace validation failed for {trace_path}:\n" + f"{result.stdout}{result.stderr}" + ) + LOG.info(result.stdout.strip()) + return trace_path + + +def validate_recovery_trace_if_enabled(network, label, expected_open_kind, timeout=20): + if not os.getenv(TRACE_VALIDATOR_ENV): + return None + return validate_recovery_trace( + network, + label, + expected_open_kind=expected_open_kind, + timeout=timeout, + ) diff --git a/tests/infra/recovery_trace_test.py b/tests/infra/recovery_trace_test.py new file mode 100644 index 00000000000..4d295e2c45f --- /dev/null +++ b/tests/infra/recovery_trace_test.py @@ -0,0 +1,210 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the Apache 2.0 License. + +import json +import pathlib +import tempfile +import unittest +from unittest import mock + +import infra.recovery_trace + +VERSION = "ccf.recovery_decision_protocol.trace/1" +EXPECTED_LOCATIONS = ["A", "B"] + + +def event(node, sequence, kind, **extra): + value = { + "version": VERSION, + "instance": "synthetic", + "expected_locations": EXPECTED_LOCATIONS, + "node": node, + "sequence": sequence, + "kind": kind, + "pre": "GOSSIPING", + "post": "GOSSIPING", + } + value.update(extra) + return value + + +class FakeNode: + def __init__(self, path, name=None): + self.path = path + self.name = name + self.remote = object() if name is not None else None + + def get_logs(self): + return str(self.path), None + + def get_sealing_recovery_location(self): + return {"name": self.name} + + +class FakeNetwork: + def __init__(self, nodes, common_dir): + self.nodes = nodes + self.common_dir = common_dir + + +class RecoveryTraceTest(unittest.TestCase): + def test_extract_linearize_and_validate(self): + with tempfile.TemporaryDirectory() as directory: + root = pathlib.Path(directory) + a_log = root / "a.out" + b_log = root / "b.out" + a_events = [ + event("A", 0, "start"), + event( + "A", + 1, + "send", + message_id="send-a-b", + send="gossip:B", + ), + ] + b_events = [ + event("B", 0, "start"), + event( + "B", + 1, + "gossip_accepted", + message_id="receive-a-b", + caused_by="send-a-b", + source="A", + view=1, + seqno=1, + ), + ] + a_log.write_text( + "".join( + f"[info] RDP_TRACE {json.dumps(trace_event)}\n" + for trace_event in a_events + ), + encoding="utf-8", + ) + b_log.write_text( + "".join( + json.dumps({"msg": f"RDP_TRACE {json.dumps(trace_event)}"}) + "\n" + for trace_event in b_events + ), + encoding="utf-8", + ) + network = FakeNetwork( + [FakeNode(b_log), FakeNode(a_log)], + directory, + ) + + extracted = infra.recovery_trace.extract_events(network.nodes) + ordered = infra.recovery_trace.linearize(extracted) + self.assertEqual( + [(item["node"], item["sequence"]) for item in ordered], + [("A", 0), ("A", 1), ("B", 0), ("B", 1)], + ) + + def test_rejects_non_contiguous_sequence(self): + broken = [ + event("A", 0, "start"), + event("A", 2, "timeout"), + ] + with self.assertRaisesRegex(ValueError, "not contiguous"): + infra.recovery_trace.linearize(broken) + + def test_rejects_unresolved_cause(self): + broken = [ + event("A", 0, "start"), + event( + "A", + 1, + "gossip_accepted", + message_id="receive", + caused_by="missing-send", + source="B", + view=1, + seqno=1, + ), + ] + with self.assertRaisesRegex(ValueError, "no matching send"): + infra.recovery_trace.linearize(broken) + + def test_rejects_identity_change(self): + broken = [ + event("A", 0, "start"), + { + **event("A", 1, "timeout"), + "instance": "different", + }, + ] + with self.assertRaisesRegex(ValueError, "changes recovery identity"): + infra.recovery_trace.linearize(broken) + + def test_rejects_duplicate_message_id(self): + broken = [ + event("A", 0, "start", message_id="duplicate"), + event("A", 1, "send", message_id="duplicate"), + ] + with self.assertRaisesRegex(ValueError, "duplicate trace message_id"): + infra.recovery_trace.linearize(broken) + + def test_rejects_causal_cycle(self): + broken = [ + event("A", 0, "start"), + event( + "A", + 1, + "gossip_accepted", + message_id="receive-b", + caused_by="send-b", + ), + event("A", 2, "send", message_id="send-a"), + event("B", 0, "start"), + event( + "B", + 1, + "gossip_accepted", + message_id="receive-a", + caused_by="send-a", + ), + event("B", 2, "send", message_id="send-b"), + ] + with self.assertRaisesRegex(ValueError, "causal cycle"): + infra.recovery_trace.linearize(broken) + + def test_disabled_validation_preserves_default_tests(self): + with mock.patch.dict( + "os.environ", + {infra.recovery_trace.TRACE_VALIDATOR_ENV: ""}, + clear=False, + ): + self.assertIsNone( + infra.recovery_trace.validate_recovery_trace_if_enabled( + FakeNetwork([], "."), "disabled", "QUORUM" + ) + ) + + def test_waits_for_terminal_scenario_evidence(self): + with tempfile.TemporaryDirectory() as directory: + log_path = pathlib.Path(directory) / "a.out" + events = [ + event("A", 0, "start"), + event("A", 1, "open", open_kind="QUORUM"), + event("A", 2, "complete"), + ] + log_path.write_text( + "".join( + f"RDP_TRACE {json.dumps(trace_event)}\n" for trace_event in events + ), + encoding="utf-8", + ) + network = FakeNetwork( + [FakeNode(log_path, "A")], + directory, + ) + self.assertEqual( + infra.recovery_trace.wait_for_terminal_events(network, "QUORUM", 0.1), + events, + ) + + +if __name__ == "__main__": + unittest.main() From cd362fc7b1a76fb2621291376fae936583edd65c Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 4 Sep 2026 12:09:18 +0100 Subject: [PATCH 5/6] Validate SNP recovery scenarios with Lean Build the isolated validator in Milan and Genoa trace-enabled jobs, validate quorum, failover, and repeated-timeout recoveries, and retain generated NDJSON artifacts on failure. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .github/workflows/README.md | 9 + .github/workflows/ci.yml | 38 ++- .../lake-manifest.json | 227 ++++++++++-------- tests/e2e_operations.py | 10 + tests/infra/recovery_trace.py | 4 +- 5 files changed, 181 insertions(+), 107 deletions(-) diff --git a/.github/workflows/README.md b/.github/workflows/README.md index ef1e552a362..6c0b2574c08 100644 --- a/.github/workflows/README.md +++ b/.github/workflows/README.md @@ -110,6 +110,15 @@ behavior checks on relevant pull requests. File: `lean-disaster-recovery.yml` 3rd party dependencies: None +# Lean Disaster Recovery Trace + +Builds the isolated strict trace validator and runs its parser, replay, and +no-sorry checks. The Milan and Genoa SNP jobs in `ci.yml` validate real +committed C++ recovery traces and upload the generated NDJSON evidence. + +File: `lean-disaster-recovery-trace.yml` +3rd party dependencies: None + # Vendored Dependency Verification Verifies that files under `3rdparty/` match the Git commits or release artifacts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e0ff26ae876..0cb9dd9b314 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -292,13 +292,28 @@ jobs: python3 tests/infra/platform_detection.py snp milan shell: bash + - name: "Build Lean recovery trace validator" + run: | + set -euo pipefail + curl --proto '=https' --tlsv1.2 -sSf \ + https://raw.githubusercontent.com/leanprover/elan/58e8d545e33641f66dbcbd22c4283109e71757be/elan-init.sh \ + -o /tmp/elan-init.sh + sh /tmp/elan-init.sh -y --default-toolchain none + rm /tmp/elan-init.sh + export PATH="${HOME}/.elan/bin:${PATH}" + elan toolchain install "$(cat lean/disaster-recovery-trace/lean-toolchain)" + cd lean/disaster-recovery-trace + lake exe cache get + lake build trace-validator + shell: bash + - name: "Build Debug" run: | set -ex git config --global --add safe.directory /__w/CCF/CCF mkdir build cd build - cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DWORKER_THREADS=1 .. + cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DWORKER_THREADS=1 -DCCF_RECOVERY_TRACE=ON .. ninja shell: bash @@ -314,6 +329,7 @@ jobs: shell: bash env: CCF_TEST_SYNC_AFTER_SETUP: 1 + CCF_LEAN_TRACE_VALIDATOR: ${{ github.workspace }}/lean/disaster-recovery-trace/.lake/build/bin/trace-validator ELECTION_TIMEOUT_MS: 10000 - name: "Capture dmesg" @@ -336,6 +352,7 @@ jobs: build/workspace/*/out build/workspace/*/err build/workspace/*/*.ledger/* + build/workspace/**/*.recovery.ndjson build/workspace/*/stack_trace build/workspace/**/openapi_coverage.json if-no-files-found: ignore @@ -378,13 +395,28 @@ jobs: python3 tests/infra/platform_detection.py snp genoa shell: bash + - name: "Build Lean recovery trace validator" + run: | + set -euo pipefail + curl --proto '=https' --tlsv1.2 -sSf \ + https://raw.githubusercontent.com/leanprover/elan/58e8d545e33641f66dbcbd22c4283109e71757be/elan-init.sh \ + -o /tmp/elan-init.sh + sh /tmp/elan-init.sh -y --default-toolchain none + rm /tmp/elan-init.sh + export PATH="${HOME}/.elan/bin:${PATH}" + elan toolchain install "$(cat lean/disaster-recovery-trace/lean-toolchain)" + cd lean/disaster-recovery-trace + lake exe cache get + lake build trace-validator + shell: bash + - name: "Build Debug" run: | set -ex git config --global --add safe.directory /__w/CCF/CCF mkdir build cd build - cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DWORKER_THREADS=1 .. + cmake -GNinja -DCMAKE_BUILD_TYPE=Debug -DWORKER_THREADS=1 -DCCF_RECOVERY_TRACE=ON .. ninja shell: bash @@ -400,6 +432,7 @@ jobs: shell: bash env: CCF_TEST_SYNC_AFTER_SETUP: 1 + CCF_LEAN_TRACE_VALIDATOR: ${{ github.workspace }}/lean/disaster-recovery-trace/.lake/build/bin/trace-validator ELECTION_TIMEOUT_MS: 10000 - name: "Capture dmesg" @@ -422,6 +455,7 @@ jobs: build/workspace/*/out build/workspace/*/err build/workspace/*/*.ledger/* + build/workspace/**/*.recovery.ndjson build/workspace/*/stack_trace build/workspace/**/openapi_coverage.json if-no-files-found: ignore diff --git a/lean/disaster-recovery-trace/lake-manifest.json b/lean/disaster-recovery-trace/lake-manifest.json index 569faa3bb97..a963e4648c8 100644 --- a/lean/disaster-recovery-trace/lake-manifest.json +++ b/lean/disaster-recovery-trace/lake-manifest.json @@ -1,102 +1,125 @@ -{"version": "1.1.0", - "packagesDir": ".lake/packages", - "packages": - [{"type": "path", - "scope": "", - "name": "disaster_recovery", - "manifestFile": "lake-manifest.json", - "inherited": false, - "dir": "../disaster-recovery", - "configFile": "lakefile.toml"}, - {"url": "https://github.com/leanprover-community/mathlib4.git", - "type": "git", - "subDir": null, - "scope": "", - "rev": "8f9d9cff6bd728b17a24e163c9402775d9e6a365", - "name": "mathlib", - "manifestFile": "lake-manifest.json", - "inputRev": "v4.28.0", - "inherited": true, - "configFile": "lakefile.lean"}, - {"url": "https://github.com/leanprover-community/plausible", - "type": "git", - "subDir": null, - "scope": "leanprover-community", - "rev": "55c8532eb21ec9f6d565d51d96b8ca50bd1fbef3", - "name": "plausible", - "manifestFile": "lake-manifest.json", - "inputRev": "main", - "inherited": true, - "configFile": "lakefile.toml"}, - {"url": "https://github.com/leanprover-community/LeanSearchClient", - "type": "git", - "subDir": null, - "scope": "leanprover-community", - "rev": "c5d5b8fe6e5158def25cd28eb94e4141ad97c843", - "name": "LeanSearchClient", - "manifestFile": "lake-manifest.json", - "inputRev": "main", - "inherited": true, - "configFile": "lakefile.toml"}, - {"url": "https://github.com/leanprover-community/import-graph", - "type": "git", - "subDir": null, - "scope": "leanprover-community", - "rev": "85b59af46828c029a9168f2f9c35119bd0721e6e", - "name": "importGraph", - "manifestFile": "lake-manifest.json", - "inputRev": "main", - "inherited": true, - "configFile": "lakefile.toml"}, - {"url": "https://github.com/leanprover-community/ProofWidgets4", - "type": "git", - "subDir": null, - "scope": "leanprover-community", - "rev": "be3b2e63b1bbf496c478cef98b86972a37c1417d", - "name": "proofwidgets", - "manifestFile": "lake-manifest.json", - "inputRev": "v0.0.87", - "inherited": true, - "configFile": "lakefile.lean"}, - {"url": "https://github.com/leanprover-community/aesop", - "type": "git", - "subDir": null, - "scope": "leanprover-community", - "rev": "f642a64c76df8ba9cb53dba3b919425a0c2aeaf1", - "name": "aesop", - "manifestFile": "lake-manifest.json", - "inputRev": "master", - "inherited": true, - "configFile": "lakefile.toml"}, - {"url": "https://github.com/leanprover-community/quote4", - "type": "git", - "subDir": null, - "scope": "leanprover-community", - "rev": "b8f98e9087e02c8553945a2c5abf07cec8e798c3", - "name": "Qq", - "manifestFile": "lake-manifest.json", - "inputRev": "master", - "inherited": true, - "configFile": "lakefile.toml"}, - {"url": "https://github.com/leanprover-community/batteries", - "type": "git", - "subDir": null, - "scope": "leanprover-community", - "rev": "495c008c3e3f4fb4256ff5582ddb3abf3198026f", - "name": "batteries", - "manifestFile": "lake-manifest.json", - "inputRev": "main", - "inherited": true, - "configFile": "lakefile.toml"}, - {"url": "https://github.com/leanprover/lean4-cli", - "type": "git", - "subDir": null, - "scope": "leanprover", - "rev": "4f10f47646cb7d5748d6f423f4a07f98f7bbcc9e", - "name": "Cli", - "manifestFile": "lake-manifest.json", - "inputRev": "v4.28.0", - "inherited": true, - "configFile": "lakefile.toml"}], - "name": "disaster_recovery_trace", - "lakeDir": ".lake"} +{ + "version": "1.1.0", + "packagesDir": ".lake/packages", + "packages": [ + { + "type": "path", + "scope": "", + "name": "disaster_recovery", + "manifestFile": "lake-manifest.json", + "inherited": false, + "dir": "../disaster-recovery", + "configFile": "lakefile.toml" + }, + { + "url": "https://github.com/leanprover-community/mathlib4.git", + "type": "git", + "subDir": null, + "scope": "", + "rev": "8f9d9cff6bd728b17a24e163c9402775d9e6a365", + "name": "mathlib", + "manifestFile": "lake-manifest.json", + "inputRev": "v4.28.0", + "inherited": true, + "configFile": "lakefile.lean" + }, + { + "url": "https://github.com/leanprover-community/plausible", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "55c8532eb21ec9f6d565d51d96b8ca50bd1fbef3", + "name": "plausible", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml" + }, + { + "url": "https://github.com/leanprover-community/LeanSearchClient", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "c5d5b8fe6e5158def25cd28eb94e4141ad97c843", + "name": "LeanSearchClient", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml" + }, + { + "url": "https://github.com/leanprover-community/import-graph", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "85b59af46828c029a9168f2f9c35119bd0721e6e", + "name": "importGraph", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml" + }, + { + "url": "https://github.com/leanprover-community/ProofWidgets4", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "be3b2e63b1bbf496c478cef98b86972a37c1417d", + "name": "proofwidgets", + "manifestFile": "lake-manifest.json", + "inputRev": "v0.0.87", + "inherited": true, + "configFile": "lakefile.lean" + }, + { + "url": "https://github.com/leanprover-community/aesop", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "f642a64c76df8ba9cb53dba3b919425a0c2aeaf1", + "name": "aesop", + "manifestFile": "lake-manifest.json", + "inputRev": "master", + "inherited": true, + "configFile": "lakefile.toml" + }, + { + "url": "https://github.com/leanprover-community/quote4", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "b8f98e9087e02c8553945a2c5abf07cec8e798c3", + "name": "Qq", + "manifestFile": "lake-manifest.json", + "inputRev": "master", + "inherited": true, + "configFile": "lakefile.toml" + }, + { + "url": "https://github.com/leanprover-community/batteries", + "type": "git", + "subDir": null, + "scope": "leanprover-community", + "rev": "495c008c3e3f4fb4256ff5582ddb3abf3198026f", + "name": "batteries", + "manifestFile": "lake-manifest.json", + "inputRev": "main", + "inherited": true, + "configFile": "lakefile.toml" + }, + { + "url": "https://github.com/leanprover/lean4-cli", + "type": "git", + "subDir": null, + "scope": "leanprover", + "rev": "4f10f47646cb7d5748d6f423f4a07f98f7bbcc9e", + "name": "Cli", + "manifestFile": "lake-manifest.json", + "inputRev": "v4.28.0", + "inherited": true, + "configFile": "lakefile.toml" + } + ], + "name": "disaster_recovery_trace", + "lakeDir": ".lake" +} diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index dec9d8f1653..dbb890ab144 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -38,6 +38,7 @@ import infra.path import infra.platform_detection import infra.proc +import infra.recovery_trace import infra.utils import suite.test_requirements as reqs from ccf.tx_id import TxID @@ -2890,6 +2891,9 @@ def run_recovery_decision_protocol(const_args): assert ( recovery_type == '"Quorum"' ), f"Network self-healing open type was {recovery_type} instead of Quorum" + infra.recovery_trace.validate_recovery_trace_if_enabled( + recovered_network, args.label, "QUORUM" + ) def run_recovery_decision_protocol_timeout_path(const_args): @@ -2942,6 +2946,9 @@ def run_recovery_decision_protocol_timeout_path(const_args): assert ( recovery_type == '"Failover"' ), f"Network self-healing open type was {recovery_type} instead of Failover" + infra.recovery_trace.validate_recovery_trace_if_enabled( + recovered_network, args.label, "FAILOVER" + ) def run_recovery_decision_protocol_multiple_timeout(const_args): @@ -2994,6 +3001,9 @@ def run_recovery_decision_protocol_multiple_timeout(const_args): node.refresh_network_state(verify_ca=False) assert len(recovered_network.get_joined_nodes()) == len(args.nodes) + infra.recovery_trace.validate_recovery_trace_if_enabled( + recovered_network, args.label, "FAILOVER" + ) def run_read_ledger_on_testdata(args): diff --git a/tests/infra/recovery_trace.py b/tests/infra/recovery_trace.py index 521a8a216ba..055c76c2908 100644 --- a/tests/infra/recovery_trace.py +++ b/tests/infra/recovery_trace.py @@ -94,9 +94,7 @@ def add_edge(source, destination): ) or len(set(expected_locations)) != len(expected_locations) ): - raise ValueError( - f"trace event {index} has invalid expected_locations" - ) + raise ValueError(f"trace event {index} has invalid expected_locations") event_identity = (instance, tuple(expected_locations)) if identity is None: identity = event_identity From b49f888203c53b53fcfd6790f1db88b74b456004 Mon Sep 17 00:00:00 2001 From: Amaury Chamayou Date: Fri, 4 Sep 2026 12:27:13 +0100 Subject: [PATCH 6/6] Run trace ordering checks in CI Exercise the focused Python extraction and causal-ordering suite in the dedicated trace workflow. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .github/workflows/lean-disaster-recovery-trace.yml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.github/workflows/lean-disaster-recovery-trace.yml b/.github/workflows/lean-disaster-recovery-trace.yml index 3c1605c8f2b..86285675e9b 100644 --- a/.github/workflows/lean-disaster-recovery-trace.yml +++ b/.github/workflows/lean-disaster-recovery-trace.yml @@ -53,3 +53,10 @@ jobs: lake build lake env lean -DwarningAsError=true AxiomChecks.lean lake exe trace-checks + + - name: Test trace extraction and ordering + working-directory: tests + shell: bash + run: | + set -euo pipefail + python3 -m unittest infra.recovery_trace_test