newton-aggregator 0.4.13

//! On-chain registry view + reader trait.
//!
//! [`RegistryView`] is the value snapshot every state-commit tick reads
//! fresh: `(currentSequenceNo, currentStateRoot, lastCommitTimestamp)`.
//! [`RegistryReader`] is the trait the orchestrator depends on; the
//! production impl [`OnchainRegistryReader`] wraps a generated
//! [`StateCommitRegistryInstance`] and issues three `eth_call`s. Tests
//! substitute a hand-rolled fake — `Send + Sync + 'static` is preserved so
//! the orchestrator can hold the reader behind [`std::sync::Arc`].
//!
//! ## Eventual-consistency caveat
//!
//! [`OnchainRegistryReader::read_view`] issues THREE separate `eth_call`s
//! with no on-chain atomicity primitive. A commit landing between calls
//! produces an internally-incoherent view (e.g. `sequenceNo` reflects the
//! new commit but `stateRoot` is still the old root). This is intentionally
//! tolerated — the registry's typed reverts (`SequenceGap`,
//! `StateRootMismatch`, `TimestampRegression`) catch any incoherence at
//! submit time, and the orchestrator's poison handling (see
//! [`crate::state_commit::error::StateCommitError::is_poison`]) re-reads and
//! rebuilds. Convergence happens in at most one extra tick without needing
//! read-side atomicity.

use alloy::{
    network::{Ethereum, Network},
    primitives::B256,
    providers::Provider,
};
use async_trait::async_trait;
use newton_chainio::error::ChainIoError;
use newton_core::state_commit_registry::StateCommitRegistry::StateCommitRegistryInstance;

use crate::state_commit::error::{from_chainio, StateCommitError};

/// Snapshot of the on-chain `StateCommitRegistry` state at a single read.
///
/// `Copy` so the orchestrator can pass it through tick stages by value
/// without lifetime juggling.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RegistryView {
    /// `currentSequenceNo()` — sequence number of the most recently
    /// committed `StateCommit`. Next valid proposal must use `+ 1`.
    pub sequence_no: u64,
    /// `currentStateRoot()` — JMT root of the most recently committed
    /// `StateCommit`. Next valid proposal must use this as `prevStateRoot`.
    pub state_root: B256,
    /// `lastCommitTimestamp()` — `commit.timestamp` of the most recently
    /// committed `StateCommit`. Next valid proposal must strictly exceed
    /// this value (`commit.timestamp > lastCommitTimestamp`).
    pub last_commit_timestamp: u64,
}

/// Source of [`RegistryView`] snapshots for the state-commit orchestrator.
///
/// `Send + Sync + 'static` lets the orchestrator hold the reader behind
/// [`std::sync::Arc`] across `tokio::spawn` boundaries. The trait is
/// object-safe (via [`async_trait`]).
#[async_trait]
pub trait RegistryReader: Send + Sync + 'static {
    /// Read the registry view fresh — no caching at this layer.
    ///
    /// # Errors
    ///
    /// - [`StateCommitError::RegistryNotConfigured`] if the reader was
    ///   constructed without a valid registry address.
    /// - [`StateCommitError::OnchainCallFailed`] for transport-side failures
    ///   (RPC, timeout, wrong chain).
    /// - One of the eight typed `IStateRootCommittable` variants is NOT
    ///   produced by `read_view` — view calls don't revert with PDS errors.
    async fn read_view(&self) -> Result<RegistryView, StateCommitError>;
}

/// Production [`RegistryReader`] backed by a generated
/// [`StateCommitRegistryInstance`].
///
/// Generic over `P: Provider<N>` so the orchestrator can swap providers
/// (HTTP, WS, sharded) without touching this module. `chain_id` is stored
/// for diagnostic logging only — the underlying provider is already chain-bound.
pub struct OnchainRegistryReader<P, N = Ethereum>
where
    P: Provider<N> + Clone + 'static,
    N: Network,
{
    instance: StateCommitRegistryInstance<P, N>,
    chain_id: u64,
}

impl<P, N> OnchainRegistryReader<P, N>
where
    P: Provider<N> + Clone + 'static,
    N: Network,
{
    /// Construct a reader from a chain-bound `StateCommitRegistryInstance`.
    pub fn new(instance: StateCommitRegistryInstance<P, N>, chain_id: u64) -> Self {
        Self { instance, chain_id }
    }

    /// Returns the chain id this reader is bound to. Diagnostic only.
    pub fn chain_id(&self) -> u64 {
        self.chain_id
    }
}

impl<P, N> std::fmt::Debug for OnchainRegistryReader<P, N>
where
    P: Provider<N> + Clone + 'static,
    N: Network,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // The wrapped instance type doesn't impl Debug for arbitrary providers,
        // so we print only the diagnostic field.
        f.debug_struct("OnchainRegistryReader")
            .field("chain_id", &self.chain_id)
            .finish()
    }
}

