diff --git a/agent/flow-trace/00_INDEX.md b/agent/flow-trace/00_INDEX.md index 0f07126f71..14869cc86a 100644 --- a/agent/flow-trace/00_INDEX.md +++ b/agent/flow-trace/00_INDEX.md @@ -233,7 +233,7 @@ them are in the reference app (`examples/CRISP`), not the protocol. | 1 | **Deregister-before-slash race** | Resolved | One unresolved-proposal counter covers both lanes. Every authorized current or retained historical slashing manager participates in the BondingRegistry exit gate, so rotation cannot release collateral belonging to an in-flight E3. Ticket withdrawal, ciphernode bond unbonding, deregistration, and exit claims remain blocked until execution, upheld appeal, or permissionless expiry terminates the proposal. | | 2 | **Committee publication decentralized** | Resolved | `publishCommittee()` is permissionless. Off-chain role selection chooses the active aggregator, while on-chain C5 proof verification and the single-publish guard prevent invalid or duplicate committee publication. | | 3 | **`gracePeriod` is dead code** | Medium | `gracePeriod` is stored and validated during config updates but never actually used in any timeout check. Either the deadlines already bake in sufficient buffer, or this is a missing feature. | -| 5 | **Active-job load balancing bug fixed** | Info | Every node builds the same buffered winner set from complete request-time ticket ranges. A selected node applies `active_jobs` only when it decides whether to submit its canonical winning ticket. `E3Failed` and terminal `E3StageChanged` events release local workload slots. | +| 5 | **Active-job load balancing bug fixed** | Info | Every node builds the same buffered winner set from complete request-time ticket ranges. A selected node applies `active_jobs` only when it decides whether to submit its canonical winning ticket. The first committee publication increments the selected nodes; duplicate or conflicting replay for that E3 does not increment them again. `E3Failed` and terminal `E3StageChanged` events release local workload slots. | | 6 | **Committee member expulsion** | Resolved | `SlashingManager` can call `expelCommitteeMember()` mid-DKG. The `Sortition` actor enriches the raw `CommitteeMemberExpelled` event with the expelled member's `party_id` (resolved from its stored `Committee` list) and re-publishes it. `ThresholdKeyshare` uses the enriched `party_id` to update its collectors. The public-key keyshare gate and aggregation reducer normalize both expulsion records and self-reported keyshare nodes to `alloy::Address`, so an expelled member cannot re-enter by changing address casing. `ThresholdKeyshare` itself does not hold committee state. | | Z-03 | **Ban eligibility synchronization** | Resolved | Every completed ban or unban refreshes the affected registered operator in `BondingRegistry`. Active status includes bans from all retained slashing managers, so a banned operator stops counting toward committee capacity and cannot submit new tickets. Existing finalized committees change only through E3-scoped expulsion. | | 7 | **ProofRequestActor failure bridge fixed** | Info | `ProofRequestActor` no longer leaves proof publication suppressed under log-only "will not be published" exits. `ComputeRequestError` and local proof-signing failures for DKG-path proofs (`C0` through `C5`) now emit `E3Failed { failed_at_stage: CommitteeFinalized, reason: DKGInvalidShares }`, while decryption-path proofs (`C6` and `C7`) emit `E3Failed { failed_at_stage: CiphertextReady, reason: DecryptionInvalidShares }`. | @@ -271,7 +271,7 @@ them are in the reference app (`examples/CRISP`), not the protocol. | 39 | **Mainnet and Sepolia P2P overlay collision** | Resolved | Stable 32-byte network IDs now scope Identify, gossip, Kademlia, and historical sync. Peers remain staged until exact Identify and capability checks pass. Gossip uses application validation and a versioned envelope bound to network, deployment, aggregate, event ID, and payload hash. Historical sync transfers only chain-bound aggregates allowed by the active network policy and validates every event before EventBus admission; local aggregate 0 cannot enter peer fetch or recovery. The initial unscoped protocol is not offered as a fallback because it cannot distinguish networks. Profile defaults use separate mainnet and Sepolia DNS bootstrap aliases; no retired bootstrap address is compiled into the node. Unreachable peers are evicted and temporarily quarantined, and unavailable bootstrap peers use quiet background retries. Legacy events found in the wrong aggregate store are ignored without deleting operator data. | | 40 | **Fail-closed P2P admission lifecycle** | Resolved | Production network policies require at least one deployment and cannot infer unrestricted access from an empty map. Identify admission retains every simultaneous connection result. Compatibility failures are permanent, but Identify timeouts use a short transient cooldown so bootstrap retries can recover. Each gossip publish has a bounded result timeout; no-peer failures use the longer network-join retry window, while invalid payloads fail without retry. Historical-sync validation failures return a terminal protocol error instead of leaving the requester to time out. | | 41 | **Canonical completion and replay-safe result publication** | Resolved | `PlaintextAggregated` is a local durable publication intent, not a gossip artifact or a terminal lifecycle fact. Peer events cannot create unknown E3 contexts. Only confirmed EVM completion drives `E3RequestComplete`. Interfold and registry writer gates rebuild local result intents during replay, defer them until `EffectsEnabled`, start submissions only under the active aggregator role, coalesce by E3, retain retryable failures, and use contract preflights for cross-restart idempotency. Request cleanup cannot erase unfinished publication. Public RPC ingestion waits one block automatically; loopback development RPCs read the head, with no operator setting. | -| 42 | **Restart-safe active-aggregator failover** | Resolved | Every committee member persists validated public-key and decryption-share inputs in its normal aggregation state. `AggregationInputsReady` starts the durable 10-minute budget only after that node can resume the phase. Only the active party launches effects; promotion redrives the persisted phase, while demotion rejects late worker results. Replay preserves an unchanged ready deadline, processes overdue work after `EffectsEnabled`, and clears phase-local skips on canonical progress. A pre-readiness snapshot migrates by clearing its false timers and skip set. Selected nodes also rebuild canonical DKG/decryption deadline watches and stagger `markE3Failed` attempts by party ID. | +| 42 | **Restart-safe active-aggregator failover** | Resolved | Every committee member persists validated public-key and decryption-share inputs in its normal aggregation state. `AggregationInputsReady` starts the durable 10-minute budget only after that node can resume the phase. Only the active party launches effects; promotion redrives the persisted phase, while demotion rejects late worker results. Replay preserves an unchanged ready deadline, processes overdue work after `EffectsEnabled`, and clears phase-local skips on canonical progress. A pre-readiness snapshot migrates by clearing its false timers and skip set. Nodes rebuild every on-chain failure watch from durable lifecycle and request-time registry state. Selected members stagger by party ID; nodes without a party ID wait for the permissionless grace period. | | 43 | **Crash-safe per-E3 effect redrive** | Resolved | EventStore replay keeps durable sequence order inside each aggregate and uses HLC order only between aggregate heads. The canonical request-router checkpoint keeps the highest sequence seen on live and recovery paths. A trailing checkpoint advances from only its missing EventStore suffix, preserving active E3 admission state. Shutdown drains open cross-aggregate snapshot batches in event order, so an older batch cannot overwrite newer state. Versioned threshold-keyshare and aggregation records retain collector inputs, proof jobs, verified bundles, final publication intents, and causal contexts. `EffectsEnabled` creates new worker correlations and re-publishes determined outputs idempotently. Missing required recovery data fails startup instead of leaving an actor waiting on process-local state. | | 44 | **Empty-network restart completion** | Resolved | Process infrastructure events are excluded from EventStore replay, and the current `NetReady` listener is armed before the transport starts. A prior empty `HistoricalNetSyncEventsReceived` cannot occupy the payload-derived EventBus dedup key and suppress the fresh completion awaited by the current boot. A node with no active E3s or eligible peer-sync aggregates now advances through `SyncEffect`, publishes `SyncEnded`, and reaches live mode. | | 45 | **Snapshot-derived local participation recovery** | Resolved | Startup reconciles selector and finalized-committee snapshots, prunes terminal E3s, fails on contradictory committees or missing metadata, and directly seeds aggregator roles, party IDs, verifier caches, and DHT interests. The router applies each derived local selection at the fenced `SyncEffect` boundary after effects attach. Recovery does not append synthetic `CiphernodeSelected` or `AggregatorChanged` events on every boot. | diff --git a/agent/flow-trace/04_DKG_AND_COMPUTATION.md b/agent/flow-trace/04_DKG_AND_COMPUTATION.md index 44ea99f060..9a5aa7f475 100644 --- a/agent/flow-trace/04_DKG_AND_COMPUTATION.md +++ b/agent/flow-trace/04_DKG_AND_COMPUTATION.md @@ -693,7 +693,10 @@ phase. │ the step; a different commitment stays an error └─ Calls contract.publishCommitteePublicKey(e3_id, publicKey) after the commitment is available, including after restart - → A terminal result clears the intent; a retryable failure keeps it and retries after 30s + → A terminal result clears the in-memory intent; a retryable failure keeps it and retries + after 30s + → RPC request-size rejection and permanent contract or payload errors are terminal for the + running writer. They produce one final error instead of an unbounded 30-second retry loop → A restart replays the intent, so an unfinished publication still reaches the chain. E3RequestComplete that arrives before EffectsEnabled comes from that same replay and drops the intent: a completed request published its candidate in an earlier run, and @@ -1339,6 +1342,27 @@ publication, and only an active aggregator can start a retained submission. `Pla not gossiped or returned by historical peer sync; only the producing node can create this EVM write intent. +The CRISP server writes its request record at `E3Requested` and writes the generic E3 record only +after the indexer verifies the committee public key against the on-chain commitment. Current-round +lookup uses the request record, so a round remains visible while its key is pending. CRISP activates +the round only when both records exist. Either handler can complete the activation after their +records converge, and deferred checks cover slow live-handler ordering. Duplicate request and +committee events do not reset the round, replace indexed output, or resubmit an already-matching +Merkle root. Startup rebuilds deadline callbacks for active and expired rounds and releases an +interrupted compute submission for retry. The compute transition is atomic, and a synchronous +program-server request error releases the claim to `Expired` so a later deadline callback can retry +it. Compute submission is at-least-once across a restart because the HTTP response or webhook can +be lost. A retry can repeat proof work, but it cannot publish a second result: `Interfold` accepts +ciphertext output only from `KeyPublished`, and the callback treats an E3 that already reached +`CiphertextReady` or `Complete` as success. + +Operator constraint: the Sepolia contract byte limit does not bypass an RPC transaction-size +limit. The observed secure-8192 public key transaction is rejected before Solidity executes. Until +a separate node, client, and indexer release provides a verifiable transport that fits the RPC +path, use the current small-key parameters only as a Sepolia E2E plumbing workaround. That +workaround does not validate secure-8192 and is not a mainnet security substitute. Do not interpret +`KeyPublished` as proof that CRISP has usable key bytes. + ### What the compute-provider crate guarantees, and what an E3 program decides `e3-compute-provider` is shared by every E3 program, so it holds only what is true for all of them: diff --git a/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md b/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md index d2ef27d8cc..d94340245a 100644 --- a/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md +++ b/agent/flow-trace/05_FAILURE_REFUND_SLASHING.md @@ -26,12 +26,15 @@ actually slashed and does not require an oracle or relabel one ERC-20 as another Anyone can call `markE3Failed()` when a deadline is missed. A ready committee remains finalizable through its absolute DKG deadline. It can fail if it remains unfinalized after that deadline. -For the aggregator-owned DKG and decryption stages, each selected ciphernode reconstructs a -canonical deadline watch from `CiphernodeSelected` and `E3StageChanged` during replay. After -`EffectsEnabled`, it reads the deadline from `Interfold`, staggers its attempt by canonical party -ID, confirms that the stage and failure condition still match, and calls `markE3Failed`. A canonical -stage change cancels the old watch. If a node restarts after the deadline, the party-ID stagger is -applied from restart time so all committee wallets do not submit at once. +The Interfold writer watches every stage that `failureCondition` supports: `Requested`, +`CommitteeFinalized`, `KeyPublished`, and `CiphertextReady`. Startup restores the stage from the +durable lifecycle map and restores the request-time registry from the DKG context. A finalized +committee member staggers its attempt by canonical party ID. A `Requested` E3 has no active +committee yet, so every node waits until the failure grace period ends and uses the permissionless +path. Before submission, the writer confirms that the stage and failure condition still match. A +canonical stage change cancels the old watch and invalidates any older stage-discovery RPC. After a +restart, finalized members keep their party-ID stagger and non-members remain outside the protected +grace window. If an honest-node allocation is smaller than the node count, the refund manager credits it to the request-time treasury instead of creating zero-value claims. @@ -149,7 +152,9 @@ charge the protocol-funded subscription. 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. A restart restores failed E3 IDs from the durable lifecycle map. The +writer retries transient failures and treats `NoPaymentToRefund` as proof that another account has +already processed the escrow. ```text Anyone calls: Interfold.processE3Failure(e3Id) diff --git a/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md b/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md index 2284d52e74..25231bbf42 100644 --- a/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md +++ b/agent/flow-trace/06_DEACTIVATION_AND_COMPLETION.md @@ -490,8 +490,9 @@ publish that exclusion leaves the intent retryable. The registry writer rebuilds ticket, committee-finalization, and public-key submission gates from durable local events. It does not submit during replay. After `EffectsEnabled`, it retries temporary RPC or contract-ordering failures, treats already-landed transactions as success, and stops retrying -a ticket after a permanent eligibility or deadline result. The Interfold writer applies the same -pattern to plaintext publication. +a ticket after a permanent eligibility or deadline result. It also stops a public-key submission +after an RPC request-size rejection or a permanent payload or contract error. The Interfold writer +applies the same pattern to plaintext publication. The request router uses one checkpoint at `//router/recovery_checkpoint` for its active contexts, completed set, and all aggregate cursors. Per-E3 context snapshots remain below their own router diff --git a/crates/ciphernode-builder/src/ciphernode_builder.rs b/crates/ciphernode-builder/src/ciphernode_builder.rs index a1f17e1c6f..b13afe92f7 100644 --- a/crates/ciphernode-builder/src/ciphernode_builder.rs +++ b/crates/ciphernode-builder/src/ciphernode_builder.rs @@ -26,8 +26,8 @@ use e3_crypto::Cipher; use e3_data::{InMemStore, RepositoriesFactory}; use e3_events::DkgFoldAttestationContext; use e3_events::{ - AggregateConfig, AggregateId, BusHandle, E3id, EventBus, EventBusConfig, EventSubscriber, - EventType, EvmEventConfig, InterfoldEvent, + AggregateConfig, AggregateId, BusHandle, E3Stage, E3id, EventBus, EventBusConfig, + EventSubscriber, EventType, EvmEventConfig, InterfoldEvent, }; use e3_evm::{ ensure_node_release, fetch_accusation_vote_validity, fetch_randomness_providers, @@ -73,6 +73,7 @@ struct EvmStartupRecovery<'a> { dkg_fold_contexts_by_e3: &'a HashMap, active_aggregators: &'a HashMap, selected_party_ids: &'a HashMap, + lifecycle_stages: &'a HashMap, committee_finalizer: &'a CommitteeFinalizerRecoveryState, } @@ -597,6 +598,11 @@ impl CiphernodeBuilder { self.contract_components.slashing_manager, ) .await?; + let lifecycle_stages = repositories + .e3_lifecycle() + .read() + .await? + .unwrap_or_default(); let dkg_fold_contexts_by_e3 = load_dkg_fold_attestation_contexts(&repositories).await?; let mut provider_cache = @@ -673,6 +679,7 @@ impl CiphernodeBuilder { dkg_fold_contexts_by_e3: &dkg_fold_contexts_by_e3, active_aggregators: &selector_state.is_aggregator, selected_party_ids: &selected_party_ids, + lifecycle_stages: &lifecycle_stages, committee_finalizer: &committee_finalizer_recovery, }, ) @@ -1225,6 +1232,7 @@ async fn setup_evm_system( dkg_fold_contexts_by_e3, active_aggregators, selected_party_ids, + lifecycle_stages, committee_finalizer, } = recovery; let mut evm_config = EvmEventConfig::new(); @@ -1261,12 +1269,39 @@ async fn setup_evm_system( .filter(|(e3_id, _)| e3_id.chain_id() == chain_id) .map(|(e3_id, party_id)| (e3_id.clone(), *party_id)) .collect(); + let chain_request_registries = dkg_fold_contexts_by_e3 + .iter() + .filter(|(e3_id, _)| e3_id.chain_id() == chain_id) + .map(|(e3_id, context)| (e3_id.clone(), context.registry)) + .collect(); + let chain_failure_stages = lifecycle_stages + .iter() + .filter(|(e3_id, stage)| { + e3_id.chain_id() == chain_id + && matches!( + stage, + E3Stage::Requested + | E3Stage::CommitteeFinalized + | E3Stage::KeyPublished + | E3Stage::CiphertextReady + ) + }) + .map(|(e3_id, stage)| (e3_id.clone(), stage.clone())) + .collect(); + let chain_failure_settlements = lifecycle_stages + .iter() + .filter(|(e3_id, stage)| e3_id.chain_id() == chain_id && **stage == E3Stage::Failed) + .map(|(e3_id, _)| e3_id.clone()) + .collect(); InterfoldSolWriter::attach_with_recovery( bus, write_provider.clone(), contract.address()?, chain_active_aggregators, chain_party_ids, + chain_request_registries, + chain_failure_stages, + chain_failure_settlements, ); system.with_contract(contract.address()?, move |next| { InterfoldSolReader::setup(&next).recipient() diff --git a/crates/evm/src/ciphernode_registry/effects.rs b/crates/evm/src/ciphernode_registry/effects.rs index 3f67be7f53..bd456137a8 100644 --- a/crates/evm/src/ciphernode_registry/effects.rs +++ b/crates/evm/src/ciphernode_registry/effects.rs @@ -3,6 +3,7 @@ //! Idempotency preflights and CiphernodeRegistry contract effects. use super::*; +use crate::contracts::IInterfold; const TICKET_GAS_SAFETY_MULTIPLIER: u64 = 2; @@ -83,6 +84,46 @@ pub(in crate::actors::ciphernode_registry_sol) fn ticket_submission_error_is_ter .any(|selector| reverts_with(error, selector)) } +/// Return true when another committee-publication attempt cannot succeed unchanged. +pub(in crate::actors::ciphernode_registry_sol) fn committee_publication_error_is_terminal( + error: &anyhow::Error, +) -> bool { + let encoded = format!("{error:?}"); + let message = encoded.to_ascii_lowercase(); + let permanent_rpc_rejection = message.contains("oversized data") + || (message.contains("transaction size") && message.contains("limit")) + || message.contains("request entity too large") + || message.contains("content length too large") + || message.contains("function selector was not recognized"); + let permanent_local_rejection = message + .contains("mandatory dkg aggregator proof payload missing") + || message.contains("mandatory dkg attestation bundle missing") + || (message.contains("on-chain committee commitment") + && message.contains("does not match local commitment")); + let permanent_contract_rejection = [ + ICiphernodeRegistry::InvalidPublicKeyLength::SELECTOR, + ICiphernodeRegistry::PkCommitmentRequired::SELECTOR, + ICiphernodeRegistry::DkgProofRequired::SELECTOR, + ICiphernodeRegistry::InvalidDkgProof::SELECTOR, + ICiphernodeRegistry::FoldAttestationsRequired::SELECTOR, + ICiphernodeRegistry::FoldAttestationVerifierNotSet::SELECTOR, + ICiphernodeRegistry::InvalidFoldAttestation::SELECTOR, + ICiphernodeRegistry::PartyIdNotInProof::SELECTOR, + ICiphernodeRegistry::AttestationBindingCountMismatch::SELECTOR, + ICiphernodeRegistry::PartyIdOutOfBounds::SELECTOR, + ICiphernodeRegistry::InvalidProof::SELECTOR, + ICiphernodeRegistry::InvalidPublicInputsLength::SELECTOR, + ICiphernodeRegistry::VkHashMismatch::SELECTOR, + ICiphernodeRegistry::PkCommitmentMismatch::SELECTOR, + ICiphernodeRegistry::DomainBindingMismatch::SELECTOR, + IInterfold::DKGDeadlinePassed::SELECTOR, + ] + .into_iter() + .any(|selector| contains_error_selector(&encoded, selector)); + + permanent_rpc_rejection || permanent_local_rejection || permanent_contract_rejection +} + /// Report whether this node's ticket is already recorded on chain. /// /// `submitTicket` reverts with `NodeAlreadySubmitted` for a sender that is @@ -412,8 +453,11 @@ pub async fn fetch_randomness_providers( #[cfg(test)] mod tests { - use super::{reverts_with, ticket_gas_limit, ticket_submission_error_is_terminal}; - use crate::contracts::ICiphernodeRegistry; + use super::{ + committee_publication_error_is_terminal, reverts_with, ticket_gas_limit, + ticket_submission_error_is_terminal, + }; + use crate::contracts::{ICiphernodeRegistry, IInterfold}; use alloy::sol_types::{Revert, SolError}; fn selector_error(selector: [u8; 4]) -> anyhow::Error { @@ -462,4 +506,31 @@ mod tests { assert_eq!(ticket_gas_limit(250_000), 500_000); assert_eq!(ticket_gas_limit(u64::MAX), u64::MAX); } + + #[test] + fn oversized_rpc_rejection_is_terminal() { + let error = anyhow::anyhow!( + "server returned error code -32000: oversized data: transaction size 356602, limit 131072" + ); + assert!(committee_publication_error_is_terminal(&error)); + assert!(!committee_publication_error_is_terminal(&anyhow::anyhow!( + "RPC connection reset" + ))); + assert!(!committee_publication_error_is_terminal(&selector_error( + ICiphernodeRegistry::CommitteeNotPublished::SELECTOR + ))); + } + + #[test] + fn invalid_public_key_length_is_terminal() { + assert!(committee_publication_error_is_terminal(&selector_error( + ICiphernodeRegistry::InvalidPublicKeyLength::SELECTOR + ))); + assert!(committee_publication_error_is_terminal(&selector_error( + ICiphernodeRegistry::InvalidProof::SELECTOR + ))); + assert!(committee_publication_error_is_terminal(&selector_error( + IInterfold::DKGDeadlinePassed::SELECTOR + ))); + } } diff --git a/crates/evm/src/ciphernode_registry/handlers.rs b/crates/evm/src/ciphernode_registry/handlers.rs index d177cbb038..54e2d943e3 100644 --- a/crates/evm/src/ciphernode_registry/handlers.rs +++ b/crates/evm/src/ciphernode_registry/handlers.rs @@ -562,11 +562,16 @@ impl Handler false } Err(err) => { + let terminal = committee_publication_error_is_terminal(&err); error!( "Failed to preflight publishCommittee: {}", format_evm_error(&err) ); - return (e3_id, false); + if terminal { + error!(e3_id = %e3_id, "Committee publication failed permanently; stopping retries"); + } + bus.err(EType::Evm, err); + return (e3_id, terminal); } Ok(true) => true, }; @@ -607,12 +612,16 @@ impl Handler match result { Ok(()) => (e3_id, true), Err(err) => { + let terminal = committee_publication_error_is_terminal(&err); error!( "Failed to publish committee data: {}", format_evm_error(&err) ); + if terminal { + error!(e3_id = %e3_id, "Committee publication failed permanently; stopping retries"); + } bus.err(EType::Evm, err); - (e3_id, false) + (e3_id, terminal) } } } diff --git a/crates/evm/src/contracts.rs b/crates/evm/src/contracts.rs index 5b87730ca4..65d1852fd3 100644 --- a/crates/evm/src/contracts.rs +++ b/crates/evm/src/contracts.rs @@ -45,6 +45,12 @@ sol! { uint256 decryptionDeadline; } + struct E3TimeoutConfig { + uint256 dkgWindow; + uint256 computeWindow; + uint256 decryptionWindow; + } + // ── Write functions ───────────────────────────────────────────────── function publishPlaintextOutput( uint256 e3Id, @@ -63,10 +69,16 @@ sol! { function getDeadlines(uint256 e3Id) external view returns (E3Deadlines memory deadlines); + function getE3TimeoutConfig( + uint256 e3Id + ) external view returns (E3TimeoutConfig memory config); + function checkFailureCondition( uint256 e3Id ) external view returns (bool canFail, uint8 reason); + function markFailedGracePeriod() external view returns (uint256); + function nodeReleaseRegistry() external view returns (address); function bondingRegistry() external view returns (address); function ciphernodeRegistry() external view returns (address); @@ -95,6 +107,7 @@ sol! { error E3AlreadyFailed(uint256 e3Id); error E3AlreadyComplete(uint256 e3Id); error MarkE3FailedInGracePeriod(uint256 e3Id, uint256 gracePeriodEnds); + error DKGDeadlinePassed(uint256 e3Id, uint256 deadline); } } @@ -208,6 +221,10 @@ sol! { // ── View functions ────────────────────────────────────────────────── function isOpen(uint256 e3Id) external view returns (bool); + function committeeThresholdMet(uint256 e3Id) external view returns (bool); + + function getCommitteeDeadline(uint256 e3Id) external view returns (uint256); + function committeePublicKey(uint256 e3Id) external view returns (bytes32 publicKeyHash); function getDkgAnchors( @@ -368,6 +385,17 @@ sol! { error DkgProofRequired(); error InvalidDkgProof(); error FoldAttestationsRequired(); + error FoldAttestationVerifierNotSet(); + error InvalidFoldAttestation(); + error PartyIdNotInProof(); + error AttestationBindingCountMismatch(); + error PartyIdOutOfBounds(uint256 partyId, uint256 committeeSize); + error InvalidProof(); + error InvalidPublicInputsLength(); + error VkHashMismatch(); + error PkCommitmentMismatch(); + error DomainBindingMismatch(); + error InvalidPublicKeyLength(uint256 supplied, uint256 maximum); } } diff --git a/crates/evm/src/interfold_writing/actor.rs b/crates/evm/src/interfold_writing/actor.rs index 7e7a1e596a..3a2a2526e2 100644 --- a/crates/evm/src/interfold_writing/actor.rs +++ b/crates/evm/src/interfold_writing/actor.rs @@ -6,10 +6,12 @@ //! Interfold contract publication boundary. -use crate::contracts::IInterfold; -use crate::domain::error_decoder::format_evm_error; -use crate::domain::plaintext_publication::failure_watch_delay; +use crate::contracts::{ICiphernodeRegistry, IInterfold}; +use crate::domain::error_decoder::{contains_error_selector, format_evm_error}; use crate::domain::plaintext_publication::validate_plaintext_output; +use crate::domain::plaintext_publication::{ + failure_watch_delay, failure_watch_party_id, FailureStageDiscoveryGate, +}; use crate::domain::publication_replay::ReplaySubmissionGate; use crate::helpers::{encode_zk_proof, transaction_nonce_guard, EthProvider}; use crate::send_tx_with_retry; @@ -21,12 +23,13 @@ use alloy::{ }; use anyhow::Result; use e3_events::{ - prelude::*, AggregatorChanged, BusHandle, CiphernodeSelected, E3RequestComplete, E3Stage, - E3StageChanged, E3id, EType, EffectsEnabled, EventType, InterfoldEvent, InterfoldEventData, - PlaintextAggregated, Proof, Shutdown, + prelude::*, AggregatorChanged, BusHandle, CiphernodeSelected, + DkgFoldAttestationContextEstablished, E3RequestComplete, E3Stage, E3StageChanged, E3id, EType, + EffectsEnabled, EventType, InterfoldEvent, InterfoldEventData, PlaintextAggregated, Proof, + Shutdown, DKG_FOLD_ATTESTATION_CONTEXT_SCHEMA_VERSION, }; use e3_utils::{require_successful_receipt, NotifySync, MAILBOX_LIMIT}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tracing::info; @@ -44,8 +47,11 @@ pub struct InterfoldSolWriter

{ active_aggregators: HashMap, publication: ReplaySubmissionGate, committee_party_ids: HashMap, + request_registries: HashMap, failure_stages: HashMap, failure_timers: HashMap, + failure_stage_discoveries: FailureStageDiscoveryGate, + pending_failure_settlements: HashSet, } impl InterfoldSolWriter

{ @@ -60,6 +66,9 @@ impl InterfoldSolWriter

{ contract_address, HashMap::new(), HashMap::new(), + HashMap::new(), + HashMap::new(), + HashSet::new(), ) } @@ -69,6 +78,9 @@ impl InterfoldSolWriter

{ contract_address: Address, active_aggregators: HashMap, committee_party_ids: HashMap, + request_registries: HashMap, + failure_stages: HashMap, + pending_failure_settlements: HashSet, ) -> Result { Ok(Self { provider, @@ -78,8 +90,11 @@ impl InterfoldSolWriter

{ active_aggregators, publication: ReplaySubmissionGate::new(), committee_party_ids, - failure_stages: HashMap::new(), + request_registries, + failure_stages, failure_timers: HashMap::new(), + failure_stage_discoveries: FailureStageDiscoveryGate::default(), + pending_failure_settlements, }) } @@ -90,6 +105,9 @@ impl InterfoldSolWriter

{ contract_address, HashMap::new(), HashMap::new(), + HashMap::new(), + HashMap::new(), + HashSet::new(), ); } @@ -99,6 +117,9 @@ impl InterfoldSolWriter

{ contract_address: Address, active_aggregators: HashMap, committee_party_ids: HashMap, + request_registries: HashMap, + failure_stages: HashMap, + pending_failure_settlements: HashSet, ) { let addr = InterfoldSolWriter::new_with_recovery( bus, @@ -106,6 +127,9 @@ impl InterfoldSolWriter

{ contract_address, active_aggregators, committee_party_ids, + request_registries, + failure_stages, + pending_failure_settlements, ) .expect("failed to create InterfoldSolWriter") .start(); @@ -114,6 +138,7 @@ impl InterfoldSolWriter

{ EventType::EffectsEnabled, EventType::AggregatorChanged, EventType::CiphernodeSelected, + EventType::DkgFoldAttestationContextEstablished, EventType::PlaintextAggregated, EventType::E3StageChanged, EventType::E3RequestComplete, @@ -155,6 +180,12 @@ struct MarkFailedAtDeadline { stage: E3Stage, } +#[derive(Message, Debug, Clone)] +#[rtype(result = "()")] +struct ProcessFailedE3 { + e3_id: E3id, +} + impl Actor for InterfoldSolWriter

{ type Context = actix::Context; diff --git a/crates/evm/src/interfold_writing/effects.rs b/crates/evm/src/interfold_writing/effects.rs index dd1a4b694b..a3089c2d58 100644 --- a/crates/evm/src/interfold_writing/effects.rs +++ b/crates/evm/src/interfold_writing/effects.rs @@ -3,6 +3,7 @@ //! Interfold contract reads and transaction effects. use super::*; +use alloy::sol_types::SolError; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(in crate::actors::interfold_sol_writer) enum MarkFailureOutcome { @@ -11,7 +12,27 @@ pub(in crate::actors::interfold_sol_writer) enum MarkFailureOutcome { StageAdvanced, } -pub(in crate::actors::interfold_sol_writer) async fn read_aggregation_failure_stage< +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(in crate::actors::interfold_sol_writer) struct FailureSchedule { + pub deadline: u64, + pub permissionless_grace: u64, +} + +fn requested_failure_deadline( + committee_deadline: u64, + committee_threshold_met: bool, + dkg_window: u64, +) -> Result { + if committee_threshold_met { + committee_deadline + .checked_add(dkg_window) + .ok_or_else(|| anyhow::anyhow!("Requested-stage deadline overflowed")) + } else { + Ok(committee_deadline) + } +} + +pub(in crate::actors::interfold_sol_writer) async fn read_watched_failure_stage< P: Provider + WalletProvider + Clone, >( provider: EthProvider

, @@ -22,7 +43,9 @@ pub(in crate::actors::interfold_sol_writer) async fn read_aggregation_failure_st let contract = IInterfold::new(contract_address, provider.provider()); let stage = contract.getE3Stage(e3_id).call().await?; Ok(match stage { + 1 => Some(E3Stage::Requested), 2 => Some(E3Stage::CommitteeFinalized), + 3 => Some(E3Stage::KeyPublished), 4 => Some(E3Stage::CiphertextReady), _ => None, }) @@ -35,18 +58,61 @@ pub(in crate::actors::interfold_sol_writer) async fn read_failure_deadline< contract_address: Address, e3_id: E3id, stage: E3Stage, -) -> Result { + request_registry: Option

, +) -> Result { let e3_id: U256 = e3_id.try_into()?; let contract = IInterfold::new(contract_address, provider.provider()); - let deadlines = contract.getDeadlines(e3_id).call().await?; - let deadline = match stage { - E3Stage::CommitteeFinalized => deadlines.dkgDeadline, - E3Stage::CiphertextReady => deadlines.decryptionDeadline, - _ => anyhow::bail!("stage {stage:?} does not have an aggregation failure deadline"), + let deadline: u64 = match stage { + E3Stage::Requested => { + let registry_address = request_registry.ok_or_else(|| { + anyhow::anyhow!("request-time registry is unavailable for Requested E3") + })?; + let registry = ICiphernodeRegistry::new(registry_address, provider.provider()); + let committee_deadline: u64 = registry + .getCommitteeDeadline(e3_id) + .call() + .await? + .try_into() + .map_err(|_| anyhow::anyhow!("committee deadline does not fit in u64"))?; + let committee_threshold_met = registry.committeeThresholdMet(e3_id).call().await?; + let dkg_window = if committee_threshold_met { + let dkg_window: u64 = contract + .getE3TimeoutConfig(e3_id) + .call() + .await? + .dkgWindow + .try_into() + .map_err(|_| anyhow::anyhow!("DKG window does not fit in u64"))?; + dkg_window + } else { + 0 + }; + requested_failure_deadline(committee_deadline, committee_threshold_met, dkg_window)? + } + E3Stage::CommitteeFinalized | E3Stage::KeyPublished | E3Stage::CiphertextReady => { + let deadlines = contract.getDeadlines(e3_id).call().await?; + let deadline = match stage { + E3Stage::CommitteeFinalized => deadlines.dkgDeadline, + E3Stage::KeyPublished => deadlines.computeDeadline, + E3Stage::CiphertextReady => deadlines.decryptionDeadline, + _ => unreachable!(), + }; + deadline + .try_into() + .map_err(|_| anyhow::anyhow!("E3 deadline does not fit in u64"))? + } + _ => anyhow::bail!("stage {stage:?} does not have a failure deadline"), }; - deadline + let permissionless_grace = contract + .markFailedGracePeriod() + .call() + .await? .try_into() - .map_err(|_| anyhow::anyhow!("E3 deadline does not fit in u64")) + .map_err(|_| anyhow::anyhow!("mark-failed grace period does not fit in u64"))?; + Ok(FailureSchedule { + deadline, + permissionless_grace, + }) } pub(in crate::actors::interfold_sol_writer) async fn mark_e3_failed_if_due< @@ -88,9 +154,11 @@ pub(in crate::actors::interfold_sol_writer) async fn mark_e3_failed_if_due< fn failure_stage_code(stage: &E3Stage) -> Result { match stage { + E3Stage::Requested => Ok(1), E3Stage::CommitteeFinalized => Ok(2), + E3Stage::KeyPublished => Ok(3), E3Stage::CiphertextReady => Ok(4), - _ => anyhow::bail!("stage {stage:?} is not watched for aggregation failure"), + _ => anyhow::bail!("stage {stage:?} is not watched for failure"), } } @@ -181,3 +249,50 @@ 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) fn failure_settlement_error_is_terminal( + error: &anyhow::Error, +) -> bool { + contains_error_selector( + &format!("{error:?}"), + IInterfold::NoPaymentToRefund::SELECTOR, + ) +} + +#[cfg(test)] +mod tests { + use super::{ + failure_settlement_error_is_terminal, failure_stage_code, requested_failure_deadline, + }; + use crate::contracts::IInterfold; + use alloy::sol_types::SolError; + use e3_events::E3Stage; + + #[test] + fn all_contract_failure_stages_are_watched() { + assert_eq!(failure_stage_code(&E3Stage::Requested).unwrap(), 1); + assert_eq!(failure_stage_code(&E3Stage::CommitteeFinalized).unwrap(), 2); + assert_eq!(failure_stage_code(&E3Stage::KeyPublished).unwrap(), 3); + assert_eq!(failure_stage_code(&E3Stage::CiphertextReady).unwrap(), 4); + assert!(failure_stage_code(&E3Stage::Complete).is_err()); + } + + #[test] + fn requested_stage_uses_the_registry_deadline_and_frozen_dkg_window() { + assert_eq!(requested_failure_deadline(100, false, 50).unwrap(), 100); + assert_eq!(requested_failure_deadline(100, true, 50).unwrap(), 150); + assert!(requested_failure_deadline(u64::MAX, true, 1).is_err()); + } + + #[test] + fn settled_failure_stops_retries() { + let error = anyhow::anyhow!( + "execution reverted: 0x{}", + hex::encode(IInterfold::NoPaymentToRefund::SELECTOR) + ); + assert!(failure_settlement_error_is_terminal(&error)); + assert!(!failure_settlement_error_is_terminal(&anyhow::anyhow!( + "RPC connection reset" + ))); + } +} diff --git a/crates/evm/src/interfold_writing/handlers.rs b/crates/evm/src/interfold_writing/handlers.rs index 8b7d0d9643..ca1b5d9842 100644 --- a/crates/evm/src/interfold_writing/handlers.rs +++ b/crates/evm/src/interfold_writing/handlers.rs @@ -5,6 +5,7 @@ use super::effects::*; use super::*; use e3_events::EventSource; +use std::collections::HashSet; const PUBLICATION_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(30); const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(30); @@ -27,12 +28,15 @@ impl InterfoldSolWriter

{ } fn try_start_failure_watch(&self, e3_id: &E3id, ctx: &mut actix::Context) { - if !self.effects_enabled || !self.committee_party_ids.contains_key(e3_id) { + if !self.effects_enabled { return; } let Some(stage) = self.failure_stages.get(e3_id).cloned() else { return; }; + if stage == E3Stage::Requested && !self.request_registries.contains_key(e3_id) { + return; + } ctx.notify(ResolveFailureDeadline { e3_id: e3_id.clone(), stage, @@ -43,14 +47,24 @@ impl InterfoldSolWriter

{ for e3_id in self.failure_stages.keys() { self.try_start_failure_watch(e3_id, ctx); } - for e3_id in self.committee_party_ids.keys() { - ctx.notify(DiscoverFailureStage { + let discovery_ids = self + .committee_party_ids + .keys() + .chain(self.request_registries.keys()) + .cloned() + .collect::>(); + for e3_id in discovery_ids { + ctx.notify(DiscoverFailureStage { e3_id }); + } + for e3_id in &self.pending_failure_settlements { + ctx.notify(ProcessFailedE3 { e3_id: e3_id.clone(), }); } } fn clear_failure_watch(&mut self, e3_id: &E3id, ctx: &mut actix::Context) { + self.failure_stage_discoveries.invalidate(e3_id); self.failure_stages.remove(e3_id); if let Some(handle) = self.failure_timers.remove(e3_id) { ctx.cancel_future(handle); @@ -61,23 +75,23 @@ impl InterfoldSolWriter

{ &mut self, e3_id: E3id, stage: E3Stage, - deadline_unix_secs: u64, + schedule: FailureSchedule, ctx: &mut actix::Context, ) { if self.failure_stages.get(&e3_id) != Some(&stage) { return; } - let Some(party_id) = self.committee_party_ids.get(&e3_id).copied() else { - return; - }; if let Some(handle) = self.failure_timers.remove(&e3_id) { ctx.cancel_future(handle); } + let party_id = + failure_watch_party_id(&stage, self.committee_party_ids.get(&e3_id).copied()); let delay = failure_watch_delay( Self::now_unix_secs(), - deadline_unix_secs, + schedule.deadline, party_id, + schedule.permissionless_grace, FAILURE_PARTY_STAGGER_SECS, ); let timer_e3_id = e3_id.clone(); @@ -104,6 +118,11 @@ impl Handler InterfoldEventData::EffectsEnabled(data) => self.notify_sync(ctx, data), InterfoldEventData::AggregatorChanged(data) => self.notify_sync(ctx, data), InterfoldEventData::CiphernodeSelected(data) => self.notify_sync(ctx, data), + InterfoldEventData::DkgFoldAttestationContextEstablished(data) => { + if self.provider.chain_id() == data.e3_id.chain_id() { + ctx.notify(data); + } + } InterfoldEventData::PlaintextAggregated(data) => { // Only a locally computed result is a publication intent. Peer results are // inputs for protocol observers and must not cross the EVM write boundary. @@ -155,6 +174,37 @@ impl Handler } } +impl Handler + for InterfoldSolWriter

+{ + type Result = (); + + fn handle( + &mut self, + msg: DkgFoldAttestationContextEstablished, + ctx: &mut Self::Context, + ) -> Self::Result { + if msg.schema_version != DKG_FOLD_ATTESTATION_CONTEXT_SCHEMA_VERSION { + self.request_registries.remove(&msg.e3_id); + self.bus.err( + EType::Evm, + anyhow::anyhow!( + "unsupported DKG attestation context schema {} for E3 {}", + msg.schema_version, + msg.e3_id + ), + ); + return; + } + self.request_registries + .insert(msg.e3_id.clone(), msg.context.registry); + self.try_start_failure_watch(&msg.e3_id, ctx); + if self.effects_enabled { + ctx.notify(DiscoverFailureStage { e3_id: msg.e3_id }); + } + } +} + impl Handler for InterfoldSolWriter

{ @@ -178,6 +228,7 @@ impl Handler fn handle(&mut self, msg: E3RequestComplete, ctx: &mut Self::Context) -> Self::Result { self.active_aggregators.remove(&msg.e3_id); self.committee_party_ids.remove(&msg.e3_id); + self.request_registries.remove(&msg.e3_id); self.clear_failure_watch(&msg.e3_id, ctx); } } @@ -310,12 +361,16 @@ impl Handler for Inter impl Handler for InterfoldSolWriter

{ - type Result = ResponseFuture<()>; + type Result = (); fn handle(&mut self, msg: E3StageChanged, ctx: &mut Self::Context) -> Self::Result { let e3_id = msg.e3_id.clone(); + self.failure_stage_discoveries.invalidate(&e3_id); match &msg.new_stage { - E3Stage::CommitteeFinalized | E3Stage::CiphertextReady => { + E3Stage::Requested + | E3Stage::CommitteeFinalized + | E3Stage::KeyPublished + | E3Stage::CiphertextReady => { self.failure_stages .insert(e3_id.clone(), msg.new_stage.clone()); self.try_start_failure_watch(&e3_id, ctx); @@ -323,33 +378,53 @@ impl Handler _ => self.clear_failure_watch(&e3_id, ctx), } - if !self.effects_enabled || msg.new_stage != E3Stage::Failed { - return Box::pin(async {}); + if msg.new_stage == E3Stage::Failed { + self.pending_failure_settlements.insert(e3_id.clone()); + if self.effects_enabled { + ctx.notify(ProcessFailedE3 { e3_id }); + } } + } +} - Box::pin({ - let contract_address = self.contract_address; - let provider = self.provider.clone(); +impl Handler + for InterfoldSolWriter

+{ + type Result = ResponseActFuture; + + fn handle(&mut self, msg: ProcessFailedE3, _ctx: &mut Self::Context) -> Self::Result { + if !self.effects_enabled || !self.pending_failure_settlements.contains(&msg.e3_id) { + return Box::pin(async {}.into_actor(self)); + } + + let provider = self.provider.clone(); + let contract_address = self.contract_address; + let e3_id = msg.e3_id; + Box::pin( async move { let result = process_e3_failure(provider, contract_address, e3_id.clone()).await; - match result { - Ok(receipt) => { - info!( - tx=%receipt.transaction_hash, - e3_id = %e3_id, - "Called processE3Failure" - ); - } - Err(err) => { - info!( - e3_id = %e3_id, - "processE3Failure did not succeed (may already be processed): {}", - format_evm_error(&err) - ); - } - } + (e3_id, result) } - }) + .into_actor(self) + .map(|(e3_id, result), actor, ctx| match result { + Ok(receipt) => { + actor.pending_failure_settlements.remove(&e3_id); + info!( + tx = %receipt.transaction_hash, + e3_id = %e3_id, + "Called processE3Failure" + ); + } + Err(error) if failure_settlement_error_is_terminal(&error) => { + actor.pending_failure_settlements.remove(&e3_id); + info!(e3_id = %e3_id, "Failure settlement was already processed"); + } + Err(error) => { + actor.bus.err(EType::Evm, error); + ctx.notify_later(ProcessFailedE3 { e3_id }, FAILURE_RETRY_DELAY); + } + }), + ) } } @@ -359,15 +434,13 @@ impl Handler; fn handle(&mut self, msg: ResolveFailureDeadline, _ctx: &mut Self::Context) -> Self::Result { - if !self.effects_enabled - || self.failure_stages.get(&msg.e3_id) != Some(&msg.stage) - || !self.committee_party_ids.contains_key(&msg.e3_id) - { + if !self.effects_enabled || self.failure_stages.get(&msg.e3_id) != Some(&msg.stage) { return Box::pin(async {}.into_actor(self)); } let provider = self.provider.clone(); let contract_address = self.contract_address; + let request_registry = self.request_registries.get(&msg.e3_id).copied(); let request = msg.clone(); Box::pin( async move { @@ -376,19 +449,20 @@ impl Handler 0 => { - actor.arm_failure_timer(request.e3_id, request.stage, deadline, ctx); + Ok(schedule) if schedule.deadline > 0 => { + actor.arm_failure_timer(request.e3_id, request.stage, schedule, ctx); } Ok(_) => { actor.bus.err( EType::Evm, - anyhow::anyhow!("canonical aggregation deadline is zero"), + anyhow::anyhow!("canonical failure deadline is zero"), ); ctx.notify_later(request, FAILURE_RETRY_DELAY); } @@ -407,29 +481,35 @@ impl Handler; fn handle(&mut self, msg: DiscoverFailureStage, _ctx: &mut Self::Context) -> Self::Result { - if !self.effects_enabled || !self.committee_party_ids.contains_key(&msg.e3_id) { + if !self.effects_enabled { return Box::pin(async {}.into_actor(self)); } let provider = self.provider.clone(); let contract_address = self.contract_address; let e3_id = msg.e3_id; + let generation = self.failure_stage_discoveries.start(e3_id.clone()); Box::pin( async move { let result = - read_aggregation_failure_stage(provider, contract_address, e3_id.clone()).await; - (e3_id, result) + read_watched_failure_stage(provider, contract_address, e3_id.clone()).await; + (e3_id, generation, result) } .into_actor(self) - .map(|(e3_id, result), actor, ctx| match result { - Ok(Some(stage)) => { - actor.failure_stages.insert(e3_id.clone(), stage); - actor.try_start_failure_watch(&e3_id, ctx); + .map(|(e3_id, generation, result), actor, ctx| { + if !actor.failure_stage_discoveries.complete(&e3_id, generation) { + return; } - Ok(None) => actor.clear_failure_watch(&e3_id, ctx), - Err(error) => { - actor.bus.err(EType::Evm, error); - ctx.notify_later(DiscoverFailureStage { e3_id }, FAILURE_RETRY_DELAY); + match result { + Ok(Some(stage)) => { + actor.failure_stages.insert(e3_id.clone(), stage); + actor.try_start_failure_watch(&e3_id, ctx); + } + Ok(None) => actor.clear_failure_watch(&e3_id, ctx), + Err(error) => { + actor.bus.err(EType::Evm, error); + ctx.notify_later(DiscoverFailureStage { e3_id }, FAILURE_RETRY_DELAY); + } } }), ) diff --git a/crates/evm/src/interfold_writing/workflow.rs b/crates/evm/src/interfold_writing/workflow.rs index cb8899710d..836bc26640 100644 --- a/crates/evm/src/interfold_writing/workflow.rs +++ b/crates/evm/src/interfold_writing/workflow.rs @@ -10,8 +10,9 @@ //! invariants hold; rejecting a malformed result is safer than a partial //! on-chain write. -use e3_events::{E3id, Proof}; +use e3_events::{E3Stage, E3id, Proof}; use e3_utils::utility_types::ArcBytes; +use std::collections::HashMap; use std::time::Duration; #[cfg(test)] @@ -61,14 +62,65 @@ pub(crate) fn validate_plaintext_output( pub(crate) fn failure_watch_delay( now_unix_secs: u64, deadline_unix_secs: u64, - party_id: u64, + party_id: Option, + permissionless_grace_secs: u64, party_stagger_secs: u64, ) -> Duration { - let deadline_wait = deadline_unix_secs - .saturating_sub(now_unix_secs) - .saturating_add(1); - let stagger = party_id.saturating_mul(party_stagger_secs); - Duration::from_secs(deadline_wait.saturating_add(stagger)) + let delay = match party_id { + Some(party_id) => deadline_unix_secs + .saturating_sub(now_unix_secs) + .saturating_add(1) + .saturating_add(party_id.saturating_mul(party_stagger_secs)), + None => { + let permissionless_at = if permissionless_grace_secs == 0 { + deadline_unix_secs.saturating_add(1) + } else { + deadline_unix_secs.saturating_add(permissionless_grace_secs) + }; + permissionless_at.saturating_sub(now_unix_secs) + } + }; + Duration::from_secs(delay) +} + +/// Return the finalized committee party that can act during the protected grace window. +/// +/// A `Requested` E3 has only provisional ticket candidates. The registry does not consider those +/// candidates active committee members, so they must wait for permissionless failure marking. +pub(crate) fn failure_watch_party_id(stage: &E3Stage, party_id: Option) -> Option { + if *stage == E3Stage::Requested { + None + } else { + party_id + } +} + +/// Reject stage-discovery results that a newer lifecycle event has superseded. +#[derive(Debug, Default)] +pub(crate) struct FailureStageDiscoveryGate { + next_generation: u64, + active: HashMap, +} + +impl FailureStageDiscoveryGate { + pub(crate) fn start(&mut self, e3_id: E3id) -> u64 { + self.next_generation = self.next_generation.wrapping_add(1); + let generation = self.next_generation; + self.active.insert(e3_id, generation); + generation + } + + pub(crate) fn invalidate(&mut self, e3_id: &E3id) { + self.active.remove(e3_id); + } + + pub(crate) fn complete(&mut self, e3_id: &E3id, generation: u64) -> bool { + if self.active.get(e3_id) != Some(&generation) { + return false; + } + self.active.remove(e3_id); + true + } } #[cfg(test)] @@ -124,9 +176,41 @@ mod tests { #[test] fn failure_watch_stagger_survives_restart() { - assert_eq!(failure_watch_delay(100, 160, 0, 15).as_secs(), 61); - assert_eq!(failure_watch_delay(100, 160, 2, 15).as_secs(), 91); - assert_eq!(failure_watch_delay(200, 160, 0, 15).as_secs(), 1); - assert_eq!(failure_watch_delay(200, 160, 2, 15).as_secs(), 31); + assert_eq!(failure_watch_delay(100, 160, Some(0), 90, 15).as_secs(), 61); + assert_eq!(failure_watch_delay(100, 160, Some(2), 90, 15).as_secs(), 91); + assert_eq!(failure_watch_delay(200, 160, Some(0), 90, 15).as_secs(), 1); + assert_eq!(failure_watch_delay(200, 160, Some(2), 90, 15).as_secs(), 31); + } + + #[test] + fn requested_stage_failure_without_party_waits_for_permissionless_grace() { + assert_eq!(failure_watch_delay(100, 160, None, 90, 15).as_secs(), 150); + assert_eq!(failure_watch_delay(200, 160, None, 90, 15).as_secs(), 50); + assert_eq!(failure_watch_delay(250, 160, None, 90, 15).as_secs(), 0); + assert_eq!(failure_watch_delay(160, 160, None, 0, 15).as_secs(), 1); + } + + #[test] + fn requested_stage_ignores_provisional_party_id() { + assert_eq!(failure_watch_party_id(&E3Stage::Requested, Some(2)), None); + assert_eq!( + failure_watch_party_id(&E3Stage::CommitteeFinalized, Some(2)), + Some(2) + ); + } + + #[test] + fn stage_discovery_ignores_superseded_results() { + let e3_id = E3id::new("7", 1); + let mut gate = FailureStageDiscoveryGate::default(); + + let first = gate.start(e3_id.clone()); + let second = gate.start(e3_id.clone()); + assert!(!gate.complete(&e3_id, first)); + assert!(gate.complete(&e3_id, second)); + + let invalidated = gate.start(e3_id.clone()); + gate.invalidate(&e3_id); + assert!(!gate.complete(&e3_id, invalidated)); } } diff --git a/crates/indexer/src/indexer.rs b/crates/indexer/src/indexer.rs index a0952ecdc7..ec1c456bf9 100644 --- a/crates/indexer/src/indexer.rs +++ b/crates/indexer/src/indexer.rs @@ -327,6 +327,95 @@ impl InterfoldIndexer { } } +async fn store_committee_public_key( + event: CommitteePublished, + ctx: Arc>, +) -> Result { + let contract = ctx.contract(); + let db = ctx.store(); + let interfold_address = ctx.interfold_address(); + let e3_id = event.e3Id.to_string(); + + info!( + "CommitteePublished: id={}, public_key_len={}, proof_len={}", + event.e3Id, + event.publicKey.len(), + event.proof.len() + ); + + let e3 = contract.get_e3(event.e3Id).await?; + let params_preset = BfvPreset::from_on_chain_param_set(e3.paramSet).ok_or_else(|| { + eyre!( + "unsupported BFV parameter set {} for E3 {e3_id}", + e3.paramSet + ) + })?; + let e3_params = encode_bfv_params(&BfvParamSet::from(params_preset).build_arc()); + let crypto_config_id = keccak256( + ( + keccak256(b"fhe.rs:BFV"), + keccak256(&e3_params), + keccak256(b"interfold-bfv-v1"), + ) + .abi_encode(), + ); + let request_crypto_config_id = contract.get_e3_crypto_config_id(event.e3Id).await?; + if request_crypto_config_id != crypto_config_id { + return Err(eyre!( + "local circuit configuration does not match request-time config for E3 {e3_id}" + )); + } + if e3.encryptionSchemeId == keccak256("fhe.rs:BFV") { + let decoded_params = decode_bfv_params(&e3_params) + .map_err(|error| eyre!("invalid BFV parameters for E3 {e3_id}: {error}"))?; + if let Err(error) = validate_pk_commitment( + &event.publicKey, + event.pkCommitment.0, + decoded_params.degree(), + decoded_params.plaintext(), + decoded_params.moduli().to_vec(), + ) { + warn!("Ignoring an unbound committee public-key candidate for E3 {e3_id}: {error}"); + return Ok(false); + } + } + let seed = e3.seed.to_be_bytes(); + let request_block = u64_try_from(e3.requestBlock)?; + let input_window = [ + u64_try_from(e3.inputWindow[0])?, + u64_try_from(e3.inputWindow[1])?, + ]; + + let e3_obj = E3 { + chain_id: ctx.chain_id(), + ciphertext_inputs: vec![], + ciphertext_output: vec![], + ciphertext_commitment: vec![], + committee_public_key: event.publicKey.to_vec(), + committee_public_key_hash: event.pkCommitment.to_vec(), + custom_params: e3.customParams.to_vec(), + e3_params: e3_params.to_vec(), + interfold_address, + encryption_scheme_id: e3.encryptionSchemeId.to_vec(), + crypto_config_id: crypto_config_id.to_vec(), + id: e3_id.clone(), + plaintext_output: vec![], + request_block, + seed, + input_window, + committee_size: e3.committeeSize, + requester: e3.requester.to_string(), + }; + + let mut repo = E3Repository::new(db, &e3_id); + if repo.set_e3_if_absent(e3_obj).await? { + info!("E3 {} created and stored", e3_id); + } else { + info!("E3 {} already has a verified committee key", e3_id); + } + Ok(true) +} + impl InterfoldIndexer { pub async fn new( mut event_listener: EventListener, @@ -431,88 +520,7 @@ impl InterfoldIndexer { async fn register_committee_published(&mut self) -> Result<()> { self.add_event_handler(move |e: CommitteePublished, ctx| async move { - let contract = ctx.contract(); - let db = ctx.store(); - let interfold_address = ctx.interfold_address(); - let e3_id = e.e3Id.to_string(); - - info!( - "CommitteePublished: id={}, public_key_len={}, proof_len={}", - e.e3Id, - e.publicKey.len(), - e.proof.len() - ); - - let e3 = contract.get_e3(e.e3Id).await?; - let params_preset = - BfvPreset::from_on_chain_param_set(e3.paramSet).ok_or_else(|| { - eyre!( - "unsupported BFV parameter set {} for E3 {e3_id}", - e3.paramSet - ) - })?; - let e3_params = encode_bfv_params(&BfvParamSet::from(params_preset).build_arc()); - let crypto_config_id = keccak256( - ( - keccak256(b"fhe.rs:BFV"), - keccak256(&e3_params), - keccak256(b"interfold-bfv-v1"), - ) - .abi_encode(), - ); - let request_crypto_config_id = contract.get_e3_crypto_config_id(e.e3Id).await?; - if request_crypto_config_id != crypto_config_id { - return Err(eyre!( - "local circuit configuration does not match request-time config for E3 {e3_id}" - )); - } - if e3.encryptionSchemeId == keccak256("fhe.rs:BFV") { - let decoded_params = decode_bfv_params(&e3_params) - .map_err(|error| eyre!("invalid BFV parameters for E3 {e3_id}: {error}"))?; - validate_pk_commitment( - &e.publicKey, - e.pkCommitment.0, - decoded_params.degree(), - decoded_params.plaintext(), - decoded_params.moduli().to_vec(), - ) - .map_err(|error| { - eyre!("rejecting unbound CommitteePublished public key for E3 {e3_id}: {error}") - })?; - } - let seed = e3.seed.to_be_bytes(); - let request_block = u64_try_from(e3.requestBlock)?; - let input_window = [ - u64_try_from(e3.inputWindow[0])?, - u64_try_from(e3.inputWindow[1])?, - ]; - - let e3_obj = E3 { - chain_id: ctx.chain_id(), - ciphertext_inputs: vec![], - ciphertext_output: vec![], - ciphertext_commitment: vec![], - committee_public_key: e.publicKey.to_vec(), - committee_public_key_hash: e.pkCommitment.to_vec(), - custom_params: e3.customParams.to_vec(), - e3_params: e3_params.to_vec(), - interfold_address, - encryption_scheme_id: e3.encryptionSchemeId.to_vec(), - crypto_config_id: crypto_config_id.to_vec(), - id: e3_id.clone(), - plaintext_output: vec![], - request_block, - seed, - input_window, - committee_size: e3.committeeSize, - requester: e3.requester.to_string(), - }; - - let mut repo = E3Repository::new(db, &e3_id); - repo.set_e3(e3_obj).await?; - - info!("E3 {} created and stored", e3_id); - + store_committee_public_key(e, ctx).await?; Ok(()) }) .await; @@ -889,6 +897,15 @@ impl InterfoldIndexer { pub fn get_store(&self) -> SharedStore { self.ctx.store.clone() } + + /// Schedule a timestamp callback without requiring an event-handler context. + pub fn schedule_at(&self, timestamp: u64, callback: F) + where + F: Fn(u64, Arc>) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + self.ctx.do_later(timestamp, callback); + } } pub async fn get_e3( diff --git a/crates/indexer/src/models.rs b/crates/indexer/src/models.rs index 313aa213fd..b210dc1a46 100644 --- a/crates/indexer/src/models.rs +++ b/crates/indexer/src/models.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; // This correlates with the information from the contract // with an addition of a chain_id -#[derive(Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] pub struct E3 { pub chain_id: u64, pub ciphertext_inputs: Vec<(Vec, u64)>, diff --git a/crates/indexer/src/repo.rs b/crates/indexer/src/repo.rs index 890d967caf..a53d0737a2 100644 --- a/crates/indexer/src/repo.rs +++ b/crates/indexer/src/repo.rs @@ -6,6 +6,10 @@ use super::{models::E3, DataStore, SharedStore}; use eyre::Result; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; pub struct E3Repository { store: SharedStore, @@ -29,6 +33,24 @@ impl E3Repository { Ok(()) } + /// Store the initial E3 record without replacing indexed round data. + pub async fn set_e3_if_absent(&mut self, value: E3) -> Result { + let key = self.e3_key(); + let inserted = Arc::new(AtomicBool::new(false)); + let inserted_in_update = Arc::clone(&inserted); + self.store + .modify(&key, move |current: Option| match current { + Some(current) => Some(current), + None => { + inserted_in_update.store(true, Ordering::Relaxed); + Some(value.clone()) + } + }) + .await + .map_err(|e| eyre::eyre!("Could not store E3 at '{key}' due to error: {e}"))?; + Ok(inserted.load(Ordering::Relaxed)) + } + pub async fn get_e3(&self) -> Result { let key = self.e3_key(); let e3_crisp = self @@ -100,3 +122,49 @@ impl E3Repository { format!("_e3:{e3_id}") } } + +#[cfg(test)] +mod tests { + use super::E3Repository; + use crate::{models::E3, InMemoryStore, SharedStore}; + use e3_evm_helpers::contracts::CommitteeSize; + use std::sync::Arc; + use tokio::sync::RwLock; + + fn e3(public_key: u8, ciphertext_output: Vec) -> E3 { + E3 { + chain_id: 1, + ciphertext_inputs: vec![(vec![3], 0)], + ciphertext_output, + ciphertext_commitment: vec![4], + committee_public_key: vec![public_key], + committee_public_key_hash: vec![public_key; 32], + e3_params: vec![5], + custom_params: vec![6], + interfold_address: "0x0000000000000000000000000000000000000001".to_string(), + encryption_scheme_id: vec![7; 32], + crypto_config_id: vec![8; 32], + id: "12".to_string(), + plaintext_output: vec![9], + request_block: 10, + seed: [11; 32], + input_window: [12, 13], + committee_size: CommitteeSize::Minimum, + requester: "0x0000000000000000000000000000000000000002".to_string(), + } + } + + #[tokio::test] + async fn committee_replay_does_not_replace_indexed_round_state() { + let store = SharedStore::new(Arc::new(RwLock::new(InMemoryStore::new()))); + let mut repo = E3Repository::new(store, "12"); + + assert!(repo.set_e3_if_absent(e3(1, vec![2])).await.unwrap()); + assert!(!repo.set_e3_if_absent(e3(99, vec![])).await.unwrap()); + + let stored = repo.get_e3().await.unwrap(); + assert_eq!(stored.committee_public_key, vec![1]); + assert_eq!(stored.ciphertext_output, vec![2]); + assert_eq!(stored.plaintext_output, vec![9]); + } +} diff --git a/crates/sortition/src/sortition/node_registry.rs b/crates/sortition/src/sortition/node_registry.rs index 401f4139b1..4ca8143072 100644 --- a/crates/sortition/src/sortition/node_registry.rs +++ b/crates/sortition/src/sortition/node_registry.rs @@ -16,7 +16,7 @@ use alloy::primitives::U256; use e3_events::E3id; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{hash_map::Entry, HashMap}; use tracing::{info, warn}; #[derive(Clone, Debug, Serialize, Deserialize)] @@ -265,6 +265,9 @@ impl NodeRegistry { /// Record a published committee and increment active-job counters for each /// of its members. + /// + /// Event replay and duplicate publication candidates can report the same + /// committee more than once. Only the first report changes the counters. pub fn record_committee_published( store: &mut HashMap, e3_id: &E3id, @@ -274,7 +277,29 @@ impl NodeRegistry { let key = committee_key(e3_id); let chain_state = store.entry(chain_id).or_default(); - chain_state.e3_committees.insert(key, nodes.to_vec()); + match chain_state.e3_committees.entry(key) { + Entry::Vacant(entry) => { + entry.insert(nodes.to_vec()); + } + Entry::Occupied(entry) => { + if entry.get().as_slice() != nodes { + warn!( + chain_id, + e3_id = ?e3_id, + recorded_nodes = ?entry.get(), + replayed_nodes = ?nodes, + "Ignored a conflicting committee publication replay" + ); + } else { + info!( + chain_id, + e3_id = ?e3_id, + "Ignored a duplicate committee publication replay" + ); + } + return; + } + } for node_addr in nodes { let node = chain_state.nodes.entry(node_addr.clone()).or_default(); diff --git a/crates/sortition/src/sortition/node_registry_tests.rs b/crates/sortition/src/sortition/node_registry_tests.rs index c8e30e0af8..95415ea8f5 100644 --- a/crates/sortition/src/sortition/node_registry_tests.rs +++ b/crates/sortition/src/sortition/node_registry_tests.rs @@ -121,6 +121,33 @@ fn release_committee_jobs_is_idempotent() { assert_eq!(store[&1].nodes["0xabc"].active_jobs, 0); } +#[test] +fn duplicate_committee_publication_increments_jobs_once() { + let mut store = HashMap::new(); + let id = e3(1, "17"); + let nodes = ["0xabc".into(), "0xdef".into()]; + + NodeRegistry::record_committee_published(&mut store, &id, &nodes); + NodeRegistry::record_committee_published(&mut store, &id, &nodes); + + assert_eq!(store[&1].nodes["0xabc"].active_jobs, 1); + assert_eq!(store[&1].nodes["0xdef"].active_jobs, 1); + assert_eq!(store[&1].e3_committees.len(), 1); +} + +#[test] +fn conflicting_committee_replay_preserves_the_first_committee() { + let mut store = HashMap::new(); + let id = e3(1, "18"); + + NodeRegistry::record_committee_published(&mut store, &id, &["0xabc".into()]); + NodeRegistry::record_committee_published(&mut store, &id, &["0xdef".into()]); + + assert_eq!(store[&1].nodes["0xabc"].active_jobs, 1); + assert!(!store[&1].nodes.contains_key("0xdef")); + assert_eq!(store[&1].e3_committees[&committee_key(&id)], vec!["0xabc"]); +} + #[test] fn get_nodes_with_tickets_filters_inactive_and_empty() { let mut store = HashMap::new(); diff --git a/examples/CRISP/client/src/context/voteManagement/VoteManagement.context.tsx b/examples/CRISP/client/src/context/voteManagement/VoteManagement.context.tsx index eb69b8da90..a25e28c085 100644 --- a/examples/CRISP/client/src/context/voteManagement/VoteManagement.context.tsx +++ b/examples/CRISP/client/src/context/voteManagement/VoteManagement.context.tsx @@ -6,7 +6,7 @@ import { createGenericContext } from '@/utils/create-generic-context' import { VoteManagementContextType, VoteManagementProviderProps } from '@/context/voteManagement' -import { useCallback, useEffect, useMemo, useState } from 'react' +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import { useAccount, useChainId } from 'wagmi' import { VoteStateLite, VotingRound } from '@/model/vote.model' import { useInterfoldServer } from '@/hooks/interfold/useInterfoldServer' @@ -47,6 +47,7 @@ const VoteManagementProvider = ({ children }: VoteManagementProviderProps) => { const [txUrl, setTxUrl] = useState(undefined) const [pollResult, setPollResult] = useState(null) const [currentRoundId, setCurrentRoundId] = useState(null) + const [pendingCurrentRoundId, setPendingCurrentRoundId] = useState(null) const [displayedRoundIsFallback, setDisplayedRoundIsFallback] = useState(false) const [hasVotedInCurrentRound, setHasVotedInCurrentRound] = useState(false) @@ -102,6 +103,21 @@ const VoteManagementProvider = ({ children }: VoteManagementProviderProps) => { [chainId, userAddress, currentRoundId], ) + const applyRoundState = useCallback((fetchedRoundState: VoteStateLite) => { + if (fetchedRoundState.committee_public_key.length === 1 && fetchedRoundState.committee_public_key[0] === 0) { + handleGenericError('getRoundStateLite', { + message: 'Interfold server failed generating the necessary pk bytes', + name: 'getRoundStateLite', + }) + } + + const startBlockNumber = Number(fetchedRoundState.start_block) + setRoundState({ ...fetchedRoundState, start_block: startBlockNumber }) + setVotingRound({ round_id: fetchedRoundState.id, pk_bytes: fetchedRoundState.committee_public_key }) + setPollOptions(generatePoll({ round_id: fetchedRoundState.id, emojis: fetchedRoundState.emojis })) + setRoundEndDate(convertTimestampToDate(fetchedRoundState.end_time)) + }, []) + const initialLoad = async () => { const currentRound = await getCurrentRound() if (!currentRound) return @@ -111,7 +127,11 @@ const VoteManagementProvider = ({ children }: VoteManagementProviderProps) => { // otherwise sit forever in "Over · Tallying…". Fall back to the latest past // round that does have a tally so the user sees something useful. const fetched = await getRoundStateLiteRequest(currentRound.id) - if (!fetched) return + if (!fetched) { + setPendingCurrentRoundId(currentRound.id) + return + } + setPendingCurrentRoundId(null) const ended = Number(fetched.end_time) <= nowInSeconds() let fallbackRoundId: string | null = null @@ -135,28 +155,55 @@ const VoteManagementProvider = ({ children }: VoteManagementProviderProps) => { } setDisplayedRoundIsFallback(fallbackRoundId !== null) - await getRoundStateLite(fallbackRoundId ?? currentRound.id) + if (fallbackRoundId) { + await getRoundStateLite(fallbackRoundId) + } else { + applyRoundState(fetched) + } } const getRoundStateLite = async (roundId: string) => { const fetchedRoundState = await getRoundStateLiteRequest(roundId) - if (fetchedRoundState?.committee_public_key.length === 1 && fetchedRoundState.committee_public_key[0] === 0) { - handleGenericError('getRoundStateLite', { - message: 'Interfold server failed generating the necessary pk bytes', - name: 'getRoundStateLite', - }) - } if (fetchedRoundState) { - const startBlockNumber = Number(fetchedRoundState.start_block) - setRoundState({ ...fetchedRoundState, start_block: startBlockNumber }) - setVotingRound({ round_id: fetchedRoundState.id, pk_bytes: fetchedRoundState.committee_public_key }) - setPollOptions(generatePoll({ round_id: fetchedRoundState.id, emojis: fetchedRoundState.emojis })) - setRoundEndDate(convertTimestampToDate(fetchedRoundState.end_time)) - setCurrentRoundId(fetchedRoundState.id) + applyRoundState(fetchedRoundState) } } + const getRoundStateLiteRequestRef = useRef(getRoundStateLiteRequest) + useEffect(() => { + getRoundStateLiteRequestRef.current = getRoundStateLiteRequest + }, [getRoundStateLiteRequest]) + + useEffect(() => { + if (!pendingCurrentRoundId) return + + let cancelled = false + let timer: ReturnType | null = null + + const poll = async () => { + if (cancelled) return + + if (typeof document === 'undefined' || !document.hidden) { + const fetched = await getRoundStateLiteRequestRef.current(pendingCurrentRoundId) + if (cancelled) return + if (fetched) { + applyRoundState(fetched) + setPendingCurrentRoundId(null) + return + } + } + + timer = setTimeout(poll, 10_000) + } + + timer = setTimeout(poll, 10_000) + return () => { + cancelled = true + if (timer) clearTimeout(timer) + } + }, [pendingCurrentRoundId, applyRoundState]) + const getPastPolls = async () => { try { const result = await getWebResult() diff --git a/examples/CRISP/client/src/hooks/generic/useFetchApi.tsx b/examples/CRISP/client/src/hooks/generic/useFetchApi.tsx index a175dd6b14..5aa6a836dc 100644 --- a/examples/CRISP/client/src/hooks/generic/useFetchApi.tsx +++ b/examples/CRISP/client/src/hooks/generic/useFetchApi.tsx @@ -8,6 +8,10 @@ import { useState } from 'react' import axios, { AxiosRequestConfig, Method } from 'axios' import { handleGenericError } from '@/utils/handle-generic-error' +type FetchConfig = AxiosRequestConfig & { + suppressNotFound?: boolean +} + export const useApi = () => { const [isLoading, setIsLoading] = useState(false) @@ -15,13 +19,15 @@ export const useApi = () => { url: string, method: Method = 'get', data?: U, - config?: AxiosRequestConfig, + config?: FetchConfig, ): Promise => { setIsLoading(true) + const { suppressNotFound = false, ...axiosConfig } = config ?? {} try { - const response = method === 'get' ? await axios.get(`${url}`, config) : await axios.post(`${url}`, data, config) + const response = method === 'get' ? await axios.get(`${url}`, axiosConfig) : await axios.post(`${url}`, data, axiosConfig) return response.data } catch (error) { + if (suppressNotFound && axios.isAxiosError(error) && error.response?.status === 404) return undefined handleGenericError(`API Error - ${url}`, error as Error) } finally { setIsLoading(false) diff --git a/examples/CRISP/client/src/hooks/interfold/useInterfoldServer.ts b/examples/CRISP/client/src/hooks/interfold/useInterfoldServer.ts index 487bd80653..597a3538ee 100644 --- a/examples/CRISP/client/src/hooks/interfold/useInterfoldServer.ts +++ b/examples/CRISP/client/src/hooks/interfold/useInterfoldServer.ts @@ -37,7 +37,8 @@ export const useInterfoldServer = () => { const { GetCurrentRound, GetWebAllResult, BroadcastVote, GetRoundStateLite, GetWebResult, GetVoteStatus } = InterfoldEndpoints const { fetchData, isLoading } = useApi() const getCurrentRound = () => fetchData(GetCurrentRound, 'post', { requesters: ROUND_REQUESTERS }) - const getRoundStateLite = (round_id: string) => fetchData(GetRoundStateLite, 'post', { round_id }) + const getRoundStateLite = (round_id: string) => + fetchData(GetRoundStateLite, 'post', { round_id }, { suppressNotFound: true }) const broadcastVote = (vote: BroadcastVoteRequest) => fetchData(BroadcastVote, 'post', vote) const getWebResult = () => fetchData(GetWebAllResult, 'post', { requesters: ROUND_REQUESTERS }) diff --git a/examples/CRISP/client/src/pages/Landing/components/DailyPoll.tsx b/examples/CRISP/client/src/pages/Landing/components/DailyPoll.tsx index 7d249f2907..d8c3a8f268 100644 --- a/examples/CRISP/client/src/pages/Landing/components/DailyPoll.tsx +++ b/examples/CRISP/client/src/pages/Landing/components/DailyPoll.tsx @@ -56,6 +56,7 @@ const DailyPollSection: React.FC = ({ loading, endTime, t isLoading, getWebResultByRound, displayedRoundIsFallback, + currentRoundId, } = useVoteManagementContext() const { canRegister, isRegistered, isRegistering, register } = useRegistration() const navigate = useNavigate() @@ -226,7 +227,12 @@ const DailyPollSection: React.FC = ({ loading, endTime, t

{title}
{hasPoll &&

Choose your favorite

} - {!roundState && !isLoading &&

No active poll found. Check back when the next round opens.

} + {!roundState && !isLoading && currentRoundId && ( +

Round #{currentRoundId} is preparing its encryption key. Voting will open when the key is available.

+ )} + {!roundState && !isLoading && !currentRoundId && ( +

No active poll found. Check back when the next round opens.

+ )} {displayedRoundIsFallback && (

Showing the latest completed poll — the current round is still being tallied under encryption.

)} diff --git a/examples/CRISP/crates/evm_helpers/src/lib.rs b/examples/CRISP/crates/evm_helpers/src/lib.rs index eb67010b4c..bc82557d91 100644 --- a/examples/CRISP/crates/evm_helpers/src/lib.rs +++ b/examples/CRISP/crates/evm_helpers/src/lib.rs @@ -129,6 +129,14 @@ impl CRISPContract { Ok(receipt) } + /// Read the Merkle root already stored for a round. + pub async fn get_merkle_root(&self, e3_id: U256) -> Result { + let contract = CRISPProgram::new(self.contract_address, self.provider.as_ref()); + let round = contract.getRoundData(e3_id).call().await?; + + Ok(round.merkleRoot) + } + /// Dry-run `publishInput` as an `eth_call` from the relay's own account. /// /// The relay signs and pays for whatever it is handed, so an input that would revert — a bad diff --git a/examples/CRISP/server/src/server/indexer.rs b/examples/CRISP/server/src/server/indexer.rs index 15e83911ee..7cf5d2f0dd 100644 --- a/examples/CRISP/server/src/server/indexer.rs +++ b/examples/CRISP/server/src/server/indexer.rs @@ -30,13 +30,14 @@ use e3_sdk::{ }, retry::call_with_retry, }, - indexer::{DataStore, InterfoldIndexer, SharedStore}, + indexer::{DataStore, IndexerContext, InterfoldIndexer, SharedStore}, }; use evm_helpers::{CRISPContractFactory, InputPublished}; use eyre::Context; use log::{error, info, warn}; use num_bigint::BigUint; use std::error::Error; +use std::sync::Arc; use std::time::Duration; use tokio::time::sleep; @@ -378,7 +379,7 @@ pub async fn register_e3_requested( .with_context(|| format!("[e3_id={}] Invalid E3 ID", e3_id))?; info!( - "[e3_id={}] Calling setMerkleRoot with root: {}", + "[e3_id={}] Ensuring CRISPProgram Merkle root: {}", e3_id, merkle_root_u256 ); @@ -392,19 +393,54 @@ pub async fn register_e3_requested( format!("[e3_id={}] Failed to create CRISP contract", e3_id) })?; - let receipt = contract - .set_merkle_root(e3_id_u256, merkle_root_u256) - .await - .with_context(|| { - format!("[e3_id={}] Failed to call setMerkleRoot", e3_id) - })?; - - info!( - "[e3_id={}] setMerkleRoot successful. TxHash: {:?}", - e3_id, receipt.transaction_hash - ); + let stored_root = contract.get_merkle_root(e3_id_u256).await?; + if stored_root == merkle_root_u256 { + info!( + "[e3_id={}] Merkle root is already set to the expected value", + e3_id + ); + } else if stored_root.is_zero() { + match contract + .set_merkle_root(e3_id_u256, merkle_root_u256) + .await + { + Ok(receipt) => info!( + "[e3_id={}] setMerkleRoot successful. TxHash: {:?}", + e3_id, receipt.transaction_hash + ), + Err(error) => { + // A live subscription and its overlap replay can race here. Accept + // the losing transaction only when the desired root landed. + let root_after_error = + contract.get_merkle_root(e3_id_u256).await?; + if root_after_error != merkle_root_u256 { + return Err(error).with_context(|| { + format!( + "[e3_id={}] Failed to call setMerkleRoot", + e3_id + ) + }); + } + info!( + "[e3_id={}] Merkle root was set by a concurrent handler", + e3_id + ); + } + } + } else { + return Err(eyre::eyre!( + "[e3_id={}] CRISPProgram has a different Merkle root: expected {}, got {}", + e3_id, + merkle_root_u256, + stored_root + )); + } } + // Committee and request handlers run concurrently for live logs. If the key was + // indexed while census preparation was still running, this closes that race. + activate_round_if_ready(e3_id.clone(), ctx).await?; + Ok(()) } }) @@ -473,6 +509,169 @@ async fn wait_for_indexed_inputs( /// handler, so two passes do not overlap. const DEADLINE_RETRY_OFFSETS: [u64; 3] = [60, 180, 420]; +const ROUND_ACTIVATION_RETRY_OFFSETS: [u64; 5] = [1, 5, 30, 120, 600]; + +fn deadline_attempt_times(expiration: u64, now: u64) -> [u64; 4] { + let first = expiration.max(now); + [ + first, + first.saturating_add(DEADLINE_RETRY_OFFSETS[0]), + first.saturating_add(DEADLINE_RETRY_OFFSETS[1]), + first.saturating_add(DEADLINE_RETRY_OFFSETS[2]), + ] +} + +async fn handle_e3_input_deadline_expiration_logged( + e3_id: String, + store: SharedStore, +) -> eyre::Result<()> { + if let Err(error) = handle_e3_input_deadline_expiration(e3_id.clone(), store).await { + error!("[e3_id={}] CRISP deadline pass failed: {}", e3_id, error); + } + Ok(()) +} + +async fn activate_round_if_ready( + e3_id: String, + ctx: Arc>, +) -> eyre::Result { + let store = ctx.store(); + let mut repo = CrispE3Repository::new(store.clone(), &e3_id); + if !repo.has_crisp_record().await? || !repo.has_indexed_public_key().await? { + return Ok(false); + } + + let expiration = repo.get_input_deadline().await?; + if !repo.try_start_round().await? { + return Ok(true); + } + + let now = chrono::Utc::now().timestamp().max(0) as u64; + for at in deadline_attempt_times(expiration, now) { + let e3_id = e3_id.clone(); + ctx.do_later(at, move |_, ctx| { + handle_e3_input_deadline_expiration_logged(e3_id.clone(), ctx.store()) + }); + } + + let mut current_round_repo = CurrentRoundRepository::new(store); + current_round_repo + .set_current_round(CurrentRound { id: e3_id.clone() }) + .await?; + info!( + "[e3_id={}] Activated CRISP round and registered deadline callbacks", + e3_id + ); + Ok(true) +} + +fn schedule_round_activation_retries( + e3_id: &str, + ctx: &Arc>, +) { + let now = chrono::Utc::now().timestamp().max(0) as u64; + for offset in ROUND_ACTIVATION_RETRY_OFFSETS { + let e3_id = e3_id.to_string(); + ctx.do_later(now.saturating_add(offset), move |_, ctx| { + let e3_id = e3_id.clone(); + async move { + if let Err(error) = activate_round_if_ready(e3_id.clone(), ctx).await { + error!( + "[e3_id={}] Deferred CRISP round activation failed: {}", + e3_id, error + ); + } + Ok(()) + } + }); + } +} + +async fn restore_round_deadline_callback( + indexer: &InterfoldIndexer, + store: SharedStore, + e3_id: String, + now: u64, +) -> Result<()> { + let mut repo = CrispE3Repository::new(store.clone(), &e3_id); + let mut status = repo.get_status().await?; + let has_indexed_public_key = repo.has_indexed_public_key().await?; + if status == "Active" && !has_indexed_public_key { + repo.update_status("Requested").await?; + status = "Requested".to_string(); + warn!( + "[e3_id={}] Reset an active round to pending because no verified public key is indexed", + e3_id + ); + } + if status == "Requested" + && has_indexed_public_key + && repo.try_start_round().await? + { + let mut current_round_repo = CurrentRoundRepository::new(store.clone()); + current_round_repo + .set_current_round(CurrentRound { id: e3_id.clone() }) + .await?; + status = "Active".to_string(); + info!( + "[e3_id={}] Activated a requested round whose verified key was indexed before restart", + e3_id + ); + } + if status == "Computing" || status == "PublishingCiphertext" { + // Submission is intentionally at-least-once across the CRISP and program-server process + // boundary. A crash can lose the HTTP response or webhook, so keeping this claim would + // strand the round. A retry can repeat proof work, but it cannot publish a second result: + // Interfold accepts ciphertext output only from KeyPublished, and the callback treats an + // already-published output as success. + repo.update_status("Expired").await?; + status = "Expired".to_string(); + warn!( + "[e3_id={}] Reset an interrupted compute submission so it can be retried", + e3_id + ); + } + if status != "Active" && status != "Expired" { + return Ok(()); + } + + let expiration = repo.get_input_deadline().await?; + for at in deadline_attempt_times(expiration, now) { + let e3_id = e3_id.clone(); + indexer.schedule_at(at, move |_, ctx| { + handle_e3_input_deadline_expiration_logged(e3_id.clone(), ctx.store()) + }); + } + info!( + "[e3_id={}] Restored deadline callbacks for CRISP round in status {}", + e3_id, status + ); + Ok(()) +} + +async fn restore_round_deadline_callbacks( + indexer: &InterfoldIndexer, +) -> Result<()> { + let store = indexer.get_store(); + let round_ids = CurrentRoundRepository::new(store.clone()) + .get_round_ids() + .await?; + let now = chrono::Utc::now().timestamp().max(0) as u64; + + for e3_id in round_ids { + if let Err(error) = + restore_round_deadline_callback(indexer, store.clone(), e3_id.clone(), now).await + { + error!( + "[e3_id={}] Could not restore CRISP deadline callbacks: {}", + e3_id, error + ); + } + } + + Ok(()) +} + /// Store key holding the `INDEX_LOG_CONTRACTS` set as of the previous run. /// /// Coverage records outlive the configuration that created them, and the store has no delete. This @@ -488,15 +687,10 @@ async fn handle_e3_input_deadline_expiration( let mut repo = CrispE3Repository::new(store.clone(), &e3_id); let e3: e3_sdk::indexer::models::E3 = repo.get_e3().await?; - // A cheap skip for a retry pass over a round that already moved on, so it does not sit through - // the indexer wait below. Not the safety barrier — `try_claim_computing` is, further down. - let status = repo.get_status().await?; - if status == "Computing" || status == "Finished" { + if !repo.try_mark_expired().await? { return Ok(()); } - repo.update_status("Expired").await?; - let voter_count = repo.get_vote_count().await?; // The contract is the authority on how many inputs there are, and this callback can run before @@ -548,9 +742,10 @@ async fn handle_e3_input_deadline_expiration( votes.len(), voter_count ); - // The barrier. Two passes can be inside the indexer wait at once, and `run_compute` is - // one-shot, so the transition to "Computing" has to be the thing that decides which one - // proceeds — in a single store operation, not a read followed by a write. + // The local concurrency barrier. Two passes can be inside the indexer wait at once, so the + // transition to "Computing" has to decide which one proceeds. It must use one store + // operation, not a read followed by a write. Restart recovery remains at-least-once because + // the contract, not this process, is the durable idempotency boundary. // // Claimed here rather than before the wait: a pass that gives up on a short index leaves // the round "Expired" so a later pass can still take it, and claiming earlier would pin it @@ -563,45 +758,65 @@ async fn handle_e3_input_deadline_expiration( return Ok(()); } - let (id, status) = run_compute( - &e3_id, - e3.chain_id, - e3.interfold_address, - e3.encryption_scheme_id, - e3.committee_public_key_hash, - e3.e3_params, - RoundInputs { - ciphertexts: snapshot.ciphertexts, - commitments: snapshot.commitments, - slots: snapshot.slots, - parents: snapshot.parents, - }, - format!( - "{}/state/add-result", - CONFIG.interfold_server_url_for_clients() - ), - ) - .await - .map_err(|e| eyre::eyre!("Error sending run compute request: {e}"))?; + let submission = async { + let (id, status) = run_compute( + &e3_id, + e3.chain_id, + e3.interfold_address, + e3.encryption_scheme_id, + e3.committee_public_key_hash, + e3.e3_params, + RoundInputs { + ciphertexts: snapshot.ciphertexts, + commitments: snapshot.commitments, + slots: snapshot.slots, + parents: snapshot.parents, + }, + format!( + "{}/state/add-result", + CONFIG.interfold_server_url_for_clients() + ), + ) + .await + .map_err(|e| eyre::eyre!("Error sending run compute request: {e}"))?; + + if id != e3_id { + return Err(eyre::eyre!( + "Computation request returned unexpected E3 ID: expected {}, got {}", + e3_id, + id + )); + } - if id != e3_id { - return Err(eyre::eyre!( - "Computation request returned unexpected E3 ID: expected {}, got {}", - e3_id, - id - )); + if status != "processing" { + return Err(eyre::eyre!( + "Computation request failed with status: {}", + status + )); + } + + Ok::<(), eyre::Report>(()) } + .await; - if status != "processing" { - return Err(eyre::eyre!( - "Computation request failed with status: {}", - status - )); + if let Err(submission_error) = submission { + if let Err(release_error) = repo.release_compute_submission().await { + error!( + "[e3_id={}] Failed to release compute claim after submission error: {}", + e3_id, release_error + ); + } + return Err(submission_error.into()); } info!("[e3_id={}] Request Computation for E3", e3_id); - repo.update_status("PublishingCiphertext").await?; + if !repo.mark_compute_submitted().await? { + warn!( + "[e3_id={}] Compute response arrived after the claim was released; a later deadline pass will retry", + e3_id + ); + } } else { info!( "[e3_id={}] E3 has no votes to decrypt. Setting status to Finished.", @@ -671,37 +886,15 @@ pub async fn register_committee_published( indexer .add_event_handler(move |event: CommitteePublished, ctx| { async move { - let store = ctx.store(); let e3_id = event.e3Id.to_string(); - let mut repo = CrispE3Repository::new(store.clone(), &e3_id); - let mut current_round_repo = CurrentRoundRepository::new(store); info!("[e3_id={}] Handling CommitteePublished", e3_id); - // Get current time - let now = get_current_timestamp_rpc().await?; - info!("[e3_id={}] Current time: {}", event.e3Id, now); - - repo.start_round().await?; - current_round_repo - .set_current_round(CurrentRound { id: e3_id.clone() }) - .await?; - - let expiration = repo.get_input_deadline().await?; - - info!("[e3_id={}] Registering hook for {}", e3_id, expiration); - // Registered once per offset, up front. A pass that finds the indexer behind - // returns without computing, and `do_later` has already dropped that callback, so - // the round would otherwise stay "Expired" for good. Every pass after the first - // returns immediately once the round is computing or finished. - for at in std::iter::once(expiration).chain( - DEADLINE_RETRY_OFFSETS - .iter() - .map(|offset| expiration + offset), - ) { - let e3_id = e3_id.clone(); - ctx.do_later(at, move |_, ctx| { - handle_e3_input_deadline_expiration(e3_id.clone(), ctx.store()) - }); + if !activate_round_if_ready(e3_id.clone(), ctx.clone()).await? { + warn!( + "[e3_id={}] Committee event arrived, but the verified public key or CRISP request record is unavailable; round remains pending", + e3_id + ); + schedule_round_activation_retries(&e3_id, &ctx); } Ok(()) @@ -989,6 +1182,7 @@ pub async fn start_indexer( } } + restore_round_deadline_callbacks(&crisp_indexer).await?; crisp_indexer.listen().await?; info!("CRISP: Indexer listen loop has finished!"); Ok(()) @@ -996,6 +1190,7 @@ pub async fn start_indexer( #[cfg(test)] mod custom_params_decoding_tests { + use super::deadline_attempt_times; use crate::server::models::CensusMode; use alloy::dyn_abi::SolType; use alloy::primitives::{Address, U256}; @@ -1081,4 +1276,10 @@ mod custom_params_decoding_tests { let decoded = ::abi_decode(&encode(3)).unwrap(); assert!(CensusMode::try_from(decoded.5.to::()).is_err()); } + + #[test] + fn restart_spreads_overdue_deadline_attempts_from_now() { + assert_eq!(deadline_attempt_times(100, 200), [200, 260, 380, 620]); + assert_eq!(deadline_attempt_times(300, 200), [300, 360, 480, 720]); + } } diff --git a/examples/CRISP/server/src/server/models.rs b/examples/CRISP/server/src/server/models.rs index d5e97d2324..95cb3ad7dc 100644 --- a/examples/CRISP/server/src/server/models.rs +++ b/examples/CRISP/server/src/server/models.rs @@ -261,7 +261,7 @@ pub struct E3 { pub requester: String, } -#[derive(Debug, Deserialize, Serialize)] +#[derive(Clone, Debug, Deserialize, Serialize)] pub struct E3Crisp { pub emojis: [String; 2], pub start_time: u64, diff --git a/examples/CRISP/server/src/server/repo.rs b/examples/CRISP/server/src/server/repo.rs index 8d7eaf0a0f..06bc052c02 100644 --- a/examples/CRISP/server/src/server/repo.rs +++ b/examples/CRISP/server/src/server/repo.rs @@ -91,25 +91,8 @@ impl CurrentRoundRepository { for round_id in self.get_round_ids().await?.into_iter().rev() { let crisp_repo = CrispE3Repository::new(self.store.clone(), &round_id); - match crisp_repo.get_e3_state_lite().await { - Ok(state) => { - if state.requester == requester { - return Ok(Some(CurrentRound { id: round_id })); - } - } - Err(e) => { - // Expected for a round between E3Requested and CommitteePublished: the CRISP - // record and the round index are written at request time, but the `_e3:` record - // only exists once the committee publishes its key, so a freshly requested - // round is half-indexed for the duration of the DKG. Persistent repeats for - // the same round mean the key never arrived — no registered ciphernodes, or a - // CommitteePublished the indexer rejected (see its log for the reason). - info!( - "Round {} is not fully indexed yet (usually: committee key pending) — skipping: {:?}", - round_id, e - ); - continue; - } + if crisp_repo.is_requested_by(&requester).await? { + return Ok(Some(CurrentRound { id: round_id })); } } @@ -154,6 +137,7 @@ impl CrispE3Repository { } } + #[cfg(test)] async fn set_crisp(&mut self, value: E3Crisp) -> Result<()> { let key = self.crisp_key(); self.store @@ -179,14 +163,27 @@ impl CrispE3Repository { /// Whether this server has a record of the round at all. /// /// The CRISP record is written when `E3Requested` is indexed; the indexer's `_e3:` record only - /// lands on `CommitteePublished`. So a round mid-DKG — or one whose committee never formed — - /// has the first and not the second, and reads needing both come back empty. Without this the - /// two are indistinguishable, and "the committee has not published a key" reads as "no such - /// round", which is a very different thing to debug. + /// lands after a public-key byte event passes commitment verification. A round can + /// therefore have the first record and not the second during DKG or after on-chain + /// `KeyPublished`. Without this check, "verified key bytes are pending" reads as "no such + /// round". pub async fn has_crisp_record(&self) -> Result { Ok(self.try_get_crisp().await?.is_some()) } + /// Whether the request-time CRISP record belongs to `requester`. + pub async fn is_requested_by(&self, requester: &str) -> Result { + Ok(self + .try_get_crisp() + .await? + .is_some_and(|round| round.requester.eq_ignore_ascii_case(requester))) + } + + /// Whether the generic indexer stored a verified committee public key for this round. + pub async fn has_indexed_public_key(&self) -> Result { + Ok(self.try_get_e3().await?.is_some()) + } + async fn get_crisp(&self) -> Result { let key = self.crisp_key(); let e3_crisp = self @@ -196,11 +193,27 @@ impl CrispE3Repository { Ok(e3_crisp) } - pub async fn start_round(&mut self) -> Result<()> { - let mut e3_crisp = self.get_crisp().await?; - e3_crisp.start_time = chrono::Utc::now().timestamp() as u64; - e3_crisp.status = "Active".to_string(); - self.set_crisp(e3_crisp).await + /// Start a requested round once. Duplicate committee events do not reset its deadline state. + pub async fn try_start_round(&mut self) -> Result { + let key = self.crisp_key(); + let mut started = false; + let now = chrono::Utc::now().timestamp() as u64; + + self.store + .modify(&key, |e3_obj: Option| { + e3_obj.map(|mut e| { + if e.status == "Requested" { + e.start_time = now; + e.status = "Active".to_string(); + started = true; + } + e + }) + }) + .await + .map_err(|_| eyre::eyre!("Could not start round at '{key}'"))?; + + Ok(started) } pub async fn insert_ciphertext_input( @@ -275,7 +288,8 @@ impl CrispE3Repository { end_time: u64, snapshot_block: u64, ) -> Result<()> { - self.set_crisp(E3Crisp { + let key = self.crisp_key(); + let initial = E3Crisp { input_commitments: Vec::new(), input_slots: Vec::new(), input_parents: Vec::new(), @@ -296,8 +310,15 @@ impl CrispE3Repository { census_mode: custom_params.census_mode, end_time, snapshot_block, - }) - .await + }; + + self.store + .modify(&key, move |current: Option| { + current.or_else(|| Some(initial.clone())) + }) + .await + .map_err(|_| eyre::eyre!("Could not initialize round at '{key}'"))?; + Ok(()) } fn get_e3_repo(&self) -> E3Repository { @@ -351,7 +372,9 @@ impl CrispE3Repository { /// Returns whether this caller made the transition. One store operation, because `modify` is a /// read-modify-write under a single write lock: reading the status and writing it back as two /// separate awaits leaves a window where two deadline passes both observe "Expired" and both - /// start the one-shot `run_compute`, publishing two results for one round. + /// submit `run_compute` concurrently. Restart recovery can submit again because the remote + /// response might have been lost; Interfold is the durable idempotency boundary and accepts + /// only the first valid ciphertext output. pub async fn try_claim_computing(&mut self) -> Result { let key = self.crisp_key(); let mut claimed = false; @@ -359,7 +382,7 @@ impl CrispE3Repository { self.store .modify(&key, |e3_obj: Option| { e3_obj.map(|mut e| { - if e.status != "Computing" && e.status != "Finished" { + if e.status == "Expired" { e.status = "Computing".to_string(); claimed = true; } @@ -372,6 +395,69 @@ impl CrispE3Repository { Ok(claimed) } + /// Record that the program server accepted the claimed computation. + pub async fn mark_compute_submitted(&mut self) -> Result { + let key = self.crisp_key(); + let mut submitted = false; + + self.store + .modify(&key, |e3_obj: Option| { + e3_obj.map(|mut e| { + if e.status == "Computing" { + e.status = "PublishingCiphertext".to_string(); + submitted = true; + } + e + }) + }) + .await + .map_err(|_| eyre::eyre!("Could not record compute submission at '{key}'"))?; + + Ok(submitted) + } + + /// Mark an active round as expired, or keep an expired round eligible for a retry pass. + pub async fn try_mark_expired(&mut self) -> Result { + let key = self.crisp_key(); + let mut eligible = false; + + self.store + .modify(&key, |e3_obj: Option| { + e3_obj.map(|mut e| { + if e.status == "Active" || e.status == "Expired" { + e.status = "Expired".to_string(); + eligible = true; + } + e + }) + }) + .await + .map_err(|_| eyre::eyre!("Could not expire round at '{key}'"))?; + + Ok(eligible) + } + + /// Release a failed compute submission so a later deadline pass can retry it. + pub async fn release_compute_submission(&mut self) -> Result { + let key = self.crisp_key(); + let mut released = false; + + self.store + .modify(&key, |e3_obj: Option| { + e3_obj.map(|mut e| { + if e.status == "Computing" || e.status == "PublishingCiphertext" { + e.status = "Expired".to_string(); + released = true; + } + e + }) + }) + .await + .map_err(|_| eyre::eyre!("Could not release computation at '{key}'"))?; + + Ok(released) + } + pub async fn update_status(&mut self, value: &str) -> Result<()> { let key = self.crisp_key(); @@ -421,12 +507,6 @@ impl CrispE3Repository { Ok(e3.committee_public_key) } - pub async fn get_web_result_request(&self) -> Result { - self.try_get_web_result_request() - .await? - .ok_or_else(|| eyre::eyre!("No state stored for round {}", self.e3_id)) - } - /// The round's result, or `None` when the round is not in the store. See /// [`Self::try_get_e3_state_lite`] for why both records have to be present. pub async fn try_get_web_result_request(&self) -> Result> { @@ -706,7 +786,43 @@ pub fn parse_slot_address(address: &str) -> Result<[u8; 20]> { #[cfg(test)] mod tests { - use super::{count_active_slots, parse_slot_address, snapshot_block}; + use super::{ + count_active_slots, parse_slot_address, snapshot_block, CrispE3Repository, + CurrentRoundRepository, + }; + use crate::server::models::{CensusMode, CreditMode, CustomParams, E3Crisp}; + use e3_sdk::indexer::{InMemoryStore, SharedStore}; + use std::sync::Arc; + use tokio::sync::RwLock; + + fn test_store() -> SharedStore { + SharedStore::new(Arc::new(RwLock::new(InMemoryStore::new()))) + } + + fn crisp_round(requester: &str, status: &str) -> E3Crisp { + E3Crisp { + emojis: ["one".to_string(), "two".to_string()], + start_time: 0, + end_time: 100, + status: status.to_string(), + tally: vec![], + token_holder_hashes: vec![], + eligible_addresses: vec![], + token_address: "0x0000000000000000000000000000000000000001".to_string(), + balance_threshold: "1".to_string(), + ciphertext_inputs: vec![], + input_commitments: vec![], + input_slots: vec![], + input_usable: vec![], + input_parents: vec![], + requester: requester.to_string(), + num_options: "2".to_string(), + credit_mode: CreditMode::Constant, + credits: Some("1".to_string()), + snapshot_block: 1, + census_mode: CensusMode::Token, + } + } #[test] fn counts_each_slot_once_no_matter_how_long_its_chain_is() { @@ -743,4 +859,76 @@ mod tests { fn does_not_underflow_on_the_genesis_block() { assert_eq!(snapshot_block(0, 0), 0); } + + #[tokio::test] + async fn requested_round_is_visible_before_the_public_key_is_indexed() { + let store = test_store(); + let requester = "0x1111111111111111111111111111111111111111"; + let mut round = CrispE3Repository::new(store.clone(), "8"); + round + .set_crisp(crisp_round(requester, "Requested")) + .await + .unwrap(); + + let mut current = CurrentRoundRepository::new(store); + current.record_round("8").await.unwrap(); + + let found = current + .get_current_round_for_requester(requester.to_uppercase()) + .await + .unwrap() + .expect("the request-time CRISP record should be sufficient"); + assert_eq!(found.id, "8"); + } + + #[tokio::test] + async fn compute_claim_is_released_after_a_submission_error() { + let store = test_store(); + let mut round = CrispE3Repository::new(store, "9"); + round + .set_crisp(crisp_round("requester", "Requested")) + .await + .unwrap(); + + assert!(round.try_start_round().await.unwrap()); + assert!(!round.try_start_round().await.unwrap()); + assert!(round.try_mark_expired().await.unwrap()); + assert!(round.try_claim_computing().await.unwrap()); + assert!(!round.try_claim_computing().await.unwrap()); + assert!(round.release_compute_submission().await.unwrap()); + assert_eq!(round.get_status().await.unwrap(), "Expired"); + assert!(round.try_claim_computing().await.unwrap()); + assert!(round.mark_compute_submitted().await.unwrap()); + assert_eq!(round.get_status().await.unwrap(), "PublishingCiphertext"); + assert!(round.release_compute_submission().await.unwrap()); + assert_eq!(round.get_status().await.unwrap(), "Expired"); + } + + #[tokio::test] + async fn requested_event_replay_does_not_reset_round_status() { + let store = test_store(); + let mut round = CrispE3Repository::new(store, "10"); + let params = || CustomParams { + token_address: "0x0000000000000000000000000000000000000001".to_string(), + balance_threshold: "1".to_string(), + num_options: "2".to_string(), + credit_mode: CreditMode::Constant, + credits: Some("1".to_string()), + census_mode: CensusMode::Token, + voting_power_divisor: "0".to_string(), + }; + + round + .initialize_round(params(), "requester".to_string(), 100, 1) + .await + .unwrap(); + round.update_status("Finished").await.unwrap(); + round + .initialize_round(params(), "requester".to_string(), 200, 2) + .await + .unwrap(); + + assert_eq!(round.get_status().await.unwrap(), "Finished"); + assert_eq!(round.get_input_deadline().await.unwrap(), 100); + } } diff --git a/examples/CRISP/server/src/server/routes/state.rs b/examples/CRISP/server/src/server/routes/state.rs index c083e08112..cbeeacf145 100644 --- a/examples/CRISP/server/src/server/routes/state.rs +++ b/examples/CRISP/server/src/server/routes/state.rs @@ -59,19 +59,19 @@ fn round_not_found(e3_id: &str) -> HttpResponse { }) } -/// The round IS indexed — its committee just has not published a key. +/// The round is indexed, but verified public-key bytes are not available. /// /// Same status as an unknown round, because there is still nothing to serve, but never the same /// message. The two have completely different causes: one means the request was never seen, the -/// other means DKG has not completed (or never will, for a round that failed). Reporting both as -/// "no state for round X" sent us looking for a broken indexer when the indexer was fine and the -/// ciphernodes were not. +/// other means the byte publication has not arrived or did not verify. `KeyPublished` on chain is +/// not sufficient because that stage records the proof-backed commitment before the byte event. async fn round_state_pending(store: &web::Data, e3_id: &str) -> HttpResponse { match store.e3(e3_id).has_crisp_record().await { Ok(true) => HttpResponse::NotFound().json(JsonResponse { response: format!( - "Round {e3_id} is indexed, but its committee has not published a key yet, so \ - there is no state to serve. Check whether the round has failed on chain." + "Round {e3_id} is indexed, but verified committee public-key bytes are not \ + available, so there is no state to serve. KeyPublished on chain confirms only \ + the commitment. Check the byte-publication event and the on-chain failure state." ), }), Ok(false) => round_not_found(e3_id), @@ -145,9 +145,8 @@ async fn handle_program_server_result(data: web::Json) -> impl R WebhookPayload::Failed { e3_id, error } => { error!("Computation failed for E3 ID: {}. Error: {}", e3_id, error); - // TODO: Update E3 state to indicate computation failed - // TODO: Handle ciphernode rewards for partial work - // TODO: Emit on-chain event if needed + // This callback is not authenticated. Do not let a caller move durable round state by + // claiming that the program server failed. HttpResponse::Ok().json(format!( "Computation failed for E3 ID: {}. Error: {}", @@ -305,8 +304,8 @@ async fn get_all_round_results( let requesters = incoming.requesters; for e3_id in round_ids { - match store.e3(&e3_id).get_web_result_request().await { - Ok(w) => { + match store.e3(&e3_id).try_get_web_result_request().await { + Ok(Some(w)) => { if !requesters.is_empty() { // if we have any requesters to filter by, do it if requesters.contains(&w.requester) { @@ -316,15 +315,16 @@ async fn get_all_round_results( states.push(w); } } - Err(e) => { - // Expected for a round whose committee key is not published yet — the `_e3:` - // record only exists after CommitteePublished. See the note in - // `get_current_round_for_requester`. + Ok(None) => { info!( - "Round {} is not fully indexed yet (usually: committee key pending) — skipping: {:?}", - e3_id, e + "Round {} has no verified public-key bytes yet; skipping it", + e3_id ); - continue; + } + Err(error) => { + error!("Could not read round {e3_id} from the store: {error:?}"); + return HttpResponse::InternalServerError() + .body("Failed to retrieve round state"); } } }