newton-chainio 0.5.2

//! Operator registry service — source of truth for the BLS-cert path's view of operator membership.
//!
//! [`OperatorRegistryService`] owns the in-memory operator state DB
//! (ECDSA-key-bound, populated by background log subscriptions and a periodic
//! refresh) plus a per-chain [`tokio::sync::broadcast`] channel that re-emits
//! `OperatorRegistry::EpochAdvanced` events to gateway-side cache consumers.
//!
//! # Epoch consistency model (NEWT-1175)
//!
//! - Mutations to operator-set membership are queued on-chain and applied at
//!   epoch boundaries by permissionless `applyPendingChanges()`. The contract
//!   emits `EpochAdvanced(epoch, startBlock, durationBlocks)` per advance.
//! - This service forwards every `EpochAdvanced` log to subscribers via
//!   [`Self::subscribe_epoch_advanced`]. Each `subscribe()` call returns its
//!   own cursor; multiple consumers (routing-cache rebuild, BLS-cert cache
//!   purge, future watchers) do not couple.
//! - The broadcast channel has a 16-slot buffer. A consumer that falls more
//!   than 16 signals behind sees `RecvError::Lagged(n)` and must rely on the
//!   eventual-consistency backstop: the periodic operator-set refresh in
//!   [`Self::start_service`] (driven by [`query_registered_operator_and_fill_db`])
//!   reconverges within one epoch even if every consumer drops a signal.
//!
//! Lag is rare in steady state (epoch boundaries are minute-scale on every
//! supported chain), but the service must remain correct under it.
//!
//! Operator-set caches MUST rebuild on every `EpochAdvanced` signal. Skipping
//! a signal silently serves the pre-mutation operator set after the queued
//! change applies — routing decisions stay correct only if the rebuild fires.
//! See `docs/OPERATOR.md` "Off-chain cache coherence" for the wire-level
//! contract that the gateway's twin watchers honor.

use crate::{
    error::ChainIoError,
    registry::operator::{
        query_registered_operator_and_fill_db, EpochAdvancedSignal, OperatorSocket, OperatorStates,
        OperatorsInfoMessage, StateSource,
    },
};
use alloy::{
    primitives::{Address, Bytes, ChainId, FixedBytes},
    providers::Provider,
    rpc::types::Filter,
    sol_types::SolEvent,
};
use ark_ec::AffineRepr;
use async_trait::async_trait;
use eigensdk::{
    client_avsregistry::error::AvsRegistryError,
    common::{get_provider, get_ws_provider, NEW_PUBKEY_REGISTRATION_EVENT, OPERATOR_SOCKET_UPDATE},
    crypto_bls::{alloy_registry_g1_point_to_g1_affine, alloy_registry_g2_point_to_g2_affine, BlsG1Point, BlsG2Point},
    services_avsregistry::AvsRegistryService,
    services_operatorsinfo::{operator_info::OperatorInfoService, operatorsinfo_inmemory::OperatorInfoServiceError},
    types::{
        avs_state::{OperatorAvsState, QuorumAvsState},
        operator::{operator_id_from_g1_pub_key, OperatorId, OperatorPubKeys, OperatorTypesError},
    },
    utils::slashing::middleware::{
        bls_apk_registry::{
            BLSApkRegistry,
            BN254::{G1Point, G2Point},
        },
        operator_state_retriever::OperatorStateRetriever::CheckSignaturesIndices,
        registry_coordinator::RegistryCoordinator,
        socket_registry::SocketRegistry,
        stake_registry::{StakeRegistry, StakeRegistry::OperatorStakeUpdate},
    },
};
use eyre::Result;
use futures_util::{future::join_all, StreamExt};
use newton_core::{
    common::{
        address::{get_bls_apk_registry_address, get_newton_operator_registry, get_socket_registry_address},
        chain::{get_block_time_ms, get_chain_id, get_epoch_blocks},
    },
    operator_registry::OperatorRegistry,
    operator_registry_epoch_governance::OperatorRegistryEpochGovernance,
};
use std::{collections::HashMap, net::SocketAddr, sync::Arc};
use thiserror::Error;
use tokio::sync::{
    broadcast,
    mpsc::{self, UnboundedSender},
    oneshot::{self, Sender},
    RwLock,
};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, instrument, warn};

/// Module for handling operator-related functionalities.
pub mod operator;

/// Fetches operator information from the registry.
/// Loads and stores operators info (addresses and public key) in memory.
#[derive(Debug, Clone)]
pub struct OperatorRegistryService {
    /// RPC URL for contract calls
    http_rpc_url: String,
    /// WebSocket URL for contract events
    ws_rpc_url: String,
    /// Operator Registry contract address
    operator_registry_address: Address,
    /// Operator State Retriever contract address.
    ///
    /// NEWT-1715: the periodic operator-set refresh enumerates the on-chain
    /// **registered** operator set through this singleton rather than the
    /// whitelist, so de-whitelisted-but-still-registered operators stay cached.
    operator_state_retriever_address: Address,
    /// Socket Registry contract address
    socket_registry_address: Address,
    /// Stake Registry contract address
    stake_registry_address: Address,
    /// Chain ID
    chain_id: ChainId,
    /// Operators info message channel
    pub_keys: UnboundedSender<OperatorsInfoMessage>,
    /// Broadcast publisher for `OperatorRegistry::EpochAdvanced` events. Lossy by design — the
    /// periodic `query_registered_operator_and_fill_db` refresh below reconverges within one
    /// epoch even if every consumer is lagged. Capacity 16 absorbs the rare case of multiple
    /// rapid `applyPendingChanges` boundary crossings without forcing slow consumers to drop.
    epoch_advanced_tx: broadcast::Sender<EpochAdvancedSignal>,
    /// Whether a non-zero `quorumCount` has ever been observed. NEWT-1715: once true,
    /// a later zero read is treated as no-signal (cache untouched) rather than an
    /// authoritative empty snapshot that would wipe the cache. Bootstrap-only zero.
    observed_quorums: Arc<std::sync::atomic::AtomicBool>,
    /// Operator states
    pub operator_states: OperatorStates,
}
#[async_trait]
impl OperatorInfoService for OperatorRegistryService {
    async fn get_operator_info(&self, address: Address) -> Result<Option<OperatorPubKeys>, OperatorInfoServiceError> {
        let (responder_tx, responder_rx) = oneshot::channel();
        debug!("[OperatorRegistryService] get_operator_info for {address}");
        let _ = self
            .pub_keys
            .send(OperatorsInfoMessage::GetPubKeys(address, responder_tx));
        Ok(responder_rx.await?)
    }

