feat(ethexe-consensus): introduce Watcher non-signing validator state

ethexe/common/src/db.rs

@@ -277,7 +277,7 @@ mod tests {
     #[test]
     fn ensure_types_unchanged() {
         const EXPECTED_TYPE_INFO_HASH: &str =
-            "fabd74202a18149f60d6396119fb880e09475ca3c0245430d3fab7dff243f8c8";
+            "740f626e5ecdc186db996b42ec13136210eab4ba45f8fa75b4335c1005a4a22f";
 
         let types = [
             meta_type::<BlockMeta>(),

ethexe/consensus/src/validator/batch/manager.rs

@@ -132,17 +132,35 @@ impl BatchCommitmentManager {
             self.limits.commitment_delay_limit,
         );
 
-        if let Ok(Some(BatchCommitment {
-            chain_commitment: Some(chain_commitment),
-            ..
-        })) = batch_commitment.as_ref()
-        {
-            self.store_outgoing_actions_for_chain_commitment(chain_commitment);
+        if let Ok(Some(batch)) = batch_commitment.as_ref() {
+            self.persist_outgoing_actions(batch);
         }
 
         batch_commitment
     }
 
+    /// Persist per-`new_state_hash` value-claim mappings to the local DB.
+    fn persist_outgoing_actions(&self, batch: &BatchCommitment) {
+        let Some(ChainCommitment { transitions, .. }) = batch.chain_commitment.as_ref() else {
+            return;
+        };
+        for StateTransition {
+            new_state_hash,
+            value_claims,
+            ..
+        } in transitions
+        {
+            let outgoing_actions: OutgoingActions = value_claims
+                .iter()
+                .cloned()
+                .map(OutgoingAction::ValueClaim)
+                .collect::<Vec<_>>()
+                .into();
+            self.db
+                .set_outgoing_actions(*new_state_hash, outgoing_actions);
+        }
+    }
+
     /// Participant: re-derive the coordinator's batch and return whether digests agree.
     /// Drops the signature (Rejected) on chain mismatch instead of erroring.
     pub async fn validate_batch_commitment(
@@ -324,7 +342,6 @@ impl BatchCommitmentManager {
                 std::mem::take(&mut chain_commitment.transitions),
             );
             super::utils::sort_transitions_by_value_to_receive(&mut chain_commitment.transitions);
-            self.store_outgoing_actions_for_chain_commitment(&chain_commitment);
             batch_parts.chain_commitment = Some(chain_commitment);
         }
 
@@ -356,34 +373,21 @@ impl BatchCommitmentManager {
             });
         }
 
-        let batch_encoded_size = Gear::BatchCommitment::from(batch).abi_encoded_size() as u64;
+        let batch_encoded_size =
+            Gear::BatchCommitment::from(batch.clone()).abi_encoded_size() as u64;
         if batch_encoded_size > self.limits.batch_size_limit {
             return Ok(ValidationStatus::Rejected {
                 request,
                 reason: ValidationRejectReason::BatchSizeLimitExceeded,
             });
         }
 
-        Ok(ValidationStatus::Accepted(digest))
-    }
-
-    fn store_outgoing_actions_for_chain_commitment(&self, commitment: &ChainCommitment) {
-        for StateTransition {
-            new_state_hash,
-            value_claims,
-            ..
-        } in &commitment.transitions
-        {
-            let mut outgoing_actions = vec![];
+        // Cache the per-state-hash value-claim mapping locally. Both validating
+        // validators and non-validator watchers persist here so any node can
+        // serve `mirror.outgoing_actions(state_hash)` for merkle-proof clients.
+        self.persist_outgoing_actions(&batch);
 
-            for value_claim in value_claims {
-                outgoing_actions.push(OutgoingAction::ValueClaim(value_claim.clone()));
-            }
-
-            let outgoing_actions: OutgoingActions = outgoing_actions.into();
-            self.db
-                .set_outgoing_actions(*new_state_hash, outgoing_actions);
-        }
+        Ok(ValidationStatus::Accepted(digest))
     }
 
     pub async fn aggregate_validators_commitment(

ethexe/consensus/src/validator/batch/tests.rs

@@ -12,17 +12,21 @@
 use super::{BatchCommitmentManager, BatchLimits, ValidationStatus, types::ValidationRejectReason};
 use crate::validator::core::MiddlewareWrapper;
 use ethexe_common::{
-    Address, Digest, ProgramStates, Schedule, SimpleBlockData, ToDigest, ValidatorsVec,
+    Address, Digest, OutgoingAction, ProgramStates, Schedule, SimpleBlockData, ToDigest,
+    ValidatorsVec,
     consensus::BatchCommitmentValidationRequest,
-    db::{BlockMetaStorageRW, CompactMb, GlobalsStorageRW, MbStorageRW, SetConfig},
-    gear::StateTransition,
+    db::{
+        BlockMetaStorageRW, CompactMb, GlobalsStorageRW, MbStorageRW, OutgoingActionStorageRO,
+        SetConfig,
+    },
+    gear::{StateTransition, ValueClaim},
     malachite::{ProcessQueuesLimits, Transaction, Transactions},
     mock::*,
 };
 use ethexe_db::Database;
 use ethexe_ethereum::middleware::{ElectionProvider, MockElectionProvider};
 use gear_core::ids::prelude::CodeIdExt;
-use gprimitives::{ActorId, CodeId, H256, U256};
+use gprimitives::{ActorId, CodeId, H256, MessageId, U256};
 use std::num::{NonZero, NonZeroU64};
 
 const BLOCK_GAS_LIMIT: u64 = ethexe_common::DEFAULT_BLOCK_GAS_LIMIT;
@@ -116,6 +120,23 @@ fn nonempty_transition(seed: u8) -> StateTransition {
     }
 }
 
