diff --git a/agent/ACTOR_AUDIT.md b/agent/ACTOR_AUDIT.md index 9a684bd4bc..615af457cf 100644 --- a/agent/ACTOR_AUDIT.md +++ b/agent/ACTOR_AUDIT.md @@ -44,8 +44,8 @@ exercise so production responsibility size remains visible. - Ephemeral effect state: correlation IDs, collector addresses, timer handles, early-arrival buffers, and in-flight submission guards. These are grouped and named rather than mixed into protocol state. -- External authority: EVM contract state. Writer preflights provide cross-restart idempotency where - no durable local outbox exists. +- External authority: EVM contract state. Writer preflights provide cross-restart idempotency, and + startup pairs durable effect intents with their completion events before re-driving open loops. `Persistable::try_mutate` now accepts the snapshot write into the bounded store mailbox before it exposes the new value in memory. This does not turn snapshots into an external-effect outbox; the @@ -59,7 +59,8 @@ construction, and cohesive FHE/math algorithms may exceed 300 lines. Large non-a infrastructure coordinators—notably `CiphernodeBuilder`, `NetInterface`, and the multithread task pool—need their own behavior-preserving projects if their responsibilities are changed. -The remaining architectural gap is durable effect intent. Transaction submission and some -cryptographic work still rely on replay plus external preflight instead of a versioned local -intent/result outbox. That is a schema and recovery change, not a safe file-movement refactor, and -must be implemented with migration and crash-matrix tests. +The append-only event log now supplies durable effect intent: startup scans it in bounded pages, +pairs supported intents with completion/terminal events, and emits internal `EffectRetry` +envelopes only after `EffectsEnabled`. This closes the snapshot-advanced/open-loop loss mode. +There is still no atomic transaction/receipt outbox or full EVM reorg rollback; contract simulation +and canonical backfill remain the authority when a crash lands between submission and observation. diff --git a/agent/CRATES_ARCHITECTURE.md b/agent/CRATES_ARCHITECTURE.md index 828f1490ed..b25f1fdc31 100644 --- a/agent/CRATES_ARCHITECTURE.md +++ b/agent/CRATES_ARCHITECTURE.md @@ -342,6 +342,7 @@ flowchart TD Restart[restart] --> Index[reconcile timestamp index in 1024-record pages] Index --> Schema[schema-version preflight before runtime actor writes] Schema --> SnapshotMeta[load aggregate cursors and initial HLC floor] + SnapshotMeta --> OpenEffects[scan complete logs for unmatched effect intents] SnapshotMeta --> Query[query every post-snapshot aggregate] Query --> Runs[sort 1024-event pages into secure temporary runs] Runs --> GlobalOrder[bounded-fan-in merge by HLC timestamp] @@ -351,7 +352,9 @@ flowchart TD EvmBackfill --> NetBackfill[bounded historical network sync] NetBackfill --> Merge[merge and sort EVM plus network history by HLC] Merge --> Enable[EffectsEnabled] - Enable -->|durable pipeline and fanout fence| PersistHistory[persist and dispatch reconciled history] + OpenEffects --> Retry[bounded set of internal EffectRetry envelopes] + Enable -->|durable pipeline and fanout fence| Retry + Retry -->|durable pipeline and fanout fence| PersistHistory[persist and dispatch reconciled history] PersistHistory -->|durable pipeline and fanout fence| End[SyncEnded] End -->|durable pipeline and fanout fence| Live[live operation] end @@ -377,22 +380,29 @@ newer log records. Replay then waits for concurrent acceptance by all current Ev An unavailable subscriber or a subscriber blocked beyond the bounded acceptance timeout aborts recovery. An `EventBusBarrier` therefore completes only after the last replay fanout has completed. A persisted `Shutdown` event from the previous process is classified as infrastructure and is not -replayed into newly constructed actors. +replayed into newly constructed actors. Startup separately scans every aggregate from sequence one +in the same bounded pages, retaining only supported effect intents that have no matching completion +or terminal lifecycle event. This full-log scan covers intents older than the latest snapshot cursor +while bounding retained work by count and encoded size. The EventBus mailbox remains bounded at `MAILBOX_LIMIT_LARGE` (2,560 messages). The replay producer no longer attempts to enqueue the entire backlog into that mailbox in one burst, and EventBus subscriber fanout no longer bypasses downstream mailbox limits. EventStore query responses also await recipient capacity, preventing a full aggregation mailbox from dropping one aggregate response -and hanging startup. Recovery publishes `EffectsEnabled`, canonical history, and `SyncEnded` as -three separately fenced phases. Runtime log-read failures are returned in the correlated query -response and flow through the existing error paths; a remote sync query therefore cannot panic the -EventStore actor. The fail-stop behavior below applies to durable append/index-write failures. An -event-log or timestamp-index write error panics the affected EventStore before live dispatch. This -preserves durable-before-dispatch safety, but under the default unwind profile an Actix actor panic -is contained at its spawned task boundary: it can kill the store actor and stall the sequencer -without terminating the process. Process-level health supervision would need to detect the stalled -pipeline, but the current runtime does not provide that guarantee. A restart, when it occurs, treats -the event log as authoritative and reconciles missing derived index rows. +and hanging startup. Recovery publishes `EffectsEnabled`, persisted `EffectRetry` envelopes, +canonical history, and `SyncEnded` as separately fenced phases. Retry envelopes contain a closed +set of effect payloads and are consumed only by effect executors, so state-building subscribers do +not observe the original domain event a second time. Persisted retry envelopes are classified as +infrastructure on the next replay; the original intent/completion pair remains the durable recovery +authority. Runtime log-read failures are returned in the correlated query response and flow through +the existing error paths; a remote sync query therefore cannot panic the EventStore actor. The +fail-stop behavior below applies to durable append/index-write failures. An event-log or +timestamp-index write error panics the affected EventStore before live dispatch. This preserves +durable-before-dispatch safety, but under the default unwind profile an Actix actor panic is contained +at its spawned task boundary: it can kill the store actor and stall the sequencer without terminating +the process. Process-level health supervision would need to detect the stalled pipeline, but the +current runtime does not provide that guarantee. A restart, when it occurs, treats the event log as +authoritative and reconciles missing derived index rows. Those replay guarantees bound local replay memory and file-descriptor use, but they do not make the whole persistence path synchronously acknowledged. Live publication and the sequencer/store response @@ -571,11 +581,11 @@ attributable to a canonical party before they become slashing evidence. Replayed semantic replay domain across deferred, in-flight, and completed submissions. Retryable submission failures release their key. Successful or known-benign terminal results retain it. -The gate is deliberately described as in-memory: there is no durable external-effect outbox or -persisted transaction intent. A crash after snapshot advancement but before receipt classification -can therefore lose the local redrive state, and a crash after submission can require on-chain -reconciliation to distinguish landed from missing work. Closing that gap requires a durable -intent/result state machine and contract preflight, not another process-local set. +The submission gate itself is in-memory, but startup reconstructs its missing work from the full +durable event log. A `SlashProposed` observation closes the matching semantic replay key during +replay; otherwise an `EffectRetry` is delivered after effects are enabled. A crash after +submission is reconciled through the contract's `evidenceConsumed` view before retry, while +`DuplicateEvidence` remains the final race-safe idempotency boundary. ## Program-server trust boundary @@ -668,7 +678,8 @@ flowchart LR | C0/share proof-verification context | Finalized-committee and ciphernode-selector repositories plus global verifier memory | Canonical slots and E3 preset/threshold metadata load before ZK actor startup, then lifecycle events maintain or clear them | | HLC, EventBus dedup, and admission state | Event pipeline actors in memory | Maximum snapshot/replay HLC; a fresh bounded dedup window is populated by replay and live events | | Network peer/buffer/interest state | libp2p and network actors in memory | Fresh peer dialing; document interest returns only when selection observations are replayed or redriven | -| Slash-submission replay gate | `SlashingManagerSolWriter` process memory | Rebuilt from replay; not a durable outbox | +| Open effect intents | Per-aggregate append-only event logs | Full bounded intent/completion scan followed by `EffectRetry` after `EffectsEnabled` | +| Slash-submission replay gate | `SlashingManagerSolWriter` process memory | Rebuilt from replay plus the durable open-effect scan | | Pending transaction nonce allocation | Per-chain writer mutex in memory | Provider pending nonce on restart | | In-flight accusation votes and timers | Per-E3 accusation actor memory | No complete durable reconstruction; only events inside the replay window may be observed again | | libp2p identity | Encrypted keypair repository | Decrypt at startup | diff --git a/agent/flow-trace/03_E3_REQUEST_AND_COMMITTEE.md b/agent/flow-trace/03_E3_REQUEST_AND_COMMITTEE.md index 67bb40da82..146c74fa7b 100644 --- a/agent/flow-trace/03_E3_REQUEST_AND_COMMITTEE.md +++ b/agent/flow-trace/03_E3_REQUEST_AND_COMMITTEE.md @@ -290,7 +290,10 @@ CiphernodeSelector receives WithSortitionTicket ``` CiphernodeRegistrySolWriter receives TicketGenerated event │ -└─ Calls contract.submitTicket(e3Id, ticketNumber).send() +├─ Simulates the exact submitTicket(e3Id, ticketNumber) call +│ └─ Treats already-submitted/finalized/deadline/ineligible states as terminal no-ops +│ +└─ If simulation succeeds, calls contract.submitTicket(e3Id, ticketNumber).send() │ │ ┌─── ON-CHAIN (CiphernodeRegistryOwnable) ──────────────┐ │ │ │ @@ -350,6 +353,11 @@ CiphernodeRegistrySolWriter receives TicketGenerated event │ └─────────────────────────────────────────────────────────┘ ``` +On restart, the sync service pairs each durable local `TicketGenerated` intent with +`TicketSubmitted`, `CommitteeFinalized`, or `CommitteeFormationFailed`. An unmatched intent is +wrapped in an internal `EffectRetry` after `EffectsEnabled`; the simulation above is the final +idempotency check when a pre-crash transaction landed but its completion log was not persisted. + --- ## Step 3: Committee Finalization diff --git a/agent/flow-trace/04_DKG_AND_COMPUTATION.md b/agent/flow-trace/04_DKG_AND_COMPUTATION.md index 349dd1b465..3938decbad 100644 --- a/agent/flow-trace/04_DKG_AND_COMPUTATION.md +++ b/agent/flow-trace/04_DKG_AND_COMPUTATION.md @@ -1252,5 +1252,6 @@ present in the running ABI catalog is exposed as `UnknownEvmLog` with raw topics During restart, `ComputeEffectGate` observes replay before compute workers are effects-enabled. It buffers and deduplicates `ComputeRequest`s, prefers the newest regenerated request, cancels terminal -E3 work, and releases pending jobs only after `EffectsEnabled`. The gate changes effect timing, not -durable event order or audit state. +E3 work, and consumes matching `ComputeResponse` / `ComputeRequestError` events so completed jobs +are not released. The full-log open-effect scan wraps only unmatched requests in `EffectRetry` +after `EffectsEnabled`. The gate changes effect timing, not durable event order or audit state. diff --git a/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md b/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md index dfe702e8c7..1f1b653d44 100644 --- a/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md +++ b/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md @@ -133,7 +133,9 @@ No decryption allocation is paid unless the E3 completes normally. Runtime note: `processE3Failure()` is a permissionless cleanup path. The Rust `InterfoldSolWriter` may auto-submit it from any effects-enabled node on the same chain, and it must not depend on active-aggregator designation because failures can happen before committee finalization or while the -current aggregator is offline. +current aggregator is offline. Before a recovered submission it simulates the exact contract call; +`NoPaymentToRefund` proves the refund was already processed and suppresses the transaction, while +other failures remain visible and retryable on later recovery. ```text Anyone calls: Interfold.processE3Failure(e3Id) @@ -515,7 +517,11 @@ AccusationQuorumReached event arrives at SlashingManagerSolWriter ├─ 1. EFFECT AND REPLAY GATE: │ Before EffectsEnabled (startup replay), retain the intent without sending a transaction │ Coalesce by the contract replay tuple (chainId, e3Id, accused, proofType) +│ A replayed SlashProposed observation closes the matching deferred tuple │ After EffectsEnabled, release each retained intent once and track it in flight +│ A full-log scan emits EffectRetry for an unmatched pre-snapshot intent +│ Before sending, read evidenceConsumed(keccak256(chainId,e3Id,accused,proofType)) +│ and skip when a pre-crash or primary-submitter transaction already landed │ ├─ 2. STAGGERED SUBMISSION (fallback submitters): │ Rank all agreeing voters by address (sorted ascending) diff --git a/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md b/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md index ca17d311b7..42b0a605ae 100644 --- a/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md +++ b/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md @@ -261,25 +261,33 @@ On restart: │ → ThresholdPlaintextAggregatorExtension records this role in the E3 context │ so a plaintext buffer created later by CiphertextOutputPublished starts │ with the correct active-aggregator flag -│ 3. Replay EventStore events since last snapshot (effects still disabled) +│ 3. Scan every aggregate from sequence one in bounded pages +│ → Pair local effect intents with their completion/terminal events +│ → Retain only open intents, bounded by count and encoded bytes +│ → Covers an intent that is older than the latest snapshot cursor +│ 4. Replay EventStore events since last snapshot (effects still disabled) │ → Read each aggregate in 1,024-event pages, sort bounded temporary runs, │ and perform a bounded-fan-in global merge by HLC timestamp │ → Each concurrent EventBus fanout is acknowledged before the next event; │ an unavailable or blocked listener aborts recovery after a bounded wait │ → Structured progress is emitted every 10,000 EventBus-handled events -│ 4. Fetch historical EVM events from last known block -│ 5. Historical libp2p sync retries failed aggregate fetches after reconnects +│ → ComputeEffectGate buffers requests and consumes matching results/errors +│ 5. Fetch historical EVM events from last known block +│ 6. Historical libp2p sync retries failed aggregate fetches after reconnects │ and also on bounded retry intervals even without a new connection event -│ 6. Sort & publish merged events by HLC timestamp +│ 7. Sort merged historical events by HLC timestamp │ → A logical event returned by a peer with its source changed from Local to Net is │ idempotent when timestamp, stable event ID, and payload match the stored record; │ a different payload at the same timestamp still fails closed as a collision │ → ComputeEffectGate has already subscribed and buffers ComputeRequest │ effects, deduplicating semantic retries while replay is in progress -│ 7. Enable effects (writers may submit only after this point) +│ 8. Enable effects (writers may submit only after this point) │ → Gate cancels work for terminal E3s and releases only the newest │ pending request for each in-flight semantic compute operation -│ 8. SyncEnded → live operations begin +│ 9. Persist and fan out EffectRetry for each unmatched intent +│ → The closed retry envelope is consumed only by effect executors +│ → Old retry envelopes are skipped on replay; original intent/result pairs stay authoritative +│ 10. Publish merged historical events, then SyncEnded → live operations begin └─ Node resumes from where it left off ``` @@ -349,12 +357,15 @@ flowchart TD Actors --> Replay["sync(): replay EventStore
effects disabled"] EventStore --> Replay + EventStore --> OpenEffects["bounded full-log scan
pair effect intents/results"] Replay --> CommitteeReplay["CommitteePublished replay
restores full committee"] Replay --> Effects["EffectsEnabled"] Effects --> Gate["ComputeEffectGate releases replay-safe compute work"] + OpenEffects --> Retry["unmatched EffectRetry envelopes"] + Effects --> Retry - Effects --> Live["Live/historical chain events"] + Retry --> Live["Live/historical chain events"] Live --> Ciphertext["CiphertextOutputPublished"] Ciphertext --> CanStart{"full + honest committee
and keyshare actor ready?"} CanStart -- yes --> NewPlaintext["Create ThresholdPlaintextAggregator
seed buffer with active aggregator role"] diff --git a/crates/events/src/interfold_event/effect_retry.rs b/crates/events/src/interfold_event/effect_retry.rs new file mode 100644 index 0000000000..d71f92b61c --- /dev/null +++ b/crates/events/src/interfold_event/effect_retry.rs @@ -0,0 +1,165 @@ +// SPDX-License-Identifier: LGPL-3.0-only +// +// This file is provided WITHOUT ANY WARRANTY; +// without even the implied warranty of MERCHANTABILITY +// or FITNESS FOR A PARTICULAR PURPOSE. + +use actix::Message; +use serde::{Deserialize, Serialize}; +use std::fmt::{self, Display}; + +use super::{ + AccusationQuorumReached, CommitteeFinalizeRequested, ComputeRequest, E3StageChanged, + InterfoldEventData, PlaintextAggregated, PublicKeyAggregated, PublishDocumentRequested, + TicketGenerated, +}; + +/// The closed set of side-effect intents that startup recovery may re-drive. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum RecoverableEffect { + ComputeRequest(ComputeRequest), + TicketGenerated(TicketGenerated), + CommitteeFinalizeRequested(CommitteeFinalizeRequested), + PublicKeyAggregated(PublicKeyAggregated), + PlaintextAggregated(PlaintextAggregated), + PublishDocumentRequested(PublishDocumentRequested), + AccusationQuorumReached(AccusationQuorumReached), + E3StageChanged(E3StageChanged), +} + +impl RecoverableEffect { + fn event_type(&self) -> &'static str { + match self { + Self::ComputeRequest(_) => "ComputeRequest", + Self::TicketGenerated(_) => "TicketGenerated", + Self::CommitteeFinalizeRequested(_) => "CommitteeFinalizeRequested", + Self::PublicKeyAggregated(_) => "PublicKeyAggregated", + Self::PlaintextAggregated(_) => "PlaintextAggregated", + Self::PublishDocumentRequested(_) => "PublishDocumentRequested", + Self::AccusationQuorumReached(_) => "AccusationQuorumReached", + Self::E3StageChanged(_) => "E3StageChanged", + } + } + + pub fn e3_id(&self) -> &crate::E3id { + match self { + Self::ComputeRequest(effect) => &effect.e3_id, + Self::TicketGenerated(effect) => &effect.e3_id, + Self::CommitteeFinalizeRequested(effect) => &effect.e3_id, + Self::PublicKeyAggregated(effect) => &effect.e3_id, + Self::PlaintextAggregated(effect) => &effect.e3_id, + Self::PublishDocumentRequested(effect) => &effect.meta.e3_id, + Self::AccusationQuorumReached(effect) => &effect.e3_id, + Self::E3StageChanged(effect) => &effect.e3_id, + } + } + + pub fn is_compute(&self) -> bool { + matches!(self, Self::ComputeRequest(_)) + } + + fn into_event_data(self) -> InterfoldEventData { + match self { + Self::ComputeRequest(effect) => effect.into(), + Self::TicketGenerated(effect) => effect.into(), + Self::CommitteeFinalizeRequested(effect) => effect.into(), + Self::PublicKeyAggregated(effect) => effect.into(), + Self::PlaintextAggregated(effect) => effect.into(), + Self::PublishDocumentRequested(effect) => effect.into(), + Self::AccusationQuorumReached(effect) => effect.into(), + Self::E3StageChanged(effect) => effect.into(), + } + } +} + +impl TryFrom for RecoverableEffect { + type Error = InterfoldEventData; + + fn try_from(effect: InterfoldEventData) -> Result { + match effect { + InterfoldEventData::ComputeRequest(effect) => Ok(Self::ComputeRequest(effect)), + InterfoldEventData::TicketGenerated(effect) => Ok(Self::TicketGenerated(effect)), + InterfoldEventData::CommitteeFinalizeRequested(effect) => { + Ok(Self::CommitteeFinalizeRequested(effect)) + } + InterfoldEventData::PublicKeyAggregated(effect) => { + Ok(Self::PublicKeyAggregated(effect)) + } + InterfoldEventData::PlaintextAggregated(effect) => { + Ok(Self::PlaintextAggregated(effect)) + } + InterfoldEventData::PublishDocumentRequested(effect) => { + Ok(Self::PublishDocumentRequested(effect)) + } + InterfoldEventData::AccusationQuorumReached(effect) => { + Ok(Self::AccusationQuorumReached(effect)) + } + InterfoldEventData::E3StageChanged(effect) => Ok(Self::E3StageChanged(effect)), + unsupported => Err(unsupported), + } + } +} + +/// Internal startup-recovery envelope for a durable effect intent whose +/// corresponding completion event is absent from the event log. +/// +/// Keeping the original payload inside a distinct event type lets effect +/// executors retry after `EffectsEnabled` without replaying the domain event +/// into state-building subscribers a second time. +#[derive(Message, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[rtype(result = "()")] +pub struct EffectRetry { + effect: RecoverableEffect, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct UnsupportedRecoverableEffect { + event_type: String, +} + +impl Display for UnsupportedRecoverableEffect { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{} is not a recoverable effect", self.event_type) + } +} + +impl std::error::Error for UnsupportedRecoverableEffect {} + +impl EffectRetry { + pub fn new(effect: InterfoldEventData) -> Result { + let event_type = effect.event_type(); + let effect = effect + .try_into() + .map_err(|_| UnsupportedRecoverableEffect { event_type })?; + Ok(Self { effect }) + } + + pub fn effect(&self) -> &RecoverableEffect { + &self.effect + } + + pub fn into_effect(self) -> InterfoldEventData { + self.effect.into_event_data() + } +} + +impl Display for EffectRetry { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "EffectRetry({})", self.effect.event_type()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{E3id, TestEvent}; + + #[test] + fn retry_envelope_accepts_only_the_closed_effect_set() { + let supported = CommitteeFinalizeRequested { + e3_id: E3id::new("1", 1), + }; + assert!(EffectRetry::new(supported.into()).is_ok()); + assert!(EffectRetry::new(TestEvent::new("state", 1).into()).is_err()); + } +} diff --git a/crates/events/src/interfold_event/mod.rs b/crates/events/src/interfold_event/mod.rs index c1f6412089..725843dae2 100644 --- a/crates/events/src/interfold_event/mod.rs +++ b/crates/events/src/interfold_event/mod.rs @@ -39,6 +39,7 @@ mod e3_failed; mod e3_request_complete; mod e3_requested; mod e3_stage_changed; +mod effect_retry; mod enable_effects; mod encryption_key_collection_failed; mod encryption_key_created; @@ -119,6 +120,7 @@ pub use e3_request_complete::*; pub use e3_requested::*; pub use e3_stage_changed::*; use e3_utils::{colorize, colorize_event_ids, Color}; +pub use effect_retry::*; pub use enable_effects::*; pub use encryption_key_collection_failed::*; pub use encryption_key_created::*; @@ -347,6 +349,7 @@ pub enum InterfoldEventData { EvmLogObserved(EvmLogObserved), BondOwnerSet(BondOwnerSet), DkgFoldAttestationContextEstablished(DkgFoldAttestationContextEstablished), + EffectRetry(EffectRetry), } impl InterfoldEventData { @@ -670,6 +673,7 @@ impl InterfoldEventData { InterfoldEventData::DkgFoldAttestationContextEstablished(ref data) => { Some(data.e3_id.clone()) } + InterfoldEventData::EffectRetry(ref data) => Some(data.effect().e3_id().clone()), _ => None, } } @@ -780,7 +784,8 @@ impl_event_types!( CommitteeViabilityUpdated, EvmLogObserved, BondOwnerSet, - DkgFoldAttestationContextEstablished + DkgFoldAttestationContextEstablished, + EffectRetry ); impl TryFrom<&InterfoldEvent> for InterfoldError { diff --git a/crates/events/src/interfold_event/signed_proof.rs b/crates/events/src/interfold_event/signed_proof.rs index e4a58827c4..db0403092e 100644 --- a/crates/events/src/interfold_event/signed_proof.rs +++ b/crates/events/src/interfold_event/signed_proof.rs @@ -52,6 +52,11 @@ pub enum ProofType { } impl ProofType { + /// On-chain replay domain used by `SlashingManager` for this proof type. + pub fn onchain_reason(&self) -> [u8; 32] { + keccak256([*self as u8]).0 + } + /// Map this proof type to its corresponding circuit names. pub fn circuit_names(&self) -> Vec { match self { diff --git a/crates/evm/src/ciphernode_registry/actor.rs b/crates/evm/src/ciphernode_registry/actor.rs index fd788d8aec..63790784f2 100644 --- a/crates/evm/src/ciphernode_registry/actor.rs +++ b/crates/evm/src/ciphernode_registry/actor.rs @@ -330,6 +330,7 @@ impl CiphernodeRegistrySolWriter EventType::PublicKeyAggregated, EventType::CommitteeFinalizeRequested, EventType::TicketGenerated, + EventType::EffectRetry, EventType::E3RequestComplete, EventType::Shutdown, ], diff --git a/crates/evm/src/ciphernode_registry/effects.rs b/crates/evm/src/ciphernode_registry/effects.rs index 66f10f8ad2..d2a4449975 100644 --- a/crates/evm/src/ciphernode_registry/effects.rs +++ b/crates/evm/src/ciphernode_registry/effects.rs @@ -109,6 +109,45 @@ async fn committee_finalization_settled( + provider: EthProvider

, + contract_address: Address, + e3_id: E3id, + ticket_number: u64, +) -> Result { + let e3_id: U256 = e3_id.try_into()?; + let contract = ICiphernodeRegistry::new(contract_address, provider.provider()); + match contract + .submitTicket(e3_id, U256::from(ticket_number)) + .call() + .await + { + Ok(_) => Ok(true), + Err(err) => { + let err = anyhow::Error::from(err); + let decoded = decode_error_from_str(&format!("{err:?}")); + if decoded.as_deref().is_some_and(ticket_retry_is_terminal) { + return Ok(false); + } + Err(err) + } + } +} + +fn ticket_retry_is_terminal(message: &str) -> bool { + [ + "NodeAlreadySubmitted", + "CommitteeAlreadyFinalized", + "CommitteeDeadlineReached", + "CommitteeNotRequested", + "NodeNotEligible", + ] + .iter() + .any(|terminal| message.contains(terminal)) +} + pub async fn finalize_committee_on_registry( provider: EthProvider

, contract_address: Address, @@ -361,3 +400,16 @@ pub async fn fetch_accusation_vote_validity( Ok(Some(validity)) } } + +#[cfg(test)] +mod tests { + use super::ticket_retry_is_terminal; + + #[test] + fn ticket_preflight_only_suppresses_terminal_contract_states() { + assert!(ticket_retry_is_terminal("NodeAlreadySubmitted()")); + assert!(ticket_retry_is_terminal("CommitteeDeadlineReached()")); + assert!(!ticket_retry_is_terminal("RpcTransportError")); + assert!(!ticket_retry_is_terminal("SubmissionWindowNotClosed()")); + } +} diff --git a/crates/evm/src/ciphernode_registry/handlers.rs b/crates/evm/src/ciphernode_registry/handlers.rs index a41ae0324b..ccc1a6b71e 100644 --- a/crates/evm/src/ciphernode_registry/handlers.rs +++ b/crates/evm/src/ciphernode_registry/handlers.rs @@ -57,6 +57,24 @@ impl Handler ctx.notify(data); } } + InterfoldEventData::EffectRetry(retry) => match retry.into_effect() { + InterfoldEventData::PublicKeyAggregated(data) + if self.provider.chain_id() == data.e3_id.chain_id() => + { + ctx.notify(data); + } + InterfoldEventData::CommitteeFinalizeRequested(data) + if self.provider.chain_id() == data.e3_id.chain_id() => + { + ctx.notify(data); + } + InterfoldEventData::TicketGenerated(data) + if self.provider.chain_id() == data.e3_id.chain_id() => + { + ctx.notify(data); + } + _ => {} + }, InterfoldEventData::E3RequestComplete(data) => self.notify_sync(ctx, data), InterfoldEventData::Shutdown(data) => self.notify_sync(ctx, data), _ => (), @@ -143,6 +161,29 @@ impl Handler Box::pin(async move { info!("Submitting ticket {} for E3 {:?}", ticket_id, e3_id); + match should_submit_ticket( + provider.clone(), + contract_address, + e3_id.clone(), + ticket_id, + ) + .await + { + Ok(false) => { + info!(e3_id = %e3_id, "Skipping submitTicket; on-chain state already makes the intent terminal"); + return; + } + Err(err) => { + error!( + "Failed to preflight submitTicket: {}", + format_evm_error(&err) + ); + bus.err(EType::Evm, err); + return; + } + Ok(true) => {} + } + let result = submit_ticket_to_registry(provider, contract_address, e3_id, ticket_id) .await; diff --git a/crates/evm/src/contracts.rs b/crates/evm/src/contracts.rs index 5937782527..5fd5d9ae2d 100644 --- a/crates/evm/src/contracts.rs +++ b/crates/evm/src/contracts.rs @@ -95,6 +95,7 @@ sol! { // ── View functions ────────────────────────────────────────────────── function ciphernodeRegistry() external view returns (address); + function evidenceConsumed(bytes32 evidenceKey) external view returns (bool); // ── Events ────────────────────────────────────────────────────────── event SlashExecuted( diff --git a/crates/evm/src/interfold_writing/actor.rs b/crates/evm/src/interfold_writing/actor.rs index f8b831c6a8..616116ee92 100644 --- a/crates/evm/src/interfold_writing/actor.rs +++ b/crates/evm/src/interfold_writing/actor.rs @@ -70,6 +70,7 @@ impl InterfoldSolWriter