    async fn get_operator_socket(&self, address: Address) -> Result<Option<String>, OperatorInfoServiceError> {
        let (responder_tx, responder_rx) = oneshot::channel();
        debug!("[OperatorRegistryService] get_operator_socket for {address}");
        let _ = self
            .pub_keys
            .send(OperatorsInfoMessage::GetSockets(address, responder_tx));
        Ok(responder_rx.await?)
    }
}

impl OperatorRegistryService {
    /// Returns the chain id this service is bound to.
    ///
    /// Used by `EpochAdvanced` consumers to tag per-chain observability metrics
    /// even when no signal payload is available (e.g., the `RecvError::Lagged`
    /// arm of a broadcast subscription, where the dropped messages are still
    /// known to originate from this watcher's per-chain channel).
    pub fn chain_id(&self) -> ChainId {
        self.chain_id
    }

    /// Creates a new operator info service given a logger, an avs registry chain reader and rpc endpoints.
    ///
    /// # Arguments
    ///
    /// * `operator_registry_address` - An operator registry address.
    /// * `operator_state_retriever_address` - The operator state retriever address (used to
    ///   enumerate the on-chain registered operator set during refresh).
    /// * `socket_registry_address` - A socket registry address.
    /// * `http_rpc_url` - A http rpc url.
    /// * `ws_rpc_url` - A websocket rpc url.
    ///
    /// # Returns
    ///
    /// A tuple of 2 elements:
    /// [`Self`] and [`mpsc::UnboundedReceiver<OperatorInfoServiceError>`] if successful, else [`OperatorInfoServiceError`]
    pub async fn new(
        operator_registry_address: Address,
        operator_state_retriever_address: Address,
        socket_registry_address: Address,
        http_rpc_url: String,
        ws_rpc_url: String,
    ) -> Result<(Self, mpsc::UnboundedReceiver<OperatorInfoServiceError>), OperatorInfoServiceError> {
        // Create provider using eigensdk helper to validate connection
        let provider = get_provider(&http_rpc_url);

        // Get chain ID
        let chain_id = provider.get_chain_id().await?;
        debug!("[OperatorRegistryService] connected to chain id: {}", chain_id);

        let operator_registry = OperatorRegistry::new(operator_registry_address, provider.clone());
        let stake_registry_address = operator_registry
            .stakeRegistry()
            .call()
            .await
            .map_err(OperatorInfoServiceError::AlloyContractError)?;

        let (pubkeys_tx, mut pubkeys_rx) = mpsc::unbounded_channel();
        let (error_tx, error_rx) = mpsc::unbounded_channel();
        let (epoch_advanced_tx, _) = broadcast::channel::<EpochAdvancedSignal>(16);

        let operator_states = OperatorStates {
            operator_info_data: Arc::new(RwLock::new(HashMap::new())),
            operator_addr_to_id: Arc::new(RwLock::new(HashMap::new())),
            socket_dict: Arc::new(RwLock::new(HashMap::new())),
            operator_stake: Arc::new(RwLock::new(HashMap::new())),
            total_stake: Arc::new(RwLock::new(HashMap::new())),
        };

        // Spawn a detached task for processing commands
        tokio::spawn({
            let operator_states = operator_states.clone();
            let error_tx = error_tx.clone();
            async move {
                while let Some(cmd) = pubkeys_rx.recv().await {
                    if let Err(e) = async {
                        match cmd {
                            OperatorsInfoMessage::InsertOperatorInfo(addr, keys, socket_info, state_src) => {
                                debug!(
                                    "[OperatorRegistryService] received insert operator info message: state_src: {}",
                                    state_src
                                );
                                if let (Some(addr), Some(keys)) = (addr, keys) {
                                    let mut data = operator_states.operator_info_data.write().await;
                                    match data.get(&addr) {
                                        // Prevent Historic data from overwriting Event data.
                                        Some((entry_src, _))
                                            if *entry_src == StateSource::Event
                                                && state_src == StateSource::Historic => {}
                                        _ => {
                                            debug!("[operator_info_data] operator keys for address: {}", addr);
                                            data.insert(addr, (state_src.clone(), *keys.clone()));
                                        }
                                    }

                                    let operator_id = operator_id_from_g1_pub_key(keys.g1_pub_key)?;

                                    let mut id_map = operator_states.operator_addr_to_id.write().await;
                                    match id_map.get(&addr) {
                                        // Prevent Historic data from overwriting Event data.
                                        Some((entry_src, _))
                                            if *entry_src == StateSource::Event
                                                && state_src == StateSource::Historic => {}
                                        _ => {
                                            debug!(
                                                "[operator_info_data] operator id for address: {} operator_id: {}",
                                                addr, operator_id
                                            );
                                            id_map.insert(addr, (state_src.clone(), operator_id));
                                        }
                                    }
                                }
                                let mut socket_data = operator_states.socket_dict.write().await;
                                if let Some(socket) = socket_info {
                                    // Prevent Historic data from overwriting Event data.
                                    match socket_data.get(&FixedBytes(*socket.id)) {
                                        Some((entry_src, _))
                                            if *entry_src == StateSource::Event
                                                && state_src == StateSource::Historic => {}
                                        _ => {
                                            debug!(
                                                "[operator_info_data] socket for operator id: {} socket: {}",
                                                socket.id, socket.socket
                                            );
                                            socket_data.insert(FixedBytes(*socket.id), (state_src, socket.socket));
                                        }
                                    }
                                }
                            }
                            OperatorsInfoMessage::InsertOperatorStake(operator_id, quorum_number, stake, block) => {
                                // Block-versioned last-writer-wins (NEWT-1715): a single-quorum
                                // event applies only if it is at least as new as the cached value.
                                debug!(
                                    "[operator_stake] operator id: {} quorum {} stake {} @ block {}",
                                    operator_id, quorum_number, stake, block
                                );
                                let mut data = operator_states.operator_stake.write().await;
                                let entry = data.entry(operator_id).or_default();
                                operator::apply_versioned_stake(entry, quorum_number, stake, block);
                            }
                            OperatorsInfoMessage::InsertTotalStake(quorum_number, stake, block) => {
                                debug!(
                                    "[operator_total_stake] quorum {} stake {} @ block {}",
                                    quorum_number, stake, block
                                );
                                let mut data = operator_states.total_stake.write().await;
                                operator::apply_versioned_stake(&mut data, quorum_number, stake, block);
                            }
                            OperatorsInfoMessage::ReplaceOperatorStake(operator_id, stakes, block) => {
                                // NEWT-1715: reconcile an operator's whole per-quorum stake map
                                // against the block-pinned snapshot. Per quorum the newer block
                                // wins, so a quorum that dropped to zero is cleared unless a
                                // strictly-newer event superseded the snapshot — no sticky
                                // source label that would let one event immunize a quorum forever.
                                debug!(
                                    "[operator_stake] reconciling stake map for operator id: {} with {} quorum(s) @ block {}",
                                    operator_id,
                                    stakes.len(),
                                    block
                                );
                                let mut data = operator_states.operator_stake.write().await;
                                let entry = data.entry(operator_id).or_default();
                                operator::reconcile_snapshot(entry, stakes, block);
                                // An operator whose stake map became empty carries no weight; drop
                                // the entry so it doesn't linger as an empty map.
                                if entry.is_empty() {
                                    data.remove(&operator_id);
                                }
                            }
                            OperatorsInfoMessage::ReplaceTotalStakes(total_stakes, block) => {
                                // NEWT-1715: reconcile the whole per-quorum total-stake view against
                                // the block-pinned snapshot, block-versioned. The snapshot is the
                                // authoritative full view at `block`; a quorum is only kept from the
                                // cache if it carries a strictly-newer event.
                                debug!(
                                    "[operator_total_stake] reconciling {} quorum total(s) @ block {}",
                                    total_stakes.len(),
                                    block
                                );
                                let mut data = operator_states.total_stake.write().await;
                                operator::reconcile_snapshot(&mut data, total_stakes, block);
                            }
                            OperatorsInfoMessage::Remove(addr) => {
                                debug!(
                                    "[OperatorRegistryService] received remove operator info message. address: {}",
                                    addr
                                );
                                // Symmetric prune across both address-keyed and
                                // operator_id-keyed state. Downstream readers
                                // (routing pool via `get_operator_id_to_addr_mappings`,
                                // BLS-cert path via `get_operator_info`, socket
                                // resolver, off-chain quorum accumulator) must
                                // observe one membership snapshot. Octane #46:
                                // previously only address-keyed maps were pruned
                                // here, so `socket_dict` and `operator_stake`
                                // (operator_id-keyed) leaked stale entries —
                                // re-registration with the same BLS key reused
                                // a stale socket; missed stake-drop events
                                // inflated off-chain quorum.
                                //
                                // Lock acquisition order matches the
                                // InsertOperatorInfo path above so writers
                                // can't deadlock; each guard is block-scoped.
                                let operator_id_opt = operator_states
                                    .operator_addr_to_id
                                    .read()
                                    .await
                                    .get(&addr)
                                    .map(|(_, id)| *id);
                                {
                                    let mut data = operator_states.operator_info_data.write().await;
                                    data.remove(&addr);
                                }
                                {
                                    let mut id_map = operator_states.operator_addr_to_id.write().await;
                                    id_map.remove(&addr);
                                }
                                if let Some(operator_id) = operator_id_opt {
                                    {
                                        let mut sockets = operator_states.socket_dict.write().await;
                                        sockets.remove(&operator_id);
                                    }
                                    {
                                        let mut stakes = operator_states.operator_stake.write().await;
                                        stakes.remove(&operator_id);
                                    }
                                }
                            }
                            OperatorsInfoMessage::GetPubKeys(addr, responder) => {
                                debug!(
                                    "[OperatorRegistryService] received get public keys message. address: {}",
                                    addr
                                );
                                let data = operator_states.operator_info_data.read().await;
                                let _ = match data.get(&addr).cloned() {
                                    Some(result) => responder.send(Some(result.1)),
                                    None => {
                                        warn!("[get_pub_keys] operator pubkey not found for address: {}", addr);
                                        responder.send(None)
                                    }
                                };
                            }
                            OperatorsInfoMessage::GetSockets(addr, responder) => {
                                debug!("[OperatorRegistryService] received get sockets message: {}", addr);
                                let operator_id = operator_states.operator_addr_to_id.read().await.get(&addr).cloned();
                                if let Some((_, id)) = operator_id {
                                    let _ = match operator_states.socket_dict.read().await.get(&id).cloned() {
                                        Some((_, socket)) => {
                                            debug!("[get_sockets] socket found for operator {addr}: {}", socket);
                                            responder.send(Some(socket.to_string()))
                                        }
                                        None => {
                                            warn!("[get_sockets] socket not found for operator {addr}");
                                            responder.send(None)
                                        }
                                    };
                                }
                            }
                            OperatorsInfoMessage::GetOperatorStake(operator_id, quorum_number, responder) => {
                                debug!(
                                    "[OperatorRegistryService] received get operator stake message. operator_id: {} quorum_number: {}",
                                    operator_id, quorum_number
                                );
                                let data = operator_states.operator_stake.read().await;
                                let _ = match data
                                    .get(&operator_id)
                                    .and_then(|stakes| stakes.get(&quorum_number))
                                    .map(|(stake, _block)| *stake)
                                {
                                    Some(stake) => {
                                        debug!("[get_operator_stake] stake found for operator {operator_id} quorum {quorum_number}: {stake}");
                                        responder.send(Some(stake))
                                    }
                                    None => {
                                        warn!("[get_operator_stake] stake not found for operator {operator_id} quorum {quorum_number}");
                                        responder.send(None)
                                    }
                                };
                            }
                            OperatorsInfoMessage::GetTotalStake(quorum_number, responder) => {
                                debug!(
                                    "[OperatorRegistryService] received get total stake message. quorum_number: {}",
                                    quorum_number
                                );
                                let data = operator_states.total_stake.read().await;
                                let _ = match data.get(&quorum_number).map(|(stake, _block)| *stake) {
                                    Some(stake) => {
                                        debug!("[get_total_stake] total stake found for quorum {quorum_number}: {stake}");
                                        responder.send(Some(stake))
                                    }
                                    None => {
                                        warn!("[get_total_stake] total stake not found for quorum {quorum_number}");
                                        responder.send(None)
                                    }
                                };
                            }
                        }
                        Ok::<(), OperatorInfoServiceError>(())
                    }
                    .await
                    {
                        // Send the error to the error channel
                        let _ = error_tx.send(e);
                    }
                }
            }
        });

        Ok((
            Self {
                operator_registry_address,
                operator_state_retriever_address,
                socket_registry_address,
                stake_registry_address,
                http_rpc_url,
                ws_rpc_url,
                chain_id,
                pub_keys: pubkeys_tx,
                epoch_advanced_tx,
                observed_quorums: Arc::new(std::sync::atomic::AtomicBool::new(false)),
                operator_states,
            },
            error_rx,
        ))
    }