+/// Same shape as [`nonempty_transition`] but carries `n` distinct value
+/// claims so tests can assert per-`new_state_hash` persistence into the
+/// `outgoing_actions` table.
+fn transition_with_value_claims(seed: u8, n: u32) -> StateTransition {
+    let value_claims = (0..n)
+        .map(|i| ValueClaim {
+            message_id: MessageId::from([(seed.wrapping_add(i as u8)); 32]),
+            destination: ActorId::from([0xCC; 32]),
+            value: ((seed as u128) << 8) | (i as u128),
+        })
+        .collect();
+    StateTransition {
+        value_claims,
+        ..nonempty_transition(seed)
+    }
+}
+
 /// Build a batch from a small canonical setup so multiple tests can
 /// share the scaffolding. Returns the chain head block plus the
 /// resulting batch.
@@ -770,3 +791,113 @@ async fn test_aggregate_validators_commitment() {
         .await
         .unwrap_err();
 }
+
+/// Both the producer (`create_batch_commitment`) and the verifier
+/// (`validate_batch_commitment`, used by Participant and Watcher) must
+/// write the `(post-squash state_hash → outgoing_actions)` mapping to the
+/// local DB. Without parity here, RPC clients hitting a non-producer node
+/// (and that's most nodes most of the time) would get empty results from
+/// `mirror.outgoing_actions`, breaking merkle-proof construction for
+/// value claims.
+#[tokio::test]
+async fn validator_path_persists_outgoing_actions() {
+    // 1. Build the canonical batch on `producer_db`. The producer's own
+    //    persist step runs as a side effect of create_batch_commitment.
+    let producer_db = Database::memory();
+    let chain = test_block_chain(3).setup(&producer_db);
+    let block = chain.blocks[3].to_simple();
+
+    let claims_seed = 7u8;
+    let claims_count = 2u32;
+    setup_mb_chain(
+        &producer_db,
+        vec![
+            vec![transition_with_value_claims(claims_seed, claims_count)],
+            vec![nonempty_transition(2)],
+        ],
+    );
+
+    let batch = mock_batch_manager(producer_db.clone())
+        .create_batch_commitment(block)
+        .await
+        .expect("create_batch_commitment must not error")
+        .expect("expected non-empty batch");
+
+    // Pluck the (state_hash, value_claims) pairs the batch ended up with —
+    // these are post-squash hashes, exactly what RPC will query by.
+    let expected_mappings: Vec<(H256, Vec<ValueClaim>)> = batch
+        .chain_commitment
+        .as_ref()
+        .expect("chain commitment present")
+        .transitions
+        .iter()
+        .filter(|t| !t.value_claims.is_empty())
+        .map(|t| (t.new_state_hash, t.value_claims.clone()))
+        .collect();
+    assert!(
+        !expected_mappings.is_empty(),
+        "test fixture must produce at least one transition with value claims",
+    );
+
+    // Sanity check: producer-side persist actually happened.
+    for (state_hash, value_claims) in &expected_mappings {
+        let stored = producer_db
+            .outgoing_actions(*state_hash)
+            .expect("producer must persist outgoing_actions")
+            .into_inner();
+        let expected: Vec<OutgoingAction> = value_claims
+            .iter()
+            .cloned()
+            .map(OutgoingAction::ValueClaim)
+            .collect();
+        assert_eq!(stored, expected, "producer persist mismatch");
+    }
+
+    // 2. Replay the same chain into a fresh DB and verify the batch via
+    //    `validate_batch_commitment`. The watcher and participant paths
+    //    flow through here too. We never call `create_batch_commitment`
+    //    on this DB, so the only writer of `outgoing_actions` is the
+    //    validator path under test.
+    let verifier_db = Database::memory();
+    test_block_chain(3).setup(&verifier_db);
+    setup_mb_chain(
+        &verifier_db,
+        vec![
+            vec![transition_with_value_claims(claims_seed, claims_count)],
+            vec![nonempty_transition(2)],
+        ],
+    );
+
+    // Confirm the verifier DB has no mappings before validation runs.
+    for (state_hash, _) in &expected_mappings {
+        assert!(
+            verifier_db.outgoing_actions(*state_hash).is_none(),
+            "fresh verifier DB unexpectedly already had outgoing_actions"
+        );
+    }
+
+    let request = BatchCommitmentValidationRequest::new(&batch);
+    let status = mock_batch_manager(verifier_db.clone())
+        .validate_batch_commitment(block, request)
+        .await
+        .unwrap();
+    assert!(
+        matches!(status, ValidationStatus::Accepted(_)),
+        "expected acceptance, got {status:?}"
+    );
+
+    // 3. After validation, the verifier DB must hold the same mappings
+    //    that the producer wrote.
+    for (state_hash, value_claims) in &expected_mappings {
+        let stored = verifier_db
+            .outgoing_actions(*state_hash)
+            .expect("validator must persist outgoing_actions after Accepted")
+            .into_inner();
+        let expected: Vec<OutgoingAction> = value_claims
+            .iter()
+            .cloned()
+            .map(OutgoingAction::ValueClaim)
+            .collect();
+        assert_eq!(stored, expected, "validator persist mismatch");
+    }
+}