{ EventType::AggregatorChanged, EventType::PlaintextAggregated, EventType::E3StageChanged, + EventType::EffectRetry, EventType::E3RequestComplete, EventType::Shutdown, ], diff --git a/crates/evm/src/interfold_writing/effects.rs b/crates/evm/src/interfold_writing/effects.rs index 246fbec201..8a735b9fb8 100644 --- a/crates/evm/src/interfold_writing/effects.rs +++ b/crates/evm/src/interfold_writing/effects.rs @@ -91,3 +91,41 @@ pub(in crate::actors::interfold_sol_writer) async fn process_e3_failure< require_successful_receipt("process E3 failure", &receipt)?; Ok(receipt) } + +pub(in crate::actors::interfold_sol_writer) async fn should_process_e3_failure< + P: Provider + WalletProvider + Clone, +>( + provider: EthProvider

, + contract_address: Address, + e3_id: E3id, +) -> Result { + let e3_id: U256 = e3_id.try_into()?; + let contract = IInterfold::new(contract_address, provider.provider()); + match contract.processE3Failure(e3_id).call().await { + Ok(_) => Ok(true), + Err(err) => { + let err = anyhow::Error::from(err); + let decoded = crate::domain::error_decoder::decode_error_from_str(&format!("{err:?}")); + if decoded.as_deref().is_some_and(failure_retry_is_terminal) { + return Ok(false); + } + Err(err) + } + } +} + +fn failure_retry_is_terminal(message: &str) -> bool { + message.contains("NoPaymentToRefund") +} + +#[cfg(test)] +mod tests { + use super::failure_retry_is_terminal; + + #[test] + fn failure_preflight_only_suppresses_an_already_processed_refund() { + assert!(failure_retry_is_terminal("NoPaymentToRefund(7)")); + assert!(!failure_retry_is_terminal("E3NotFailed(7)")); + assert!(!failure_retry_is_terminal("RpcTransportError")); + } +} diff --git a/crates/evm/src/interfold_writing/handlers.rs b/crates/evm/src/interfold_writing/handlers.rs index e58c8df16e..84a2886726 100644 --- a/crates/evm/src/interfold_writing/handlers.rs +++ b/crates/evm/src/interfold_writing/handlers.rs @@ -29,6 +29,20 @@ impl Handler ctx.notify(data); } } + InterfoldEventData::EffectRetry(retry) => match retry.into_effect() { + InterfoldEventData::PlaintextAggregated(data) + if self.provider.chain_id() == data.e3_id.chain_id() => + { + ctx.notify(data); + } + InterfoldEventData::E3StageChanged(data) + if data.new_stage == E3Stage::Failed + && self.provider.chain_id() == data.e3_id.chain_id() => + { + ctx.notify(data); + } + _ => {} + }, InterfoldEventData::E3RequestComplete(data) => self.notify_sync(ctx, data), InterfoldEventData::Shutdown(data) => self.notify_sync(ctx, data), _ => (), @@ -192,7 +206,27 @@ impl Handler let e3_id = msg.e3_id.clone(); let contract_address = self.contract_address; let provider = self.provider.clone(); + let bus = self.bus.clone(); async move { + match should_process_e3_failure(provider.clone(), contract_address, e3_id.clone()) + .await + { + Ok(false) => { + info!(e3_id = %e3_id, "Skipping processE3Failure; refund was already processed"); + return; + } + Err(err) => { + bus.err( + EType::Evm, + anyhow::anyhow!( + "Error preflighting E3 failure processing: {}", + format_evm_error(&err) + ), + ); + return; + } + Ok(true) => {} + } let result = process_e3_failure(provider, contract_address, e3_id.clone()).await; match result { Ok(receipt) => { diff --git a/crates/evm/src/slashing_writing/actor.rs b/crates/evm/src/slashing_writing/actor.rs index e97fcaa2f4..a9f83fdb04 100644 --- a/crates/evm/src/slashing_writing/actor.rs +++ b/crates/evm/src/slashing_writing/actor.rs @@ -86,6 +86,8 @@ impl SlashingManagerSolWriter