    /// Creates an `OperatorRegistryService` with safe dummy values for testing.
    ///
    /// This avoids the undefined behavior of `MaybeUninit::zeroed().assume_init()`
    /// on a struct containing `String`, `UnboundedSender`, and other non-zero types.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn new_for_test(operator_states: OperatorStates) -> Self {
        let (pubkeys_tx, _pubkeys_rx) = tokio::sync::mpsc::unbounded_channel();
        let (epoch_advanced_tx, _) = broadcast::channel::<EpochAdvancedSignal>(16);
        Self {
            http_rpc_url: String::new(),
            ws_rpc_url: String::new(),
            operator_registry_address: Address::ZERO,
            operator_state_retriever_address: Address::ZERO,
            socket_registry_address: Address::ZERO,
            stake_registry_address: Address::ZERO,
            chain_id: 0,
            pub_keys: pubkeys_tx,
            epoch_advanced_tx,
            observed_quorums: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            operator_states,
        }
    }

    /// Starts the operator info service.
    ///
    /// # Arguments
    ///
    /// * `cancellation_token` - A cancellation token than can be used to stop the service.
    ///
    /// # Returns
    ///
    /// Ok(()) if successful, otherwise an error.
    #[instrument(skip_all)]
    pub async fn start_service(&self, cancellation_token: &CancellationToken) -> eyre::Result<()> {
        info!("[OperatorRegistryService] starting operator info service");
        // Run asynchonous thread querying past operator registrations
        let ws_rpc_url = self.ws_rpc_url.clone();
        let pub_keys = self.pub_keys.clone();
        let (tx, mut rx) = mpsc::channel(1);
        let block_time_ms = get_block_time_ms(&self.http_rpc_url)
            .await
            .map_err(|err| eyre::eyre!(err))?;
        let epoch_blocks = get_epoch_blocks(&self.http_rpc_url)
            .await
            .map_err(|err| eyre::eyre!(err))?;
        const FALLBACK_REFRESH_SECS: u64 = 86_400;
        let refresh_interval = if epoch_blocks == 0 {
            tracing::warn!(
                "OperatorRegistry.epochDurationBlocks() == 0 — epochs not yet initialized. \
                 Using 24h fallback refresh interval. Run `initializeEpochs` when ready."
            );
            std::time::Duration::from_secs(FALLBACK_REFRESH_SECS)
        } else {
            std::time::Duration::from_secs((epoch_blocks as u64 * block_time_ms) / 1000)
        };

        debug!(
            "[OperatorRegistryService] registry reloads every {} blocks ({} seconds)",
            epoch_blocks,
            refresh_interval.as_secs()
        );

        let operator_registry_address = self.operator_registry_address;
        let operator_state_retriever_address = self.operator_state_retriever_address;
        let http_rpc_url = self.http_rpc_url.clone();
        let pub_keys_clone = pub_keys.clone();
        let states_for_refresh = self.operator_states.clone();
        let observed_quorums = self.observed_quorums.clone();

        let handle = tokio::spawn(async move {
            let mut interval = tokio::time::interval(refresh_interval);
            interval.tick().await; // Skip first immediate tick

            loop {
                debug!("[OperatorRegistryService] triggering periodic operator registry refresh");
                let res = query_registered_operator_and_fill_db(
                    operator_registry_address,
                    operator_state_retriever_address,
                    http_rpc_url.clone(),
                    pub_keys_clone.clone(),
                    &states_for_refresh,
                    &observed_quorums,
                )
                .await;
                let _ = tx.send(res).await;

                interval.tick().await;
            }
        });

        let provider = get_ws_provider(&self.ws_rpc_url).await?;
        let current_block_number = provider.get_block_number().await?;

        // Subscribe to new pubkey registration events
        let new_pubkey_registration_filter = Filter::new()
            .event(NEW_PUBKEY_REGISTRATION_EVENT)
            .from_block(current_block_number);

        let operator_socket_update_filter = Filter::new()
            .event(OPERATOR_SOCKET_UPDATE)
            .from_block(current_block_number);

        let subcription_new_operator_registration_stream =
            provider.subscribe_logs(&new_pubkey_registration_filter).await?;
        let subscription_operator_socket_update_filter =
            provider.subscribe_logs(&operator_socket_update_filter).await?;

        // Subscribe to operator stake update events
        let operator_stake_update_filter = Filter::new()
            .event(OperatorStakeUpdate::SIGNATURE)
            .address(self.stake_registry_address)
            .from_block(current_block_number);

        let subscription_operator_stake_update_filter = provider.subscribe_logs(&operator_stake_update_filter).await?;

        // Subscribe to EpochAdvanced events from `OperatorRegistryEpochGovernance`
        // (the sibling contract that owns the epoch state machine after the
        // EIP-170 split — pre-split this lived on `OperatorRegistry`). The
        // governance address is read once at startup from the registry's
        // one-shot `epochGovernance()` pointer; address-bound to disambiguate
        // from any same-named events on other contracts.
        let registry_for_gov_lookup = OperatorRegistry::new(self.operator_registry_address, &provider);
        let epoch_governance_address = registry_for_gov_lookup.epochGovernance().call().await?;
        let epoch_advanced_filter = Filter::new()
            .event(OperatorRegistryEpochGovernance::EpochAdvanced::SIGNATURE)
            .address(epoch_governance_address)
            .from_block(current_block_number);

        let subscription_epoch_advanced_filter = provider.subscribe_logs(&epoch_advanced_filter).await?;

        let mut new_operator_registration_stream = subcription_new_operator_registration_stream.into_stream().fuse();
        let mut operator_socket_update_stream = subscription_operator_socket_update_filter.into_stream().fuse();
        let mut operator_stake_update_stream = subscription_operator_stake_update_filter.into_stream().fuse();
        let mut epoch_advanced_stream = subscription_epoch_advanced_filter.into_stream().fuse();
        let pub_keys = self.pub_keys.clone();
        let epoch_advanced_tx = self.epoch_advanced_tx.clone();
        let chain_id = self.chain_id;

        let stake_registry = StakeRegistry::new(self.stake_registry_address, get_provider(&self.http_rpc_url));

        loop {
            tokio::select! {
                _ = cancellation_token.cancelled() => {
                    info!("[OperatorRegistryService] cancellation signal received, stopping the stream.");
                    handle.abort();
                    break;
                },
                res = rx.recv() => {
                    match res {
                        Some(Err(err)) => {
                            error!("Failed to query registered operator: {err:?}");
                            return Err(err);
                        }
                        _ => continue,
                    }
                },
                log = new_operator_registration_stream.next() => {
                    match log {
                        Some(log) => {

                            let data = log
                                .log_decode::<BLSApkRegistry::NewPubkeyRegistration>()
                                .ok();

                            if let Some(new_pub_key_event) = data {
                                let event_data = new_pub_key_event.data();
                                let operator_pub_key = OperatorPubKeys {
                                    g1_pub_key: BlsG1Point::new(alloy_registry_g1_point_to_g1_affine(
                                        G1Point {
                                            X: event_data.pubkeyG1.X,
                                            Y: event_data.pubkeyG1.Y,
                                        },
                                    )),
                                    g2_pub_key: BlsG2Point::new(alloy_registry_g2_point_to_g2_affine(
                                        G2Point {
                                            X: event_data.pubkeyG2.X,
                                            Y: event_data.pubkeyG2.Y,
                                        },
                                    )),
                                };
                                // Send message

                                debug!(
                                    "[OperatorRegistryService] new pub key found operator_address: {}",
                                    event_data.operator,
                                );

                                let _ = pub_keys.send(OperatorsInfoMessage::InsertOperatorInfo(
                                    Some(event_data.operator),
                                    Some(Box::new(operator_pub_key)),
                                    None,
                                    StateSource::Event
                                ));
                            }
                        },
                        None => {
                            break;
                        }
                    }
                },

                log =operator_socket_update_stream.next() =>{

                    match log {
                        Some(log) => {

                            let data = log
                                .log_decode::<RegistryCoordinator::OperatorSocketUpdate>()
                                .ok();

                            if let Some(operator_socket_update_event) = data {
                                let event_data = operator_socket_update_event.data();
                                let operator_socket = OperatorSocket {
                                    id: event_data.operatorId,
                                    socket: event_data.socket.clone()
                                };
                                // Send message

                                debug!(
                                    "[OperatorRegistryService] received new socket registration event operator_id: {}, socket: {}",
                                    event_data.operatorId, event_data.socket
                                );

                                let _ = pub_keys.send(OperatorsInfoMessage::InsertOperatorInfo(
                                    None,
                                    None,
                                    Some(OperatorSocket{socket:operator_socket.socket , id:operator_socket.id }),
                                    StateSource::Event
                                ));
                            }
                        },
                        None => {
                            break;
                        }
                    }

                },

                log = operator_stake_update_stream.next() => {
                    match log {
                        Some(log) => {
                            let data = log.log_decode::<OperatorStakeUpdate>().ok();

                            if let Some(stake_update_event) = data {
                                let event_data = stake_update_event.data();
                                let operator_id = event_data.operatorId.0;
                                let quorum_number = event_data.quorumNumber;
                                let stake = event_data.stake;
                                let stake = stake.to::<u128>();
                                // Block-versioned last-writer-wins (NEWT-1715): stamp the event
                                // with the block it was emitted at so the receiver reconciles it
                                // against the periodic snapshot by block, not by a sticky source
                                // label. A log without a block number can't be ordered; skip it
                                // (the periodic refresh reconverges the cache).
                                let Some(event_block) = log.block_number else {
                                    warn!("[OperatorRegistryService] OperatorStakeUpdate log missing block number; skipping (periodic refresh will reconcile)");
                                    continue;
                                };
                                debug!(
                                    "[OperatorRegistryService] received operator stake update event operator_id: {}, quorum: {}, stake: {} @ block {}",
                                    hex!(operator_id), quorum_number, stake, event_block
                                );

                                let _ = pub_keys.send(OperatorsInfoMessage::InsertOperatorStake(
                                    operator_id.into(),
                                    quorum_number,
                                    stake,
                                    event_block,
                                ));

                                // Also fetch and update total stake for this quorum, pinned to the
                                // event block so the value carries a correct version stamp.
                                // We spawn a task to avoid blocking the loop.
                                let stake_registry = stake_registry.clone();
                                let pub_keys = pub_keys.clone();
                                tokio::spawn(async move {
                                    match stake_registry
                                        .getCurrentTotalStake(quorum_number)
                                        .block(alloy::eips::BlockId::from(event_block))
                                        .call()
                                        .await
                                    {
                                        Ok(total_stake) => {
                                            let total_stake_u128 = total_stake.to::<u128>();
                                            debug!(
                                                "[OperatorRegistryService] fetched updated total stake for quorum {} @ block {}: {}",
                                                quorum_number, event_block, total_stake_u128
                                            );
                                            let _ = pub_keys.send(OperatorsInfoMessage::InsertTotalStake(
                                                quorum_number,
                                                total_stake_u128,
                                                event_block,
                                            ));
                                        }
                                        Err(e) => {
                                            error!(
                                                "[OperatorRegistryService] failed to fetch updated total stake for quorum {}: {}",
                                                quorum_number, e
                                            );
                                        }
                                    }
                                });
                            }
                        },
                        None => {
                            break;
                        }
                    }
                },

                log = epoch_advanced_stream.next() => {
                    match log {
                        Some(log) => {
                            let data = log.log_decode::<OperatorRegistryEpochGovernance::EpochAdvanced>().ok();

                            if let Some(epoch_advanced_event) = data {
                                let event_data = epoch_advanced_event.data();
                                let signal = EpochAdvancedSignal {
                                    chain_id,
                                    epoch: event_data.epoch,
                                    start_block: event_data.startBlock,
                                    duration_blocks: event_data.durationBlocks,
                                };
                                debug!(
                                    "[OperatorRegistryService] EpochAdvanced epoch={} start_block={} duration_blocks={}",
                                    signal.epoch, signal.start_block, signal.duration_blocks
                                );
                                // Lossy-by-design: a `SendError` means no live subscribers; the
                                // periodic refresh keeps consumers eventually consistent.
                                let _ = epoch_advanced_tx.send(signal);
                            }
                        },
                        None => {
                            break;
                        }
                    }
                }
            }
        }

        Ok(())
    }

    /// Returns the operator ID to address mappings for cache warming.
    ///
    /// This allows pre-populating caches in other services (like AvsRegistryServiceChainCaller)
    /// with the already-loaded operator data, eliminating RPC lookups on first request.
    ///
    /// # Returns
    ///
    /// Vector of (operator_id, address) tuples for all known operators
    pub async fn get_operator_id_to_addr_mappings(&self) -> Vec<(FixedBytes<32>, Address)> {
        let addr_to_id = self.operator_states.operator_addr_to_id.read().await;
        addr_to_id
            .iter()
            .map(|(addr, (_, operator_id))| (FixedBytes(**operator_id), *addr))
            .collect()
    }

    /// Returns the number of operators currently loaded in the service.
    pub async fn operator_count(&self) -> usize {
        self.operator_states.operator_addr_to_id.read().await.len()
    }

    /// Per-quorum stake held by registered operators that are **not routable** —
    /// i.e. cached (registered, non-zero id) but with a missing or empty socket.
    ///
    /// NEWT-1715: the registered set is a superset of the routable set. Such
    /// operators still count toward `total_stake` (the quorum denominator) but can
    /// never contribute `signed_stake` (the numerator), because the gateway skips
    /// them when building the broadcast pool. Surfacing this per-quorum stake in
    /// the `QuorumNotReached` diagnostics makes the resulting denominator>numerator
    /// gap observable instead of presenting as an unexplained timeout.
    ///
    /// Returns a map of quorum number → summed unroutable stake (only quorums with
    /// a non-zero amount are included).
    pub async fn unroutable_stake_by_quorum(&self) -> HashMap<u8, u128> {
        let addr_to_id = self.operator_states.operator_addr_to_id.read().await;
        let sockets = self.operator_states.socket_dict.read().await;
        let stakes = self.operator_states.operator_stake.read().await;

        let mut unroutable: HashMap<u8, u128> = HashMap::new();
        for (_addr, (_, operator_id)) in addr_to_id.iter() {
            let routable = sockets.get(operator_id).is_some_and(|(_, socket)| !socket.is_empty());
            if routable {
                continue;
            }
            if let Some(quorum_stakes) = stakes.get(operator_id) {
                for (quorum_number, (stake, _block)) in quorum_stakes {
                    *unroutable.entry(*quorum_number).or_default() += *stake;
                }
            }
        }
        unroutable.retain(|_, stake| *stake > 0);
        unroutable
    }

    /// Total stake (summed across all quorums) cached for a single operator id.
    ///
    /// Used at pool-build time to escalate the logging when a registered operator is
    /// skipped for a missing/empty socket: a skipped operator that holds stake is a
    /// liveness problem (its stake counts toward `total_stake` but can never sign),
    /// so the skip must be surfaced loudly rather than swallowed silently.
    pub async fn operator_total_stake(&self, operator_id: &OperatorId) -> u128 {
        self.operator_states
            .operator_stake
            .read()
            .await
            .get(operator_id)
            .map(|quorum_stakes| quorum_stakes.values().map(|(stake, _block)| *stake).sum())
            .unwrap_or(0)
    }

    /// The set of quorum numbers currently known to the cache (from `total_stake`).
    ///
    /// Used by the readiness loop to reset per-quorum gauges (e.g. unroutable stake)
    /// to `0` before re-emitting the non-zero values, so a quorum that recovers
    /// clears its stale gauge value instead of lingering on dashboards.
    pub async fn known_quorums(&self) -> Vec<u8> {
        self.operator_states.total_stake.read().await.keys().copied().collect()
    }

    /// Per-quorum total registered stake (the quorum denominator).
    ///
    /// Emitted as a metric alongside `unroutable_stake_by_quorum` so the alerting
    /// layer can compute the unroutable/total ratio per quorum.
    pub async fn total_stake_by_quorum(&self) -> HashMap<u8, u128> {
        self.operator_states
            .total_stake
            .read()
            .await
            .iter()
            .map(|(quorum_number, (stake, _block))| (*quorum_number, *stake))
            .collect()
    }

    /// Subscribes to `OperatorRegistry::EpochAdvanced` events emitted by this registry.
    ///
    /// The returned receiver yields one `EpochAdvancedSignal` per event observed by the
    /// background subscription loop in [`Self::start_service`]. Subscribers may **lag** —
    /// `tokio::sync::broadcast` drops the oldest signal once a receiver falls more than 16
    /// signals behind. The periodic operator-set refresh in `start_service` (driven by
    /// [`query_registered_operator_and_fill_db`]) reconverges within one epoch even if
    /// every consumer drops a signal — it is the eventual-consistency backstop, not a
    /// substitute for primary correctness on the broadcast path.
    ///
    /// # Returns
    ///
    /// A fresh `broadcast::Receiver` joined to the live event stream. Multiple consumers
    /// (e.g., gateway-side `OperatorPool` rebuild watcher, metrics reporter) may subscribe
    /// independently; each receives a copy of every signal until lag.
    pub fn subscribe_epoch_advanced(&self) -> broadcast::Receiver<EpochAdvancedSignal> {
        self.epoch_advanced_tx.subscribe()
    }

    /// Loads operators synchronously at startup for cache warming.
    ///
    /// This method fetches all registered operators from the chain and populates
    /// the in-memory data structures. Call this before `start_service()` to ensure
    /// the first request has warm caches.
    ///
    /// # Returns
    ///
    /// Ok(count) with number of operators loaded, or error if load fails
    pub async fn load_operators_sync(&self) -> eyre::Result<usize> {
        debug!("[OperatorRegistryService] loading operators synchronously for cache warming...");
        let load_start = std::time::Instant::now();

        query_registered_operator_and_fill_db(
            self.operator_registry_address,
            self.operator_state_retriever_address,
            self.http_rpc_url.clone(),
            self.pub_keys.clone(),
            &self.operator_states,
            &self.observed_quorums,
        )
        .await?;

        // Give the message processor time to handle the messages
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;

        let count = self.operator_count().await;
        debug!(
            "[OperatorRegistryService] loaded {} operators in {} ms",
            count,
            load_start.elapsed().as_millis()
        );
        Ok(count)
    }

    /// Verifies cached G2 keys against fresh on-chain reads.
    ///
    /// This diagnostic method compares the G2 keys stored in cache against
    /// what's currently stored on-chain in the BLSApkRegistry. Any mismatches
    /// are logged as errors, which can help diagnose BN254 pairing failures.
    ///
    /// # Arguments
    ///
    /// * `operator_addresses` - List of operator addresses to verify
    ///
    /// # Returns
    ///
    /// Ok(true) if all G2 keys match, Ok(false) if any mismatch, Err if query fails
    pub async fn verify_g2_keys_against_chain(&self, operator_addresses: &[Address]) -> eyre::Result<bool> {
        debug!(
            "[G2_VERIFY] Starting G2 key verification for {} operators",
            operator_addresses.len()
        );

        let provider = get_provider(&self.http_rpc_url);
        let operator_registry = OperatorRegistry::new(self.operator_registry_address, provider.clone());

        let bls_apk_registry_address = operator_registry
            .blsApkRegistry()
            .call()
            .await
            .map_err(|e| eyre::eyre!("Failed to get BLS APK registry address: {}", e))?;

        let bls_apk_registry = BLSApkRegistry::new(bls_apk_registry_address, provider);
        let cached_data = self.operator_states.operator_info_data.read().await;

        let mut all_match = true;

        for operator_addr in operator_addresses {
            // Get cached G2 key
            let cached_g2 = cached_data.get(operator_addr).map(|(_, keys)| &keys.g2_pub_key);

            // Get on-chain G2 key
            let onchain_g2_result = bls_apk_registry.getOperatorPubkeyG2(*operator_addr).call().await;

            match (cached_g2, onchain_g2_result) {
                (Some(cached), Ok(onchain_g2_point)) => {
                    let onchain_g2 = BlsG2Point::new(alloy_registry_g2_point_to_g2_affine(onchain_g2_point));

                    // Compare G2 points
                    let cached_g2_affine = cached.g2();
                    let onchain_g2_affine = onchain_g2.g2();

                    let match_result = cached_g2_affine == onchain_g2_affine;

                    if match_result {
                        debug!("[G2_VERIFY] Operator {} G2 key MATCH", operator_addr);
                    } else {
                        error!(
                            "[G2_VERIFY] Operator {} G2 key MISMATCH - cached != on-chain",
                            operator_addr
                        );
                        // Log detailed coordinates for debugging
                        if let (Some(cx), Some(cy)) = (cached_g2_affine.x(), cached_g2_affine.y()) {
                            error!(
                                "[G2_VERIFY] Cached G2: X_c0={} X_c1={} Y_c0={} Y_c1={}",
                                cx.c0, cx.c1, cy.c0, cy.c1
                            );
                        }
                        if let (Some(ox), Some(oy)) = (onchain_g2_affine.x(), onchain_g2_affine.y()) {
                            error!(
                                "[G2_VERIFY] OnChain G2: X_c0={} X_c1={} Y_c0={} Y_c1={}",
                                ox.c0, ox.c1, oy.c0, oy.c1
                            );
                        }
                        all_match = false;
                    }
                }
                (None, Ok(_)) => {
                    warn!(
                        "[G2_VERIFY] Operator {} not in cache but exists on-chain",
                        operator_addr
                    );
                    all_match = false;
                }
                (Some(_), Err(e)) => {
                    error!(
                        "[G2_VERIFY] Failed to fetch on-chain G2 for operator {}: {}",
                        operator_addr, e
                    );
                    all_match = false;
                }
                (None, Err(e)) => {
                    error!(
                        "[G2_VERIFY] Operator {} not in cache and on-chain query failed: {}",
                        operator_addr, e
                    );
                    all_match = false;
                }
            }
        }

        debug!("[G2_VERIFY] Verification complete: all_match={}", all_match);
        Ok(all_match)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::registry::operator::OperatorStakeMap;
    use eigensdk::crypto_bls::OperatorId;
    use std::collections::HashMap;

    fn empty_states() -> OperatorStates {
        OperatorStates {
            operator_info_data: Arc::new(RwLock::new(HashMap::new())),
            operator_addr_to_id: Arc::new(RwLock::new(HashMap::new())),
            socket_dict: Arc::new(RwLock::new(HashMap::new())),
            operator_stake: Arc::new(RwLock::new(OperatorStakeMap::new())),
            total_stake: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    fn id(b: u8) -> OperatorId {
        FixedBytes::<32>::repeat_byte(b)
    }

    #[tokio::test]
    async fn unroutable_stake_counts_only_socketless_registered_operators() {
        // NEWT-1715 / C5: a registered operator with a missing or empty socket is
        // counted in total_stake but skipped when building the broadcast pool, so its
        // stake can never contribute signed_stake. `unroutable_stake_by_quorum` must
        // surface exactly that per-quorum stake and nothing else.
        let states = empty_states();
        let routable = id(0x01);
        let socketless = id(0x02);
        let empty_socket = id(0x03);

        {
            let mut addr_to_id = states.operator_addr_to_id.write().await;
            addr_to_id.insert(Address::repeat_byte(0x01), (StateSource::Historic, routable));
            addr_to_id.insert(Address::repeat_byte(0x02), (StateSource::Historic, socketless));
            addr_to_id.insert(Address::repeat_byte(0x03), (StateSource::Historic, empty_socket));
        }
        {
            let mut sockets = states.socket_dict.write().await;
            sockets.insert(routable, (StateSource::Historic, "http://op1:9000".to_string()));
            // socketless: no socket_dict entry at all.
            sockets.insert(empty_socket, (StateSource::Historic, String::new()));
        }
        {
            let mut stakes = states.operator_stake.write().await;
            stakes.insert(routable, HashMap::from([(0u8, (1000u128, 1u64))]));
            stakes.insert(
                socketless,
                HashMap::from([(0u8, (300u128, 1u64)), (1u8, (50u128, 1u64))]),
            );
            stakes.insert(empty_socket, HashMap::from([(0u8, (200u128, 1u64))]));
        }

        let service = OperatorRegistryService::new_for_test(states);
        let unroutable = service.unroutable_stake_by_quorum().await;

        // Quorum 0: socketless (300) + empty_socket (200) = 500; routable excluded.
        assert_eq!(unroutable.get(&0), Some(&500));
        // Quorum 1: socketless only.
        assert_eq!(unroutable.get(&1), Some(&50));
        assert_eq!(unroutable.len(), 2);
    }

    #[tokio::test]
    async fn operator_total_stake_sums_across_quorums() {
        let states = empty_states();
        let staked = id(0x01);
        states
            .operator_stake
            .write()
            .await
            .insert(staked, HashMap::from([(0u8, (300u128, 1u64)), (1u8, (50u128, 1u64))]));

        let service = OperatorRegistryService::new_for_test(states);
        assert_eq!(service.operator_total_stake(&staked).await, 350);
        // Unknown operator id → zero (not present in the cache).
        assert_eq!(service.operator_total_stake(&id(0x99)).await, 0);
    }

    #[tokio::test]
    async fn known_quorums_reflects_total_stake_keys() {
        let states = empty_states();
        {
            let mut totals = states.total_stake.write().await;
            totals.insert(0u8, (1000u128, 1u64));
            totals.insert(2u8, (500u128, 1u64));
        }
        let service = OperatorRegistryService::new_for_test(states);
        let mut quorums = service.known_quorums().await;
        quorums.sort_unstable();
        assert_eq!(quorums, vec![0, 2]);
    }

    #[tokio::test]
    async fn total_stake_by_quorum_reflects_cache() {
        let states = empty_states();
        {
            let mut totals = states.total_stake.write().await;
            totals.insert(0u8, (1000u128, 1u64));
            totals.insert(1u8, (250u128, 2u64));
        }
        let service = OperatorRegistryService::new_for_test(states);
        let total = service.total_stake_by_quorum().await;
        assert_eq!(total.get(&0), Some(&1000));
        assert_eq!(total.get(&1), Some(&250));
        assert_eq!(total.len(), 2);
    }

    #[tokio::test]
    async fn unroutable_stake_empty_when_all_routable() {
        let states = empty_states();
        let routable = id(0x01);
        {
            states
                .operator_addr_to_id
                .write()
                .await
                .insert(Address::repeat_byte(0x01), (StateSource::Historic, routable));
            states
                .socket_dict
                .write()
                .await
                .insert(routable, (StateSource::Historic, "http://op1:9000".to_string()));
            states
                .operator_stake
                .write()
                .await
                .insert(routable, HashMap::from([(0u8, (1000u128, 1u64))]));
        }

        let service = OperatorRegistryService::new_for_test(states);
        assert!(service.unroutable_stake_by_quorum().await.is_empty());
    }
}