ethexe/consensus/src/validator/coordinator.rs

@@ -18,7 +18,7 @@ use anyhow::{Context as _, Result, anyhow, ensure};
 use derive_more::Display;
 use ethexe_common::{
     Address, SimpleBlockData, ToDigest, ValidatorsVec, consensus::BatchCommitmentValidationRequest,
-    gear::BatchCommitment, network::ValidatorMessage,
+    ecdsa::PublicKey, gear::BatchCommitment, network::ValidatorMessage,
 };
 use futures::{FutureExt, future::BoxFuture};
 use gsigner::secp256k1::Secp256k1SignerExt;
@@ -42,6 +42,7 @@ pub struct CoordinatorBoot {
     ctx: ValidatorContext,
     block: SimpleBlockData,
     validators: ValidatorsVec,
+    pub_key: PublicKey,
     /// `Some` while we're either sleeping or awaiting the batch builder.
     pending: Option<BoxFuture<'static, Result<Option<BatchCommitment>>>>,
 }
@@ -59,6 +60,7 @@ impl CoordinatorBoot {
         ctx: ValidatorContext,
         block: SimpleBlockData,
         validators: ValidatorsVec,
+        pub_key: PublicKey,
     ) -> Result<ValidatorState> {
         let delay = ctx.core.coordinator_aggregation_delay;
         let batch_manager = ctx.core.batch_manager.clone();
@@ -75,6 +77,7 @@ impl CoordinatorBoot {
             ctx,
             block,
             validators,
+            pub_key,
             pending: Some(pending),
         }
         .into())