bus.subscribe_all( &[ EventType::AccusationQuorumReached, + EventType::EffectRetry, + EventType::EvmLogObserved, EventType::EffectsEnabled, EventType::Shutdown, ], diff --git a/crates/evm/src/slashing_writing/effects.rs b/crates/evm/src/slashing_writing/effects.rs index 295e38f0a6..96027ef3de 100644 --- a/crates/evm/src/slashing_writing/effects.rs +++ b/crates/evm/src/slashing_writing/effects.rs @@ -84,6 +84,17 @@ pub(in crate::actors::slashing_manager_sol_writer) async fn submit_slash_proposa .await } +pub(in crate::actors::slashing_manager_sol_writer) async fn slash_evidence_consumed< + P: Provider + WalletProvider + Clone, +>( + provider: EthProvider

, + contract_address: Address, + key: &SlashIntentKey, +) -> Result { + let contract = ISlashingManager::new(contract_address, provider.provider()); + Ok(contract.evidenceConsumed(key.evidence_key()).call().await?) +} + async fn resolve_party_id_for_operator( provider: EthProvider

, contract_address: Address, diff --git a/crates/evm/src/slashing_writing/handlers.rs b/crates/evm/src/slashing_writing/handlers.rs index 09a86364b6..0c6f7146fe 100644 --- a/crates/evm/src/slashing_writing/handlers.rs +++ b/crates/evm/src/slashing_writing/handlers.rs @@ -2,7 +2,7 @@ //! Admission, scheduling, and submission-outcome handlers. -use super::effects::submit_slash_proposal; +use super::effects::{slash_evidence_consumed, submit_slash_proposal}; use super::*; impl Handler @@ -11,7 +11,11 @@ impl Handler type Result = (); fn handle(&mut self, msg: InterfoldEvent, ctx: &mut Self::Context) -> Self::Result { - match msg.into_data() { + let data = match msg.into_data() { + InterfoldEventData::EffectRetry(retry) => retry.into_effect(), + data => data, + }; + match data { InterfoldEventData::AccusationQuorumReached(data) => { // Only submit if: // 1. This is the right chain @@ -77,6 +81,11 @@ impl Handler ); } } + InterfoldEventData::EvmLogObserved(observation) => { + if let Some(key) = SlashIntentKey::from_observation(&observation) { + self.submissions.complete_observed(key); + } + } InterfoldEventData::Shutdown(data) => self.notify_sync(ctx, data), _ => (), } @@ -111,31 +120,53 @@ impl Handler tokio::time::sleep(delay).await; } - let result = submit_slash_proposal(provider, contract_address, msg).await; - let terminal = match result { - Ok(receipt) => { - info!(tx=%receipt.transaction_hash, "Submitted attestation-based slash proposal on-chain"); + let terminal = match slash_evidence_consumed( + provider.clone(), + contract_address, + &key, + ) + .await + { + Ok(true) => { + info!(e3_id = %msg.e3_id, "Skipping slash intent; evidence is already consumed on-chain"); true } Err(err) => { - let decoded = format_evm_error(&err); - let benign = decoded.contains("OperatorNotInCommittee") - || decoded.contains("VoterNotInCommittee") - || decoded.contains("DuplicateEvidence"); - if benign { - // Fallback submitters expect DuplicateEvidence reverts - // when the primary submitter has already landed the tx. - // Operator/VoterNotInCommittee indicate a stale off-chain accusation - // (e.g. cross-E3 race) — not a node-local fault. - warn!("Slash submission skipped (rank {rank}): {decoded}"); - } else { - bus.err( - EType::Evm, - anyhow::anyhow!("Error submitting slash proposal: {decoded}"), - ); - } - benign + bus.err( + EType::Evm, + anyhow::anyhow!( + "Error preflighting slash evidence replay: {}", + format_evm_error(&err) + ), + ); + false } + Ok(false) => match submit_slash_proposal(provider, contract_address, msg).await + { + Ok(receipt) => { + info!(tx=%receipt.transaction_hash, "Submitted attestation-based slash proposal on-chain"); + true + } + Err(err) => { + let decoded = format_evm_error(&err); + let benign = decoded.contains("OperatorNotInCommittee") + || decoded.contains("VoterNotInCommittee") + || decoded.contains("DuplicateEvidence"); + if benign { + // Fallback submitters expect DuplicateEvidence reverts + // when the primary submitter has already landed the tx. + // Operator/VoterNotInCommittee indicate a stale off-chain accusation + // (e.g. cross-E3 race) — not a node-local fault. + warn!("Slash submission skipped (rank {rank}): {decoded}"); + } else { + bus.err( + EType::Evm, + anyhow::anyhow!("Error submitting slash proposal: {decoded}"), + ); + } + benign + } + }, }; if let Err(error) = address .send(SlashSubmissionFinished { key, terminal }) diff --git a/crates/evm/src/slashing_writing/workflow.rs b/crates/evm/src/slashing_writing/workflow.rs index 71b7e3ea4f..1521d8c8a3 100644 --- a/crates/evm/src/slashing_writing/workflow.rs +++ b/crates/evm/src/slashing_writing/workflow.rs @@ -17,9 +17,9 @@ use std::{ time::Duration, }; -use alloy::primitives::{Address, U256}; +use alloy::primitives::{keccak256, Address, B256, U256}; use anyhow::{Context, Result}; -use e3_events::{AccusationOutcome, AccusationQuorumReached}; +use e3_events::{AccusationOutcome, AccusationQuorumReached, EvmLogObserved}; /// Maximum number of voters eligible to attempt on-chain submission. /// Rank 0 submits immediately, rank 1 after one delay interval, etc. @@ -53,6 +53,39 @@ impl SlashIntentKey { proof_type: event.proof_type as u8, }) } + + pub(crate) fn from_observation(event: &EvmLogObserved) -> Option { + if event.contract != "SlashingManager" || event.event_name != "SlashProposed" { + return None; + } + let e3_id: U256 = event.e3_id.clone()?.try_into().ok()?; + let topic = event.topics.get(3)?; + let hex = topic.strip_prefix("0x").unwrap_or(topic); + if hex.len() != 64 || !hex.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return None; + } + let operator = format!("0x{}", &hex[24..]).parse().ok()?; + let data = event.data.extract_bytes(); + let reason = data.get(..32)?; + let proof_type = + (0u8..=10).find(|candidate| keccak256([*candidate]).as_slice() == reason)?; + + Some(Self { + chain_id: event.chain_id, + e3_id, + operator, + proof_type, + }) + } + + pub(crate) fn evidence_key(&self) -> B256 { + let mut encoded = Vec::with_capacity(32 + 32 + 20 + 1); + encoded.extend_from_slice(&U256::from(self.chain_id).to_be_bytes::<32>()); + encoded.extend_from_slice(&self.e3_id.to_be_bytes::<32>()); + encoded.extend_from_slice(self.operator.as_slice()); + encoded.push(self.proof_type); + keccak256(encoded) + } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -124,6 +157,12 @@ impl SlashSubmissionGate { self.completed.insert(key.clone()); } } + + pub(crate) fn complete_observed(&mut self, key: SlashIntentKey) { + self.deferred.remove(&key); + self.in_flight.remove(&key); + self.completed.insert(key); + } } /// Determine this node's submission rank: its position in the voter set after @@ -160,9 +199,19 @@ pub(crate) fn submission_delay(rank: usize) -> Duration { mod tests { use super::*; use alloy::primitives::{Bytes, B256}; - use e3_events::{AccusationQuorumReached, AccusationVote, E3id, ProofType}; + use alloy::sol_types::SolValue; + use e3_events::{AccusationQuorumReached, AccusationVote, E3id, EvmLogObserved, ProofType}; use e3_utils::ArcBytes; + alloy::sol! { + struct EvidenceKeyDomain { + uint256 chainId; + uint256 e3Id; + address operator; + uint8 proofType; + } + } + fn vote(voter: Address) -> AccusationVote { AccusationVote { e3_id: E3id::new("1", 1), @@ -187,6 +236,26 @@ mod tests { } } + fn slash_proposed() -> EvmLogObserved { + let mut data = keccak256([ProofType::C0PkBfv as u8]).to_vec(); + data.resize(32 * 6, 0); + EvmLogObserved { + contract: "SlashingManager".to_owned(), + chain_id: 1, + e3_id: Some(E3id::new("1", 1)), + event_name: "SlashProposed".to_owned(), + signature: None, + known: true, + topics: vec![ + String::new(), + String::new(), + String::new(), + format!("0x{}{}", "00".repeat(12), "08".repeat(20)), + ], + data: ArcBytes::from_bytes(&data), + } + } + #[test] fn test_submission_rank_sorts_ascending() { let a = Address::repeat_byte(0x01); @@ -198,6 +267,35 @@ mod tests { assert_eq!(submission_rank([c, a, b], c), Some(2)); } + #[test] + fn observed_proposal_closes_deferred_replay_intent() { + let mut gate = SlashSubmissionGate::new(); + let (_, decision) = gate.admit(quorum(vec![Address::repeat_byte(1)])).unwrap(); + assert_eq!(decision, SlashSubmissionDecision::Defer); + + let key = SlashIntentKey::from_observation(&slash_proposed()).unwrap(); + gate.complete_observed(key); + + assert!(gate.enable_effects().is_empty()); + } + + #[test] + fn evidence_key_matches_solidity_encode_packed_domain() { + let event = quorum(vec![Address::repeat_byte(1)]); + let key = SlashIntentKey::from_quorum(&event).unwrap(); + let expected = keccak256( + EvidenceKeyDomain { + chainId: U256::from(1), + e3Id: U256::from(1), + operator: Address::repeat_byte(8), + proofType: ProofType::C0PkBfv as u8, + } + .abi_encode_packed(), + ); + + assert_eq!(key.evidence_key(), expected); + } + #[test] fn test_submission_rank_none_when_not_voter() { let a = Address::repeat_byte(0x01); diff --git a/crates/multithread/src/effect_gate.rs b/crates/multithread/src/effect_gate.rs index c6b5b1eabc..92dd517ada 100644 --- a/crates/multithread/src/effect_gate.rs +++ b/crates/multithread/src/effect_gate.rs @@ -8,8 +8,8 @@ use actix::{Actor, Context, Handler, Recipient}; use e3_events::{ - ComputeRequestKind, E3Stage, E3id, Event, EventContextAccessors, EventSubscriber, EventType, - InterfoldEvent, InterfoldEventData, + ComputeRequestKind, CorrelationId, E3Stage, E3id, Event, EventContextAccessors, + EventSubscriber, EventType, InterfoldEvent, InterfoldEventData, }; use e3_utils::MAILBOX_LIMIT; use std::collections::{HashMap, HashSet}; @@ -36,6 +36,7 @@ pub(crate) struct ComputeEffectGate { enabled: bool, pending: HashMap, forwarded: HashSet, + correlations: HashMap<(E3id, CorrelationId), RequestKey>, } impl ComputeEffectGate { @@ -45,6 +46,7 @@ impl ComputeEffectGate { enabled: false, pending: HashMap::new(), forwarded: HashSet::new(), + correlations: HashMap::new(), } } @@ -77,6 +79,9 @@ impl ComputeEffectGate { bus.subscribe_all( &[ EventType::ComputeRequest, + EventType::ComputeResponse, + EventType::ComputeRequestError, + EventType::EffectRetry, EventType::EffectsEnabled, EventType::E3RequestComplete, EventType::E3Failed, @@ -91,6 +96,8 @@ impl ComputeEffectGate { return; }; let key = (request.e3_id.clone(), request.request.clone()); + self.correlations + .insert((request.e3_id.clone(), request.correlation_id), key.clone()); match self.pending.entry(key) { std::collections::hash_map::Entry::Occupied(mut entry) if event.ts() > entry.get().ts() => @@ -120,6 +127,17 @@ impl ComputeEffectGate { .retain(|(pending_id, _), _| pending_id != e3_id); self.forwarded .retain(|(forwarded_id, _)| forwarded_id != e3_id); + self.correlations + .retain(|(correlation_e3_id, _), _| correlation_e3_id != e3_id); + } + + fn complete(&mut self, e3_id: &E3id, correlation_id: CorrelationId) { + let Some(key) = self.correlations.remove(&(e3_id.clone(), correlation_id)) else { + return; + }; + self.pending.remove(&key); + self.forwarded.remove(&key); + self.correlations.retain(|_, mapped| mapped != &key); } } @@ -139,6 +157,18 @@ impl Handler for ComputeEffectGate { InterfoldEventData::ComputeRequest(_) if self.enabled => { self.forward(event); } + InterfoldEventData::EffectRetry(retry) + if self.enabled && retry.effect().is_compute() => + { + self.forward(event); + } + InterfoldEventData::ComputeResponse(response) => { + self.complete(&response.e3_id, response.correlation_id) + } + InterfoldEventData::ComputeRequestError(error) => { + let request = error.request(); + self.complete(&request.e3_id, request.correlation_id); + } InterfoldEventData::ComputeRequest(_) => self.queue(event), InterfoldEventData::EffectsEnabled(_) => self.enable(), InterfoldEventData::E3RequestComplete(complete) => self.cancel(&complete.e3_id), @@ -158,9 +188,9 @@ mod tests { use super::*; use actix::{Message, ResponseFuture}; use e3_events::{ - ComputeRequest, CorrelationId, E3RequestComplete, EffectsEnabled, - EventConstructorWithTimestamp, EventSource, InterfoldEvent, PkBfvProofRequest, Unsequenced, - ZkRequest, + ComputeRequest, ComputeRequestError, ComputeRequestErrorKind, CorrelationId, + E3RequestComplete, EffectsEnabled, EventConstructorWithTimestamp, EventSource, + InterfoldEvent, PkBfvProofRequest, Unsequenced, ZkError, ZkRequest, }; use e3_fhe_params::BfvPreset; use e3_utils::ArcBytes; @@ -241,6 +271,25 @@ mod tests { .into_sequenced(2) } + fn compute_error(correlation_id: CorrelationId) -> InterfoldEvent { + let InterfoldEventData::ComputeRequest(request) = compute(correlation_id, 10).into_data() + else { + unreachable!(); + }; + InterfoldEvent::::new_with_timestamp( + ComputeRequestError::new( + ComputeRequestErrorKind::Zk(ZkError::InvalidParams("test".to_owned())), + request, + ) + .into(), + None, + 20, + None, + EventSource::Local, + ) + .into_sequenced(2) + } + #[actix::test] async fn buffers_until_enabled_and_keeps_newest_semantic_retry() { let recorder = Recorder::default().start(); @@ -268,6 +317,19 @@ mod tests { assert!(recorder.send(Received).await.unwrap().is_empty()); } + #[actix::test] + async fn completed_compute_is_not_released_after_replay() { + let recorder = Recorder::default().start(); + let gate = ComputeEffectGate::new(recorder.clone().recipient()).start(); + let correlation_id = CorrelationId::new(); + + gate.send(compute(correlation_id, 10)).await.unwrap(); + gate.send(compute_error(correlation_id)).await.unwrap(); + gate.send(effects_enabled()).await.unwrap(); + + assert!(recorder.send(Received).await.unwrap().is_empty()); + } + #[actix::test] async fn drops_redriven_duplicate_after_enable() { let recorder = Recorder::default().start(); diff --git a/crates/multithread/src/multithread.rs b/crates/multithread/src/multithread.rs index 3945200706..3274a1ba2e 100644 --- a/crates/multithread/src/multithread.rs +++ b/crates/multithread/src/multithread.rs @@ -189,6 +189,10 @@ impl Handler for Multithread { type Result = (); fn handle(&mut self, msg: InterfoldEvent, ctx: &mut Self::Context) -> Self::Result { let (data, ec) = msg.into_components(); + let data = match data { + InterfoldEventData::EffectRetry(retry) => retry.into_effect(), + data => data, + }; if let InterfoldEventData::ComputeRequest(data) = data { ctx.notify(TypedEvent::new(data, ec)) } diff --git a/crates/net/src/document_publishing/handlers.rs b/crates/net/src/document_publishing/handlers.rs index 184f41fafd..ab26667f41 100644 --- a/crates/net/src/document_publishing/handlers.rs +++ b/crates/net/src/document_publishing/handlers.rs @@ -15,6 +15,10 @@ impl Handler for DocumentPublisher { type Result = (); fn handle(&mut self, msg: InterfoldEvent, ctx: &mut Self::Context) -> Self::Result { let (msg, ec) = msg.into_components(); + let msg = match msg { + InterfoldEventData::EffectRetry(retry) => retry.into_effect(), + msg => msg, + }; match msg { InterfoldEventData::PublishDocumentRequested(data) => { ctx.notify(TypedEvent::new(data, ec)) diff --git a/crates/sync/src/lib.rs b/crates/sync/src/lib.rs index 2d07d366fc..ee8b8795b6 100644 --- a/crates/sync/src/lib.rs +++ b/crates/sync/src/lib.rs @@ -5,6 +5,7 @@ // or FITNESS FOR A PARTICULAR PURPOSE. mod domain; +mod open_effects; mod replay_spool; mod repo; mod runtime; diff --git a/crates/sync/src/open_effects.rs b/crates/sync/src/open_effects.rs new file mode 100644 index 0000000000..d3052e7e0a --- /dev/null +++ b/crates/sync/src/open_effects.rs @@ -0,0 +1,548 @@ +// SPDX-License-Identifier: LGPL-3.0-only +// +// This file is provided WITHOUT ANY WARRANTY; +// without even the implied warranty of MERCHANTABILITY +// or FITNESS FOR A PARTICULAR PURPOSE. + +//! Durable open-loop detection for startup effect recovery. + +use crate::replay_spool::{query_page, REPLAY_QUERY_PAGE_SIZE}; +use actix::Recipient; +use anyhow::{bail, Context, Result}; +use e3_events::{ + AggregateId, ComputeRequestKind, CorrelationId, DocumentMeta, E3Stage, E3id, Event, + EventContextAccessors, EventContextSeq, EventSource, EventStoreQueryBy, InterfoldEvent, + InterfoldEventData, SeqAgg, TicketId, +}; +use e3_utils::ArcBytes; +use std::collections::HashMap; + +const MAX_OPEN_EFFECTS: usize = 50_000; +const MAX_OPEN_EFFECT_BYTES: usize = 128 * 1024 * 1024; + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +enum EffectKey { + Compute { + e3_id: E3id, + request: Box, + }, + Ticket { + e3_id: E3id, + node: String, + ticket_id: u64, + }, + FinalizeCommittee(E3id), + PublishCommittee(E3id), + PublishPlaintext(E3id), + ProcessFailure(E3id), + PublishDocument { + meta: DocumentMeta, + value: ArcBytes, + }, + Slash { + e3_id: E3id, + operator: String, + reason: [u8; 32], + }, +} + +impl EffectKey { + fn e3_id(&self) -> &E3id { + match self { + Self::Compute { e3_id, .. } + | Self::Ticket { e3_id, .. } + | Self::Slash { e3_id, .. } + | Self::FinalizeCommittee(e3_id) + | Self::PublishCommittee(e3_id) + | Self::PublishPlaintext(e3_id) + | Self::ProcessFailure(e3_id) => e3_id, + Self::PublishDocument { meta, .. } => &meta.e3_id, + } + } +} + +struct OpenIntent { + timestamp: u128, + bytes: usize, + payload: InterfoldEventData, +} + +#[derive(Default)] +struct OpenEffectDetector { + intents: HashMap, + compute_requests: HashMap<(E3id, CorrelationId), EffectKey>, + open_bytes: usize, +} + +impl OpenEffectDetector { + fn observe(&mut self, event: &InterfoldEvent) -> Result<()> { + let timestamp = event.ts(); + match event.get_data() { + // Only the node that originated an effect owns its retry. Canonical EVM failure + // transitions are the exception: every configured chain writer already consumes them. + InterfoldEventData::ComputeRequest(data) if event.source() == EventSource::Local => { + let key = EffectKey::Compute { + e3_id: data.e3_id.clone(), + request: Box::new(data.request.clone()), + }; + self.compute_requests + .insert((data.e3_id.clone(), data.correlation_id), key.clone()); + if self.compute_requests.len() > MAX_OPEN_EFFECTS { + bail!( + "open effect recovery exceeds compute-correlation bound of {}", + MAX_OPEN_EFFECTS + ); + } + self.insert(key, timestamp, event.get_data().clone())?; + } + InterfoldEventData::TicketGenerated(data) if event.source() == EventSource::Local => { + let TicketId::Score(ticket_id) = data.ticket_id; + self.insert( + EffectKey::Ticket { + e3_id: data.e3_id.clone(), + node: normalize_node(&data.node), + ticket_id, + }, + timestamp, + event.get_data().clone(), + )?; + } + InterfoldEventData::CommitteeFinalizeRequested(data) + if event.source() == EventSource::Local => + { + self.insert( + EffectKey::FinalizeCommittee(data.e3_id.clone()), + timestamp, + event.get_data().clone(), + )?; + } + InterfoldEventData::PublicKeyAggregated(data) + if event.source() == EventSource::Local => + { + self.insert( + EffectKey::PublishCommittee(data.e3_id.clone()), + timestamp, + event.get_data().clone(), + )?; + } + InterfoldEventData::PlaintextAggregated(data) + if event.source() == EventSource::Local => + { + self.insert( + EffectKey::PublishPlaintext(data.e3_id.clone()), + timestamp, + event.get_data().clone(), + )?; + } + InterfoldEventData::PublishDocumentRequested(data) + if event.source() == EventSource::Local => + { + self.insert( + EffectKey::PublishDocument { + meta: data.meta.clone(), + value: data.value.clone(), + }, + timestamp, + event.get_data().clone(), + )?; + } + InterfoldEventData::AccusationQuorumReached(data) + if event.source() == EventSource::Local + && matches!( + data.outcome, + e3_events::AccusationOutcome::AccusedFaulted + | e3_events::AccusationOutcome::Equivocation + ) + && !data.votes_for.is_empty() + && !data.evidence.is_empty() => + { + self.insert( + EffectKey::Slash { + e3_id: data.e3_id.clone(), + operator: normalize_node(&data.accused.to_string()), + reason: data.proof_type.onchain_reason(), + }, + timestamp, + event.get_data().clone(), + )?; + } + InterfoldEventData::E3StageChanged(data) if data.new_stage == E3Stage::Failed => { + self.close_e3(&data.e3_id, true); + self.insert( + EffectKey::ProcessFailure(data.e3_id.clone()), + timestamp, + event.get_data().clone(), + )?; + } + + InterfoldEventData::ComputeResponse(data) => { + self.complete_compute(&data.e3_id, data.correlation_id) + } + InterfoldEventData::ComputeRequestError(data) => { + let request = data.request(); + self.complete_compute(&request.e3_id, request.correlation_id); + } + InterfoldEventData::TicketSubmitted(data) => self.remove(&EffectKey::Ticket { + e3_id: data.e3_id.clone(), + node: normalize_node(&data.node), + ticket_id: data.ticket_id, + }), + InterfoldEventData::CommitteeFinalized(data) => { + self.remove(&EffectKey::FinalizeCommittee(data.e3_id.clone())); + self.close_tickets(&data.e3_id); + } + InterfoldEventData::CommitteeFormationFailed(data) => { + self.remove(&EffectKey::FinalizeCommittee(data.e3_id.clone())); + self.close_tickets(&data.e3_id); + } + InterfoldEventData::CommitteePublished(data) => { + self.remove(&EffectKey::PublishCommittee(data.e3_id.clone())); + } + InterfoldEventData::PlaintextOutputPublished(data) => { + self.remove(&EffectKey::PublishPlaintext(data.e3_id.clone())); + } + InterfoldEventData::DocumentReceived(data) => { + self.remove(&EffectKey::PublishDocument { + meta: data.meta.clone(), + value: data.value.clone(), + }); + } + InterfoldEventData::EvmLogObserved(data) => self.observe_evm_completion(data), + InterfoldEventData::E3RequestComplete(data) => self.close_e3(&data.e3_id, false), + InterfoldEventData::E3Failed(data) => self.close_e3(&data.e3_id, true), + InterfoldEventData::E3StageChanged(data) if data.new_stage == E3Stage::Complete => { + self.close_e3(&data.e3_id, false); + } + _ => {} + } + Ok(()) + } + + fn insert( + &mut self, + key: EffectKey, + timestamp: u128, + payload: InterfoldEventData, + ) -> Result<()> { + let bytes = bincode::serialized_size(&payload) + .context("failed to size recoverable effect intent")? + .try_into() + .context("recoverable effect size does not fit usize")?; + if let Some(previous) = self.intents.remove(&key) { + self.open_bytes = self.open_bytes.saturating_sub(previous.bytes); + } + self.open_bytes = self + .open_bytes + .checked_add(bytes) + .context("open effect byte count overflow")?; + self.intents.insert( + key, + OpenIntent { + timestamp, + bytes, + payload, + }, + ); + if self.intents.len() > MAX_OPEN_EFFECTS || self.open_bytes > MAX_OPEN_EFFECT_BYTES { + bail!( + "open effect recovery exceeds startup bounds: {} intents / {} bytes (limits: {} / {})", + self.intents.len(), + self.open_bytes, + MAX_OPEN_EFFECTS, + MAX_OPEN_EFFECT_BYTES + ); + } + Ok(()) + } + + fn remove(&mut self, key: &EffectKey) { + if let Some(intent) = self.intents.remove(key) { + self.open_bytes = self.open_bytes.saturating_sub(intent.bytes); + } + if matches!(key, EffectKey::Compute { .. }) { + self.compute_requests.retain(|_, mapped| mapped != key); + } + } + + fn complete_compute(&mut self, e3_id: &E3id, correlation_id: CorrelationId) { + if let Some(key) = self + .compute_requests + .remove(&(e3_id.clone(), correlation_id)) + { + self.remove(&key); + } + } + + fn close_tickets(&mut self, e3_id: &E3id) { + self.retain(|key| !matches!(key, EffectKey::Ticket { e3_id: id, .. } if id == e3_id)); + } + + fn close_e3(&mut self, e3_id: &E3id, preserve_failure_processing: bool) { + self.retain(|key| { + key.e3_id() != e3_id + || (preserve_failure_processing + && matches!(key, EffectKey::ProcessFailure(_) | EffectKey::Slash { .. })) + }); + self.compute_requests + .retain(|(request_e3_id, _), _| request_e3_id != e3_id); + } + + fn retain(&mut self, keep: impl Fn(&EffectKey) -> bool) { + let mut removed_bytes = 0usize; + self.intents.retain(|key, intent| { + let retain = keep(key); + if !retain { + removed_bytes = removed_bytes.saturating_add(intent.bytes); + } + retain + }); + self.open_bytes = self.open_bytes.saturating_sub(removed_bytes); + } + + fn observe_evm_completion(&mut self, event: &e3_events::EvmLogObserved) { + let Some(e3_id) = event.e3_id.clone() else { + return; + }; + if event.contract == "Interfold" && event.event_name == "E3FailureProcessed" { + self.remove(&EffectKey::ProcessFailure(e3_id)); + return; + } + if event.contract != "SlashingManager" || event.event_name != "SlashProposed" { + return; + } + let Some(operator) = event.topics.get(3).and_then(|topic| topic_address(topic)) else { + return; + }; + let data = event.data.extract_bytes(); + let Some(reason) = data.get(..32).and_then(|bytes| bytes.try_into().ok()) else { + return; + }; + self.remove(&EffectKey::Slash { + e3_id, + operator, + reason, + }); + } + + fn finish(self) -> Vec { + let mut open: Vec<_> = self.intents.into_values().collect(); + open.sort_by_key(|intent| intent.timestamp); + open.into_iter().map(|intent| intent.payload).collect() + } +} + +fn normalize_node(node: &str) -> String { + node.to_ascii_lowercase() +} + +fn topic_address(topic: &str) -> Option { + let hex = topic.strip_prefix("0x").unwrap_or(topic); + if hex.len() != 64 || !hex.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return None; + } + Some(format!("0x{}", &hex[24..]).to_ascii_lowercase()) +} + +/// Scan the complete durable log in bounded pages and return only effect intents +/// for which no matching completion or terminal lifecycle event exists. +pub(crate) async fn detect_open_effects( + eventstore: &Recipient>, + mut aggregates: Vec, +) -> Result> { + aggregates.sort_unstable(); + aggregates.dedup(); + let mut detector = OpenEffectDetector::default(); + + for aggregate_id in aggregates { + let mut cursor = 1u64; + loop { + let page = query_page(eventstore, aggregate_id, cursor).await?; + if page.is_empty() { + break; + } + if page.len() > REPLAY_QUERY_PAGE_SIZE { + bail!( + "EventStore returned {} effect-scan events for aggregate {}, exceeding page limit {}", + page.len(), + aggregate_id, + REPLAY_QUERY_PAGE_SIZE + ); + } + + let mut expected_sequence = cursor; + for event in &page { + if event.aggregate_id() != aggregate_id || event.seq() != expected_sequence { + bail!( + "EventStore effect scan lost continuity for aggregate {}: expected sequence {}, got aggregate {} sequence {}", + aggregate_id, + expected_sequence, + event.aggregate_id(), + event.seq() + ); + } + detector.observe(event)?; + expected_sequence = expected_sequence + .checked_add(1) + .context("EventStore effect-scan sequence overflow")?; + } + cursor = expected_sequence; + if page.len() < REPLAY_QUERY_PAGE_SIZE { + break; + } + } + } + + Ok(detector.finish()) +} + +#[cfg(test)] +mod tests { + use super::*; + use e3_events::{ + CommitteeFinalizeRequested, CommitteeFinalized, E3StageChanged, + EventConstructorWithTimestamp, TicketGenerated, TicketSubmitted, Unsequenced, + }; + + fn event(data: InterfoldEventData, timestamp: u128, sequence: u64) -> InterfoldEvent { + InterfoldEvent::::new_with_timestamp( + data, + None, + timestamp, + None, + EventSource::Local, + ) + .into_sequenced(sequence) + } + + #[test] + fn detects_only_the_unmatched_blockchain_write() { + let e3_id = E3id::new("7", 1); + let mut detector = OpenEffectDetector::default(); + detector + .observe(&event( + TicketGenerated { + e3_id: e3_id.clone(), + ticket_id: TicketId::Score(11), + node: "0xAAA".to_owned(), + party_index: None, + } + .into(), + 1, + 1, + )) + .unwrap(); + detector + .observe(&event( + CommitteeFinalizeRequested { + e3_id: e3_id.clone(), + } + .into(), + 2, + 2, + )) + .unwrap(); + detector + .observe(&event( + TicketSubmitted { + e3_id, + node: "0xaaa".to_owned(), + ticket_id: 11, + score: "11".to_owned(), + chain_id: 1, + } + .into(), + 3, + 3, + )) + .unwrap(); + + let open = detector.finish(); + assert_eq!(open.len(), 1); + assert!(matches!( + open[0], + InterfoldEventData::CommitteeFinalizeRequested(_) + )); + } + + #[test] + fn committee_completion_closes_finalize_and_ticket_effects() { + let e3_id = E3id::new("8", 1); + let mut detector = OpenEffectDetector::default(); + detector + .observe(&event( + TicketGenerated { + e3_id: e3_id.clone(), + ticket_id: TicketId::Score(1), + node: "0x01".to_owned(), + party_index: None, + } + .into(), + 1, + 1, + )) + .unwrap(); + detector + .observe(&event( + CommitteeFinalizeRequested { + e3_id: e3_id.clone(), + } + .into(), + 2, + 2, + )) + .unwrap(); + detector + .observe(&event( + CommitteeFinalized { + e3_id, + committee: vec![], + scores: vec![], + chain_id: 1, + } + .into(), + 3, + 3, + )) + .unwrap(); + + assert!(detector.finish().is_empty()); + } + + #[test] + fn failed_stage_remains_open_until_failure_processing_is_observed() { + let e3_id = E3id::new("9", 1); + let mut detector = OpenEffectDetector::default(); + detector + .observe(&event( + E3StageChanged { + e3_id: e3_id.clone(), + previous_stage: E3Stage::Requested, + new_stage: E3Stage::Failed, + } + .into(), + 1, + 1, + )) + .unwrap(); + assert_eq!(detector.intents.len(), 1); + + detector + .observe(&event( + e3_events::EvmLogObserved { + contract: "Interfold".to_owned(), + chain_id: 1, + e3_id: Some(e3_id), + event_name: "E3FailureProcessed".to_owned(), + signature: None, + known: true, + topics: vec![], + data: ArcBytes::from_bytes(&[]), + } + .into(), + 2, + 2, + )) + .unwrap(); + assert!(detector.finish().is_empty()); + } +} diff --git a/crates/sync/src/replay_spool.rs b/crates/sync/src/replay_spool.rs index c5eb0cee3e..060d5a3b1e 100644 --- a/crates/sync/src/replay_spool.rs +++ b/crates/sync/src/replay_spool.rs @@ -24,7 +24,7 @@ use tracing::info; use crate::{ReplayDecision, SyncPlanner}; -const REPLAY_QUERY_PAGE_SIZE: usize = 1_024; +pub(crate) const REPLAY_QUERY_PAGE_SIZE: usize = 1_024; const REPLAY_MERGE_FAN_IN: usize = 32; const MAX_SPOOLED_EVENT_BYTES: usize = 64 * 1024 * 1024; const REPLAY_PROGRESS_INTERVAL: usize = 10_000; @@ -147,7 +147,7 @@ impl ReplaySpool { } } -async fn query_page( +pub(crate) async fn query_page( eventstore: &Recipient>, aggregate_id: AggregateId, cursor: u64, diff --git a/crates/sync/src/sync/service.rs b/crates/sync/src/sync/service.rs index 8d683d8fe0..22b95800a4 100644 --- a/crates/sync/src/sync/service.rs +++ b/crates/sync/src/sync/service.rs @@ -10,16 +10,17 @@ use crate::domain::{ decide_schema_version, CollectOutcome, HistoricalEvmCollector, SchemaVersionDecision, SnapshotMeta, SyncPlanner, SCHEMA_VERSION, }; +use crate::open_effects::detect_open_effects; use crate::replay_spool::ReplaySpool; use crate::SyncRepositoryFactory; use actix::{Message, Recipient}; use anyhow::{bail, Context, Result}; use e3_data::Repositories; use e3_events::{ - AggregateConfig, BusHandle, CorrelationId, EffectsEnabled, Event, EventPublisher, - EventStoreQueryBy, EventStoreQueryResponse, EventSubscriber, EventType, EvmEventConfig, - HistoricalEvmEventsReceived, HistoricalEvmSyncStart, HistoricalNetSyncStart, InterfoldEvent, - InterfoldEventData, SeqAgg, StoreKeys, SyncEnded, Unsequenced, + AggregateConfig, BusHandle, CorrelationId, EffectRetry, EffectsEnabled, Event, EventFactory, + EventPublisher, EventStoreQueryBy, EventStoreQueryResponse, EventSubscriber, EventType, + EvmEventConfig, HistoricalEvmEventsReceived, HistoricalEvmSyncStart, HistoricalNetSyncStart, + InterfoldEvent, InterfoldEventData, SeqAgg, StoreKeys, SyncEnded, Unsequenced, }; #[cfg(test)] use e3_events::{EventBusBarrier, EventBusFanout, EventContextAccessors}; @@ -30,6 +31,7 @@ use tracing::info; #[cfg(test)] const REPLAY_PROGRESS_INTERVAL: usize = 10_000; +const EFFECT_RETRY_FLUSH_BATCH_SIZE: usize = 1_024; pub async fn sync( bus: &BusHandle, @@ -56,6 +58,16 @@ pub async fn sync( snapshot.aggregates().len() ); + // The snapshot cursor can advance beyond an effect intent even when its completion never + // arrived. Scan the append-only source of truth from sequence one so those pre-snapshot open + // loops are recoverable too; the scan is paged and retains only currently-open intents. + info!("Scanning durable history for incomplete effects..."); + let open_effects = detect_open_effects(eventstore, snapshot.aggregates()).await?; + info!( + open_effects = open_effects.len(), + "Incomplete effect scan finished." + ); + // 1b. Restore the HLC ordering floor from the highest persisted aggregate // timestamp so events created after this restart remain strictly after // durable history, including its logical counter, even if wall time moved backwards. @@ -85,32 +97,6 @@ pub async fn sync( let replayed = replay_spool.replay(bus).await?; info!(replayed_events = replayed, "Events replayed."); - // Loose ends after a crash: - // - // Terminal E3 work that *completed while this node was down* is recovered by the - // historical EVM re-fetch in step 5 below: the terminal on-chain events - // (PlaintextOutputPublished / E3Failed / committee completion) are re-delivered once - // effects are enabled, which re-drives the Sortition release path and frees any tickets - // the node was still holding. So "an E3 finished while we were offline" needs no special - // handling here — it is reconciled by replaying the canonical chain state. - // - // What is intentionally NOT auto-re-driven *here in sync* is this node's *own* in-flight - // request work by replaying the originating request events. Blindly re-publishing the - // originating request event is a no-op: the event bus dedups by EventId (payload hash), so - // the replayed event is dropped. Forcibly minting a fresh EventId to force re-execution is - // unsafe on a value-bearing protocol (it can double-emit or race the canonical chain state) - // and is therefore deliberately left out of the sync path. - // - // Note: this is *not* a global absence of restart recovery. Actors that hold determined, - // idempotent in-flight results re-drive themselves when `EffectsEnabled` is broadcast at the - // end of this sync (e.g. `ThresholdKeyshare::resume_in_flight_work` re-publishes a computed - // keyshare / decryption share). What sync deliberately avoids is replaying *request* events. - // - // Detection of loose ends that cannot be locally re-driven is exposed offline and - // non-destructively via `interfold node validate`, which cross-checks the persisted committee - // slots against terminal events in the log and reports orphaned tickets. See - // `crates/entrypoint/src/validate.rs`. - // 5. Load the historical evm events to memory from all chains info!("Loading historical blockchain events..."); let (addr, rx) = actix_toolbox::mpsc::(256); @@ -155,7 +141,7 @@ pub async fn sync( // 8-10. Enable effects, publish canonical history, then enter live mode. Each phase is fenced // through durable storage and EventBus fanout so aggregate-specific EventStore response order // cannot move history ahead of EffectsEnabled or SyncEnded ahead of history. - publish_reconciled_history(bus, historical).await?; + publish_reconciled_history(bus, open_effects, historical).await?; // normal live operations Ok(()) @@ -163,6 +149,7 @@ pub async fn sync( async fn publish_reconciled_history( bus: &BusHandle, + open_effects: Vec, historical: Vec>, ) -> Result<()> { info!("Enabling effects..."); @@ -170,6 +157,21 @@ async fn publish_reconciled_history( bus.flush_event_pipeline().await?; info!("Effects enabled."); + // A distinct internal envelope targets effect executors without replaying the original + // domain event into state-building actors. Each retry is persisted before fanout, so another + // crash simply leaves the original loop open for the next bounded scan. + let retry_count = open_effects.len(); + for (index, effect) in open_effects.into_iter().enumerate() { + let retry = EffectRetry::new(effect)?; + let event = bus.event_from(retry, None)?; + bus.naked_dispatch_async(event).await?; + if (index + 1).is_multiple_of(EFFECT_RETRY_FLUSH_BATCH_SIZE) { + bus.flush_event_pipeline().await?; + } + } + bus.flush_event_pipeline().await?; + info!(retry_count, "Incomplete effects re-driven."); + info!("Publishing historical events to actors..."); for event in historical { bus.naked_dispatch_async(event).await?; diff --git a/crates/sync/src/sync/tests/gates.rs b/crates/sync/src/sync/tests/gates.rs index ce79dcab0c..67f21ce59c 100644 --- a/crates/sync/src/sync/tests/gates.rs +++ b/crates/sync/src/sync/tests/gates.rs @@ -22,7 +22,11 @@ async fn startup_history_is_fenced_between_effects_and_live_mode() -> anyhow::Re .build(), ]; - publish_reconciled_history(&bus, historical).await?; + let open_effects = vec![e3_events::CommitteeFinalizeRequested { + e3_id: e3_events::E3id::new("3", 1), + } + .into()]; + publish_reconciled_history(&bus, open_effects, historical).await?; let received = history.send(GetEvents::new()).await?; let types = received @@ -31,7 +35,13 @@ async fn startup_history_is_fenced_between_effects_and_live_mode() -> anyhow::Re .collect::>(); assert_eq!( types, - ["EffectsEnabled", "TestEvent", "TestEvent", "SyncEnded"] + [ + "EffectsEnabled", + "EffectRetry", + "TestEvent", + "TestEvent", + "SyncEnded" + ] ); Ok(()) } diff --git a/crates/sync/src/sync/tests/replay.rs b/crates/sync/src/sync/tests/replay.rs index 9497a6f136..d85d80f3b4 100644 --- a/crates/sync/src/sync/tests/replay.rs +++ b/crates/sync/src/sync/tests/replay.rs @@ -29,9 +29,21 @@ async fn infrastructure_events_are_filtered_during_replay() -> anyhow::Result<() .data(make_historical_evm_sync_start()) .seq(4) .build(), + InterfoldEvent::::test_event("retry") + .data( + e3_events::EffectRetry::new( + e3_events::CommitteeFinalizeRequested { + e3_id: e3_events::E3id::new("1", 1), + } + .into(), + ) + .unwrap(), + ) + .seq(5) + .build(), InterfoldEvent::::test_event("after") .id(2) - .seq(5) + .seq(6) .build(), ]; diff --git a/crates/sync/src/sync/workflow.rs b/crates/sync/src/sync/workflow.rs index 30382900f2..52d6e30d22 100644 --- a/crates/sync/src/sync/workflow.rs +++ b/crates/sync/src/sync/workflow.rs @@ -13,8 +13,9 @@ use std::collections::BTreeMap; /// Decision returned for each event encountered during EventStore replay. /// /// Infrastructure events (`SyncEnded`, `EffectsEnabled`, `HistoricalEvmSyncStart`, -/// `HistoricalNetSyncStart`) are re-published by the sync process itself, so replaying them -/// would poison the EventBus deduplication window. They must be skipped during replay. +/// `HistoricalNetSyncStart`, and persisted `EffectRetry` envelopes) are re-published or +/// reconstructed by the sync process itself, so replaying them would poison the EventBus +/// deduplication window. They must be skipped during replay. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ReplayDecision { /// Forward the event to listeners. @@ -49,6 +50,7 @@ impl SyncPlanner { | InterfoldEventData::EffectsEnabled(_) | InterfoldEventData::HistoricalEvmSyncStart(_) | InterfoldEventData::HistoricalNetSyncStart(_) + | InterfoldEventData::EffectRetry(_) | InterfoldEventData::Shutdown(_) ) } @@ -135,15 +137,28 @@ mod tests { .data(Shutdown) .seq(4) .build(); + let retry = InterfoldEvent::::test_event("retry") + .data( + e3_events::EffectRetry::new( + e3_events::CommitteeFinalizeRequested { + e3_id: E3id::new("1", 1), + } + .into(), + ) + .unwrap(), + ) + .seq(5) + .build(); let test_event = InterfoldEvent::::test_event("hello") .id(42) - .seq(5) + .seq(6) .build(); assert!(SyncPlanner::is_infrastructure_event(&sync_ended)); assert!(SyncPlanner::is_infrastructure_event(&effects_enabled)); assert!(SyncPlanner::is_infrastructure_event(&evm_sync_start)); assert!(SyncPlanner::is_infrastructure_event(&shutdown)); + assert!(SyncPlanner::is_infrastructure_event(&retry)); assert!(!SyncPlanner::is_infrastructure_event(&test_event)); }