#[async_trait]
impl<P, N> RegistryReader for OnchainRegistryReader<P, N>
where
    P: Provider<N> + Clone + 'static,
    N: Network,
{
    async fn read_view(&self) -> Result<RegistryView, StateCommitError> {
        // Three concurrent view calls. Eventual-consistency caveat unchanged
        // (see module rustdoc): a commit landing across the three calls still
        // produces an internally-incoherent view, but parallel issuance only
        // shrinks the window — it doesn't introduce a new failure mode. The
        // typed registry reverts at submit time still catch incoherence.
        //
        // Call builders must outlive the futures they spawn — otherwise the
        // try_join! macro drops them before polling begins.
        let seq_call = self.instance.currentSequenceNo();
        let root_call = self.instance.currentStateRoot();
        let ts_call = self.instance.lastCommitTimestamp();
        let (sequence_no, state_root, last_commit_timestamp) =
            tokio::try_join!(seq_call.call(), root_call.call(), ts_call.call())
                .map_err(|e| from_chainio(ChainIoError::ContractError(e)))?;

        Ok(RegistryView {
            sequence_no,
            state_root,
            last_commit_timestamp,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use tokio::sync::Mutex;

    /// Hand-rolled fake [`RegistryReader`]. Returns a fixed view by default;
    /// `set_view` swaps in a new one between calls; `set_error` makes the
    /// next call fail with [`StateCommitError::RegistryNotConfigured`].
    /// Per `.claude/skills/prover-testing-rust` we hand-roll this rather
    /// than reach for mockall.
    struct FakeRegistryReader {
        state: Arc<Mutex<FakeState>>,
    }

    #[derive(Clone)]
    struct FakeState {
        view: RegistryView,
        next_error: bool,
        call_count: usize,
    }

    impl FakeRegistryReader {
        fn new(view: RegistryView) -> Self {
            Self {
                state: Arc::new(Mutex::new(FakeState {
                    view,
                    next_error: false,
                    call_count: 0,
                })),
            }
        }

        async fn set_view(&self, view: RegistryView) {
            self.state.lock().await.view = view;
        }

        async fn set_next_error(&self) {
            self.state.lock().await.next_error = true;
        }

        async fn call_count(&self) -> usize {
            self.state.lock().await.call_count
        }
    }

    #[async_trait]
    impl RegistryReader for FakeRegistryReader {
        async fn read_view(&self) -> Result<RegistryView, StateCommitError> {
            let mut state = self.state.lock().await;
            state.call_count += 1;
            if state.next_error {
                state.next_error = false;
                return Err(StateCommitError::RegistryNotConfigured);
            }
            Ok(state.view)
        }
    }

    #[tokio::test]
    async fn fake_returns_initial_view() {
        let initial = RegistryView {
            sequence_no: 5,
            state_root: B256::repeat_byte(0xab),
            last_commit_timestamp: 1_000,
        };
        let reader = FakeRegistryReader::new(initial);
        let view = reader.read_view().await.expect("fake reader succeeds");
        assert_eq!(view, initial);
        assert_eq!(reader.call_count().await, 1);
    }

    #[tokio::test]
    async fn fake_reflects_view_updates_between_calls() {
        let initial = RegistryView {
            sequence_no: 5,
            state_root: B256::repeat_byte(0xab),
            last_commit_timestamp: 1_000,
        };
        let reader = FakeRegistryReader::new(initial);

        let v1 = reader.read_view().await.expect("first read");
        assert_eq!(v1.sequence_no, 5);

        let next = RegistryView {
            sequence_no: 6,
            state_root: B256::repeat_byte(0xcd),
            last_commit_timestamp: 2_000,
        };
        reader.set_view(next).await;

        let v2 = reader.read_view().await.expect("second read");
        assert_eq!(v2, next);
        assert_eq!(reader.call_count().await, 2);
    }

    #[tokio::test]
    async fn trait_is_object_safe_behind_arc() {
        let view = RegistryView {
            sequence_no: 0,
            state_root: B256::ZERO,
            last_commit_timestamp: 0,
        };
        let reader: Arc<dyn RegistryReader> = Arc::new(FakeRegistryReader::new(view));
        let read = reader.read_view().await.expect("fake reader succeeds");
        assert_eq!(read, view);
    }

    #[tokio::test]
    async fn fake_can_inject_errors_for_orchestrator_tests() {
        let reader = FakeRegistryReader::new(RegistryView {
            sequence_no: 0,
            state_root: B256::ZERO,
            last_commit_timestamp: 0,
        });
        reader.set_next_error().await;
        let err = reader.read_view().await.expect_err("error injected");
        assert!(matches!(err, StateCommitError::RegistryNotConfigured));
        // Subsequent calls succeed — error is one-shot.
        reader.read_view().await.expect("recovers after one-shot error");
    }

    #[test]
    fn registry_view_is_copy_and_eq() {
        let v1 = RegistryView {
            sequence_no: 1,
            state_root: B256::repeat_byte(0x01),
            last_commit_timestamp: 100,
        };
        let v2 = v1; // Copy
        assert_eq!(v1, v2);
        let v3 = RegistryView { sequence_no: 2, ..v1 };
        assert_ne!(v1, v3);
    }
}