@@ -113,7 +116,13 @@ impl StateHandler for CoordinatorBoot {
                 Ok((Poll::Ready(()), next))
             }
             Poll::Ready(Ok(Some(batch))) => {
-                let next = Coordinator::create(self.ctx, self.validators, batch, self.block)?;
+                let next = Coordinator::create(
+                    self.ctx,
+                    self.validators,
+                    batch,
+                    self.block,
+                    self.pub_key,
+                )?;
                 Ok((Poll::Ready(()), next))
             }
         }
@@ -128,6 +137,7 @@ impl StateHandler for CoordinatorBoot {
 pub struct Coordinator {
     ctx: ValidatorContext,
     validators: BTreeSet<Address>,
+    pub_key: PublicKey,
     multisigned_batch: MultisignedBatchCommitment,
 }
 
@@ -174,6 +184,7 @@ impl Coordinator {
         validators: ValidatorsVec,
         batch: BatchCommitment,
         block: SimpleBlockData,
+        pub_key: PublicKey,
     ) -> Result<ValidatorState> {
         debug_assert_eq!(batch.block_hash, block.hash, "Block hash mismatch");
         ensure!(
@@ -190,7 +201,7 @@ impl Coordinator {
             batch,
             &ctx.core.signer,
             ctx.core.router_address,
-            ctx.core.pub_key,
+            pub_key,
         )?;
 
         ctx.core
@@ -210,16 +221,14 @@ impl Coordinator {
         let payload = BatchCommitmentValidationRequest::new(multisigned_batch.batch());
         let message = ValidatorMessage { era_index, payload };
 
-        let validation_request = ctx
-            .core
-            .signer
-            .signed_data(ctx.core.pub_key, message, None)?;
+        let validation_request = ctx.core.signer.signed_data(pub_key, message, None)?;
 
         ctx.output(ConsensusEvent::PublishMessage(validation_request.into()));
 
         Ok(Self {
             ctx,
             validators: validators.into_iter().collect(),
+            pub_key,
             multisigned_batch,
         }
         .into())

ethexe/consensus/src/validator/core.rs

@@ -23,7 +23,9 @@ use tokio::sync::RwLock;
 pub struct ValidatorCore {
     pub signatures_threshold: u64,
     pub router_address: Address,
-    pub pub_key: PublicKey,
+    /// `Some` only for nodes that hold a validator key.
+    /// `None` means watcher.
+    pub pub_key: Option<PublicKey>,
     pub timelines: ProtocolTimelines,
 
     #[debug(skip)]

ethexe/consensus/src/validator/idle.rs

@@ -18,6 +18,7 @@
 
 use super::{
     Participant, StateHandler, ValidatorContext, ValidatorState, coordinator::CoordinatorBoot,
+    watcher::Watcher,
 };
 use anyhow::{Context as _, Result, anyhow};
 use derive_more::{Debug, Display};
@@ -164,10 +165,14 @@ impl Idle {
             .block_coordinator_at(&validators, block.header.timestamp)
             .ok_or_else(|| anyhow!("cannot determine coordinator for block {}", block.hash))?;
 
-        if coordinator_addr == self.ctx.core.pub_key.to_address() {
-            CoordinatorBoot::start(self.ctx, block, validators)
-        } else {
-            Participant::create(self.ctx, block, coordinator_addr)
+        match self.ctx.core.pub_key {
+            Some(pub_key) if pub_key.to_address() == coordinator_addr => {
+                CoordinatorBoot::start(self.ctx, block, validators, pub_key)
+            }
+            Some(pub_key) if validators.iter().any(|v| *v == pub_key.to_address()) => {
+                Participant::create(self.ctx, block, coordinator_addr, pub_key)
+            }
+            _ => Watcher::create(self.ctx, block, coordinator_addr),
         }
     }
 }