freenet 0.2.56

pub(crate) mod op_ctx_task;

use freenet_stdlib::client_api::{ErrorKind, HostResponse};
use freenet_stdlib::prelude::*;

pub(crate) use self::messages::{BroadcastStreamingPayload, UpdateMsg, UpdateStreamingPayload};
use super::{OpError, OpOutcome};
use crate::contract::{ContractHandlerEvent, ExecutorError, StoreResponse};
use crate::message::{NodeEvent, Transaction};
use crate::node::IsOperationCompleted;
use crate::ring::{Location, PeerKeyLocation};
use crate::{client_events::HostResult, node::OpManager};
use std::collections::VecDeque;
use std::net::SocketAddr;

use dashmap::DashMap;
use tokio::time::Instant;

/// Cache for deduplicating broadcast payloads.
///
/// When the same delta/state is broadcast to us by multiple peers (which is
/// expected in the gossip topology), we skip the expensive WASM merge call
/// for duplicates. Uses ahash for fast hashing of payload bytes.
pub(crate) struct BroadcastDedupCache {
    /// Per-contract dedup entries, newest at back.
    entries: DashMap<ContractKey, VecDeque<DedupEntry>>,
}

struct DedupEntry {
    delta_hash: u64,
    inserted_at: Instant,
}

/// Maximum entries per contract in the dedup cache.
const DEDUP_MAX_ENTRIES_PER_CONTRACT: usize = 64;

/// TTL for dedup entries — entries older than this are evicted.
const DEDUP_TTL: std::time::Duration = std::time::Duration::from_secs(60);

impl BroadcastDedupCache {
    pub fn new() -> Self {
        Self {
            entries: DashMap::new(),
        }
    }

    /// Check if this payload was already seen for this contract.
    /// If not, insert it and return `false` (not a duplicate).
    /// If yes, return `true` (duplicate — skip processing).
    ///
    /// `is_delta` distinguishes delta payloads from full state payloads so they
    /// don't collide in the hash space (a delta and full state could have the
    /// same bytes but represent different semantic operations).
    ///
    /// Note: Uses non-cryptographic ahash for speed. A hash collision would
    /// cause a legitimate update to be silently skipped, but the gossip
    /// protocol will deliver the data via another peer eventually.
    pub fn check_and_insert(
        &self,
        key: &ContractKey,
        payload_bytes: &[u8],
        is_delta: bool,
        now: Instant,
    ) -> bool {
        use ahash::AHasher;
        use std::hash::Hasher;

        let mut hasher = AHasher::default();
        // Include payload type discriminant to avoid delta/full-state collisions
        hasher.write_u8(if is_delta { 1 } else { 0 });
        hasher.write(payload_bytes);
        let delta_hash = hasher.finish();

        let mut entry = self.entries.entry(*key).or_default();
        let queue = entry.value_mut();

        // Evict expired entries from the front
        while let Some(front) = queue.front() {
            if now.duration_since(front.inserted_at) > DEDUP_TTL {
                queue.pop_front();
            } else {
                break;
            }
        }

        // Check if hash already exists
        if queue.iter().any(|e| e.delta_hash == delta_hash) {
            return true; // Duplicate
        }

        // Evict oldest if at capacity
        while queue.len() >= DEDUP_MAX_ENTRIES_PER_CONTRACT {
            queue.pop_front();
        }

        queue.push_back(DedupEntry {
            delta_hash,
            inserted_at: now,
        });

        false // Not a duplicate
    }
}

/// Result of `get_broadcast_targets_update()` with skip-reason counters
/// for broadcast delivery diagnostics (issue #3046).
pub(crate) struct BroadcastTargetResult {
    pub targets: Vec<PeerKeyLocation>,
    pub proximity_found: usize,
    pub proximity_resolve_failed: usize,
    pub interest_found: usize,
    pub interest_resolve_failed: usize,
    pub skipped_self: usize,
    pub skipped_sender: usize,
}

// `UpdateOp` and its inherent impl + `IsOperationCompleted` impl
// became orphan after #1454 phase 5 final retired the
// `OpEnum::Update` carrier and the `ops.update` DashMap. The struct
// (and its helper methods / `UpdateStats` / `UpdateState` /
// `FinishedData`) are kept under `#[allow(dead_code)]` so the inline
// tests in this module stay green; phase 6 will delete the carrier
// and migrate the surviving outcome / failure-routing tests to the
// task-per-tx driver tests in `op_ctx_task.rs`.
#[allow(dead_code)]
pub(crate) struct UpdateOp {
    pub id: Transaction,
    pub(crate) state: Option<UpdateState>,
    pub(crate) stats: Option<UpdateStats>,
    /// The address we received this operation's message from.
    /// Used for connection-based routing: responses are sent back to this address.
    pub(crate) upstream_addr: Option<std::net::SocketAddr>,
}

#[allow(dead_code)]
impl UpdateOp {
    pub fn id(&self) -> &Transaction {
        &self.id
    }

    pub fn outcome(&self) -> OpOutcome<'_> {
        if self.finalized() {
            if let Some(UpdateStats {
                target: Some(ref target),
                contract_location: Some(loc),
            }) = self.stats
            {
                return OpOutcome::ContractOpSuccessUntimed {
                    target_peer: target,
                    contract_location: loc,
                };
            }
            return OpOutcome::Irrelevant;
        }
        // Not completed — if we have stats with target+location, report as failure
        if let Some(UpdateStats {
            target: Some(ref target),
            contract_location: Some(loc),
        }) = self.stats
        {
            OpOutcome::ContractOpFailure {
                target_peer: target,
                contract_location: loc,
            }
        } else {
            OpOutcome::Incomplete
        }
    }

    /// Returns true if this UPDATE was initiated by a local client (not forwarded from a peer).
    pub(crate) fn is_client_initiated(&self) -> bool {
        self.upstream_addr.is_none()
    }

    /// Extract routing failure info for timeout reporting.
    pub(crate) fn failure_routing_info(&self) -> Option<(PeerKeyLocation, Location)> {
        match &self.stats {
            Some(UpdateStats {
                target: Some(target),
                contract_location: Some(loc),
            }) => Some((target.clone(), *loc)),
            _ => None,
        }
    }

    pub fn finalized(&self) -> bool {
        matches!(self.state, None | Some(UpdateState::Finished(_)))
    }

    /// Get the next hop address for hop-by-hop routing.
    /// For UPDATE, this extracts the socket address from the stats.target field.
    pub fn get_next_hop_addr(&self) -> Option<std::net::SocketAddr> {
        self.stats
            .as_ref()
            .and_then(|s| s.target.as_ref())
            .and_then(|t| t.socket_addr())
    }

    pub(super) fn to_host_result(&self) -> HostResult {
        if let Some(UpdateState::Finished(data)) = &self.state {
            let (key, summary) = (&data.key, &data.summary);
            tracing::debug!(
                "Creating UpdateResponse for transaction {} with key {} and summary length {}",
                self.id,
                key,
                summary.size()
            );
            Ok(HostResponse::ContractResponse(
                freenet_stdlib::client_api::ContractResponse::UpdateResponse {
                    key: *key,
                    summary: summary.clone(),
                },
            ))
        } else {
            tracing::error!(
                tx = %self.id,
                state = ?self.state,
                phase = "error",
                "UPDATE operation failed to finish successfully"
            );
            Err(ErrorKind::OperationError {
                cause: "update didn't finish successfully".into(),
            }
            .into())
        }
    }

    /// Handle aborted connections by failing the operation immediately.
    ///
    /// UPDATE operations don't have alternative routes to try. When the connection
    /// drops, we notify the client of the failure so they can retry.
    pub(crate) async fn handle_abort(self, op_manager: &OpManager) -> Result<(), OpError> {
        tracing::warn!(
            tx = %self.id,
            "Update operation aborted due to connection failure"
        );

        // Create an error result to notify the client
        let error_result: crate::client_events::HostResult =
            Err(freenet_stdlib::client_api::ErrorKind::OperationError {
                cause: "Update operation failed: peer connection dropped".into(),
            }
            .into());

        // Send the error to the client via the result router.
        // Use try_send to avoid blocking the event loop (see channel-safety.md).
        if let Err(err) = op_manager
            .result_router_tx
            .try_send((self.id, error_result))
        {
            tracing::error!(
                tx = %self.id,
                error = %err,
                "Failed to send abort notification to client \
                 (result router channel full or closed)"
            );
        }

        // Mark the operation as completed so it's removed from tracking
        op_manager.completed(self.id);
        Ok(())
    }
}

#[allow(dead_code)]
pub(crate) struct UpdateStats {
    pub(crate) target: Option<PeerKeyLocation>,
    pub(crate) contract_location: Option<Location>,
}

pub(crate) struct UpdateExecution {
    pub(crate) value: WrappedState,
    pub(crate) summary: StateSummary<'static>,
    pub(crate) changed: bool,
}

// `UpdateResult` was the `Operation::Result` associated type for the
// retired `impl Operation for UpdateOp`. Deleted in #1454 phase 5 final.

/// Cooldown before retrying a self-healing contract fetch, in milliseconds.
/// 5 minutes: long enough to avoid hammering peers if the contract genuinely
/// doesn't exist, short enough to recover within a session if there was a
/// transient routing failure.
pub(crate) const CONTRACT_FETCH_COOLDOWN_MS: u64 = 300_000;

impl OpManager {
    /// Trigger a background GET when an UPDATE broadcast fails because the node
    /// doesn't have the contract's parameters (code + params). This self-heals
    /// the node by fetching the contract directly from the UPDATE sender, who
    /// is known to have the contract.
    ///
    /// Rate-limited: at most one fetch attempt per contract per 5 minutes.
    pub(crate) fn try_auto_fetch_contract(&self, key: &ContractKey, sender_addr: SocketAddr) {
        use crate::config::GlobalSimulationTime;

        let instance_id = *key.id();
        let now_ms = GlobalSimulationTime::read_time_ms();

        // Atomic rate-limit: try to insert a new entry. If one exists and hasn't
        // expired, bail out. Uses entry API to avoid TOCTOU between check and insert.
        {
            use dashmap::mapref::entry::Entry;
            match self.pending_contract_fetches.entry(instance_id) {
                Entry::Occupied(mut existing) => {
                    let elapsed_ms = now_ms.saturating_sub(*existing.get());
                    if elapsed_ms < CONTRACT_FETCH_COOLDOWN_MS {
                        return; // Still in cooldown
                    }
                    // Cooldown expired — update timestamp while still holding the lock
                    *existing.get_mut() = now_ms;
                }
                Entry::Vacant(slot) => {
                    slot.insert(now_ms);
                }
            }
        }

        // Look up the sender's PeerKeyLocation so we can target them directly
        let sender_pkl = match self.ring.connection_manager.get_peer_by_addr(sender_addr) {
            Some(pkl) => pkl,
            None => {
                tracing::debug!(
                    contract = %key,
                    sender = %sender_addr,
                    "Cannot auto-fetch: UPDATE sender not found in connection manager"
                );
                self.pending_contract_fetches.remove(&instance_id);
                return;
            }
        };

        tracing::info!(
            contract = %key,
            sender = %sender_addr,
            "Auto-fetching contract from UPDATE sender (missing parameters)"
        );

        // Spawn a targeted task-per-tx GET. The driver targets `sender_pkl`
        // for its first hop and falls back to `k_closest_potentially_hosting`
        // for any retries. Fire-and-forget — the side effect (contract cached
        // locally via `cache_contract_locally`) is what callers depend on.
        // Outcome is logged inside the driver.
        let _tx = super::get::op_ctx_task::start_targeted_sub_op_get(self, instance_id, sender_pkl);
    }

    /// Get the list of network subscribers to broadcast an UPDATE to.
    ///
    /// **Architecture Note (Issue #2075):**
    /// This function returns only **network peer** subscribers, not local client subscriptions.
    /// Local clients receive updates through a separate path via the contract executor's
    /// `update_notifications` channels (see `send_update_notification` in runtime.rs).
    ///
    /// **Parameter `sender`:**
    /// The address of the peer that initiated or forwarded this UPDATE to us.
    /// - Used to filter out the sender from broadcast targets (avoid echo)
    /// - When sender equals our own address (local UPDATE initiation), we include ourselves
    ///   in neighbor hosting targets if we're hosting the contract
    ///
    /// # Hybrid Architecture (2026-01 Refactor)
    ///
    /// Updates are propagated via TWO sources:
    /// 1. Neighbor hosting: peers who have announced they host this contract (fast, local knowledge)
    /// 2. Interest manager: peers who have expressed interest via the Interest/Summary protocol
    ///
    /// This hybrid approach ensures updates reach all interested peers even if HostingAnnounce
    /// messages haven't fully propagated yet.
    pub(crate) fn get_broadcast_targets_update(
        &self,
        key: &ContractKey,
        sender: &SocketAddr,
    ) -> BroadcastTargetResult {
        use std::collections::HashSet;

        let self_addr = self.ring.connection_manager.get_own_addr();
        let is_local_update_initiator = self_addr.as_ref().map(|me| me == sender).unwrap_or(false);

        let mut targets: HashSet<PeerKeyLocation> = HashSet::new();
        let mut proximity_resolve_failed: usize = 0;
        let mut interest_resolve_failed: usize = 0;
        let mut skipped_self: usize = 0;
        let mut skipped_sender: usize = 0;

        // Source 1: Proximity cache (peers who announced they seed this contract)
        // Returns TransportPublicKey (stable identity), resolve to PeerKeyLocation via pub_key lookup
        let proximity_pub_keys = self.neighbor_hosting.neighbors_with_contract(key);
        let proximity_found = proximity_pub_keys.len();

        for pub_key in proximity_pub_keys {
            // Resolve pub_key to PeerKeyLocation (which includes current address)
            if let Some(pkl) = self.ring.connection_manager.get_peer_by_pub_key(&pub_key) {
                // Skip sender to avoid echo (unless we're the originator)
                if let Some(pkl_addr) = pkl.socket_addr() {
                    if &pkl_addr == sender && !is_local_update_initiator {
                        skipped_sender += 1;
                        continue;
                    }
                    // Skip ourselves if not local originator
                    if !is_local_update_initiator && self_addr.as_ref() == Some(&pkl_addr) {
                        skipped_self += 1;
                        continue;
                    }
                }
                targets.insert(pkl);
            } else {
                proximity_resolve_failed += 1;
                tracing::warn!(
                    contract = %format!("{:.8}", key),
                    proximity_neighbor = %pub_key,
                    is_local = is_local_update_initiator,
                    phase = "target_lookup_failed",
                    "Proximity cache neighbor not found in connection manager"
                );
            }
        }

        // Source 2: Interest manager (peers who expressed interest via protocol)
        let interested_peers = self.interest_manager.get_interested_peers(key);
        let interest_found = interested_peers.len();

        for (peer_key, _interest) in interested_peers {
            // Look up peer by public key
            if let Some(pkl) = self
                .ring
                .connection_manager
                .get_peer_by_pub_key(&peer_key.0)
            {
                // Skip sender to avoid echo
                if let Some(pkl_addr) = pkl.socket_addr() {
                    if &pkl_addr == sender && !is_local_update_initiator {
                        skipped_sender += 1;
                        continue;
                    }
                    // Skip ourselves
                    if !is_local_update_initiator && self_addr.as_ref() == Some(&pkl_addr) {
                        skipped_self += 1;
                        continue;
                    }
                }
                targets.insert(pkl);
            } else {
                interest_resolve_failed += 1;
                tracing::warn!(
                    contract = %format!("{:.8}", key),
                    interest_peer = %peer_key.0,
                    is_local = is_local_update_initiator,
                    phase = "target_lookup_failed",
                    "Interest manager peer not found in connection manager"
                );
            }
        }

        // Sort targets for deterministic iteration order
        let mut result: Vec<PeerKeyLocation> = targets.into_iter().collect();
        result.sort();

        // Trace update propagation for debugging
        if !result.is_empty() {
            tracing::info!(
                contract = %format!("{:.8}", key),
                peer_addr = %sender,
                targets = %result
                    .iter()
                    .filter_map(|s| s.socket_addr())
                    .map(|addr| format!("{:.8}", addr))
                    .collect::<Vec<_>>()
                    .join(","),
                count = result.len(),
                proximity_sources = proximity_found,
                interest_sources = interest_found,
                phase = "broadcast",
                "UPDATE_PROPAGATION"
            );
        } else {
            tracing::debug!(
                contract = %format!("{:.8}", key),
                peer_addr = %sender,
                self_addr = ?self_addr.map(|a| format!("{:.8}", a)),
                proximity_sources = proximity_found,
                interest_sources = interest_found,
                phase = "warning",
                "UPDATE_PROPAGATION: NO_TARGETS - update will not propagate further"
            );
        }

        BroadcastTargetResult {
            targets: result,
            proximity_found,
            proximity_resolve_failed,
            interest_found,
            interest_resolve_failed,
            skipped_self,
            skipped_sender,
        }
    }
}

/// Logs the failure outcome of `update_contract`.
///
/// Splits into two cases (issue #3914):
///
/// 1. The contract WASM merge function rejected the incoming state with a
///    typed `InvalidUpdate{,WithInfo}` error (e.g. "New state version 100
///    must be higher than current version 100"). On production gateways
///    this fires on every re-broadcast that misses the 60s dedup cache,
///    generating 80-130 ERROR/hr per gateway with no actionable signal.
///    Logged at INFO with `event="merge_rejected_invalid_update"`.
///
/// 2. Anything else (missing contract code, storage error, OOG, WASM trap,
///    timeout, internal bug) keeps the original ERROR level. The
///    discriminator is `is_invalid_update_rejection`, which matches the
///    contract-side `InvalidUpdate{,WithInfo}` cause string EXCLUSIVELY,
///    so runtime failures like OOG remain visible to operators.
fn log_update_contract_failure(key: &ContractKey, err: &ExecutorError) {
    if err.is_invalid_update_rejection() {
        tracing::info!(
            contract = %key,
            error = %err,
            event = "merge_rejected_invalid_update",
            "Update rejected by contract: incoming state invalid (likely stale rebroadcast), keeping local"
        );
    } else {
        tracing::error!(
            contract = %key,
            error = %err,
            phase = "error",
            "Failed to update contract value"
        );
    }
}

/// Logs the failure outcome of a `BroadcastToStreaming` relay attempt and
/// returns whether the caller should trigger a self-healing GET.
///
/// The two decisions (log severity, auto-fetch) use DIFFERENT predicates
/// because they answer different questions:
///
/// - Log severity uses `is_invalid_update_rejection` (narrow): only the
///   contract WASM's typed "invalid contract update" rejection counts as
///   benign. Out-of-gas, traps, timeouts stay at WARN even though the
///   contract code is present.
///
/// - Auto-fetch uses `is_contract_exec_rejection` (broad): any time the
///   contract code DID execute (whether successfully rejecting a stale
///   state or running out of gas), the code is present locally and a
///   self-heal GET would be wasted. Auto-fetch only fires for failures
///   where the contract is actually missing (e.g., missing parameters
///   after restart, storage error).
///
/// Returning `true` means "fetch missing contract code"; returning `false`
/// means "contract is present, skip auto-fetch".
///
/// Note: this helper takes `&OpError` while `log_update_contract_failure`
/// takes `&ExecutorError` because the two call sites have different error
/// types in scope (the streaming branch operates on the OpError already
/// produced by `update_contract`'s `Err(err.into())`).
pub(crate) fn log_broadcast_to_streaming_failure(
    tx: &Transaction,
    key: &ContractKey,
    err: &OpError,
) -> bool {
    if err.is_invalid_update_rejection() {
        tracing::info!(
            tx = %tx,
            %key,
            error = %err,
            event = "merge_rejected_invalid_update",
            "BroadcastToStreaming merge rejected: incoming state invalid (likely stale rebroadcast), keeping local"
        );
    } else {
        tracing::warn!(
            tx = %tx,
            %key,
            error = %err,
            "BroadcastToStreaming update skipped: contract not ready locally"
        );
    }
    // Preserves the pre-#3914 auto-fetch behavior: skip the self-heal
    // GET whenever the contract code is present (broader check than the
    // log-severity decision above).
    !err.is_contract_exec_rejection()
}

/// Apply an update to a contract.
///
/// This function:
/// 1. Fetches the current state (for change detection)
/// 2. Calls UpdateQuery to merge the update and persist
/// 3. Returns the merged state, summary, and whether the state changed
///
/// The `update_data` parameter can be:
/// - `UpdateData::Delta(delta)` - A delta from the client, merged with current state
/// - `UpdateData::State(state)` - A full state from PUT or executor
pub(crate) async fn update_contract(
    op_manager: &OpManager,
    key: ContractKey,
    update_data: UpdateData<'static>,
    related_contracts: RelatedContracts<'static>,
) -> Result<UpdateExecution, OpError> {
    let previous_state = match op_manager
        .notify_contract_handler(ContractHandlerEvent::GetQuery {
            instance_id: *key.id(),
            return_contract_code: false,
        })
        .await
    {
        Ok(ContractHandlerEvent::GetResponse {
            response: Ok(StoreResponse { state, .. }),
            ..
        }) => state,
        Ok(other) => {
            tracing::trace!(?other, %key, "Unexpected get response while preparing update summary");
            None
        }
        Err(err) => {
            tracing::debug!(%key, %err, "Failed to fetch existing contract state before update");
            None
        }
    };

    match op_manager
        .notify_contract_handler(ContractHandlerEvent::UpdateQuery {
            key,
            data: update_data.clone(),
            related_contracts,
        })
        .await
    {
        Ok(ContractHandlerEvent::UpdateResponse {
            new_value: Ok(new_val),
            state_changed,
        }) => {
            // Invariant: after a successful UPDATE, the resulting state must be non-empty.
            // A successful UpdateResponse with an empty value indicates a contract handler bug.
            debug_assert!(
                new_val.size() > 0,
                "update_contract: state must be non-empty after successful UPDATE for contract {key}"
            );
            let new_bytes = State::from(new_val.clone()).into_bytes();
            let summary = StateSummary::from(new_bytes);

            Ok(UpdateExecution {
                value: new_val,
                summary,
                changed: state_changed,
            })
        }
        Ok(ContractHandlerEvent::UpdateResponse {
            new_value: Err(err),
            ..
        }) => {
            log_update_contract_failure(&key, &err);
            Err(err.into())
        }
        Ok(ContractHandlerEvent::UpdateNoChange { .. }) => {
            // Helper to extract state from UpdateData variants that contain state
            fn extract_state_from_update_data(
                update_data: &UpdateData<'static>,
            ) -> Option<WrappedState> {
                match update_data {
                    UpdateData::State(s) => Some(WrappedState::from(s.clone().into_bytes())),
                    UpdateData::StateAndDelta { state, .. }
                    | UpdateData::RelatedState { state, .. }
                    | UpdateData::RelatedStateAndDelta { state, .. } => {
                        Some(WrappedState::from(state.clone().into_bytes()))
                    }
                    UpdateData::Delta(_) | UpdateData::RelatedDelta { .. } => None,
                    // `UpdateData` is `#[non_exhaustive]` since stdlib
                    // 0.6.0. We can't materialize state from an unknown
                    // variant, so return None and let the caller treat
                    // it as "no state to extract."
                    _ => None,
                }
            }

            let resolved_state = match previous_state {
                Some(prev_state) => prev_state,
                None => {
                    // Try to fetch current state from store
                    let fetched_state = op_manager
                        .notify_contract_handler(ContractHandlerEvent::GetQuery {
                            instance_id: *key.id(),
                            return_contract_code: false,
                        })
                        .await
                        .ok()
                        .and_then(|event| match event {
                            ContractHandlerEvent::GetResponse {
                                response: Ok(StoreResponse { state, .. }),
                                ..
                            } => state,
                            ContractHandlerEvent::DelegateRequest { .. }
                            | ContractHandlerEvent::DelegateResponse(_)
                            | ContractHandlerEvent::PutQuery { .. }
                            | ContractHandlerEvent::PutResponse { .. }
                            | ContractHandlerEvent::GetQuery { .. }
                            | ContractHandlerEvent::GetResponse { .. }
                            | ContractHandlerEvent::UpdateQuery { .. }
                            | ContractHandlerEvent::UpdateResponse { .. }
                            | ContractHandlerEvent::UpdateNoChange { .. }
                            | ContractHandlerEvent::RegisterSubscriberListener { .. }
                            | ContractHandlerEvent::RegisterSubscriberListenerResponse
                            | ContractHandlerEvent::QuerySubscriptions { .. }
                            | ContractHandlerEvent::QuerySubscriptionsResponse
                            | ContractHandlerEvent::GetSummaryQuery { .. }
                            | ContractHandlerEvent::GetSummaryResponse { .. }
                            | ContractHandlerEvent::GetDeltaQuery { .. }
                            | ContractHandlerEvent::GetDeltaResponse { .. }
                            | ContractHandlerEvent::NotifySubscriptionError { .. }
                            | ContractHandlerEvent::NotifySubscriptionErrorResponse
                            | ContractHandlerEvent::ClientDisconnect { .. } => None,
                        });

                    match fetched_state {
                        Some(state) => state,
                        None => {
                            tracing::debug!(
                                %key,
                                "Fallback fetch for UpdateNoChange returned no state; trying to extract from update_data"
                            );
                            match extract_state_from_update_data(&update_data) {
                                Some(state) => state,
                                None => {
                                    tracing::error!(
                                        %key,
                                        "Cannot extract state from delta-only UpdateData in NoChange fallback"
                                    );
                                    return Err(OpError::UnexpectedOpState);
                                }
                            }
                        }
                    }
                }
            };

            let bytes = State::from(resolved_state.clone()).into_bytes();
            let summary = StateSummary::from(bytes);
            Ok(UpdateExecution {
                value: resolved_state,
                summary,
                changed: false,
            })
        }
        Err(err) => Err(err.into()),
        Ok(other) => {
            tracing::error!(event = ?other, contract = %key, phase = "error", "Unexpected event from contract handler during update");
            Err(OpError::UnexpectedOpState)
        }
    }
}

#[allow(dead_code)]
impl IsOperationCompleted for UpdateOp {
    fn is_completed(&self) -> bool {
        matches!(self.state, Some(UpdateState::Finished(_)))
    }
}

/// Send proactive summary notifications to interested peers after a successful
/// state change. This tells neighbors "my state just updated — here's my new
/// summary" so they can update their cached view of us and skip sending
/// redundant broadcasts.
///
/// Accepts the already-computed summary from `UpdateExecution` to avoid an
/// extra WASM `summarize_state` call.
pub(crate) async fn send_proactive_summary_notification(
    op_manager: &OpManager,
    key: &ContractKey,
    sender_addr: SocketAddr,
    summary: StateSummary<'static>,
) {
    use crate::message::{InterestMessage, SummaryEntry};
    use crate::ring::interest::contract_hash;

    // Throttle: at most one notification per contract per 100ms
    if !op_manager
        .interest_manager
        .should_send_summary_notification(key)
    {
        return;
    }

    // Build the Summaries message with our updated summary
    let hash = contract_hash(key);
    let message = InterestMessage::Summaries {
        entries: vec![SummaryEntry::from_summary(hash, Some(&summary))],
    };

    // Get interested peers and send to each (excluding the sender who just sent us the update)
    let interested = op_manager.interest_manager.get_interested_peers(key);
    let self_addr = op_manager.ring.connection_manager.get_own_addr();

    for (peer_key, _interest) in &interested {
        // Resolve peer to socket address
        let peer_addr = match op_manager
            .ring
            .connection_manager
            .get_peer_by_pub_key(&peer_key.0)
        {
            Some(pkl) => match pkl.socket_addr() {
                Some(addr) => addr,
                None => continue,
            },
            None => continue,
        };

        // Skip sender (they just gave us this data) and ourselves
        if peer_addr == sender_addr {
            continue;
        }
        if self_addr.as_ref() == Some(&peer_addr) {
            continue;
        }

        if let Err(e) = op_manager
            .notify_node_event(NodeEvent::SendInterestMessage {
                target: peer_addr,
                message: message.clone(),
            })
            .await
        {
            tracing::debug!(
                contract = %key,
                peer = %peer_addr,
                error = %e,
                "Failed to send proactive summary notification"
            );
        }
    }

    tracing::debug!(
        contract = %key,
        peer_count = interested.len(),
        "Sent proactive summary notifications after state change"
    );
}

/// Send our current summary to the peer whose broadcast we just rejected,
/// **only when the peer's included summary equals our own** — i.e. the
/// peer and we already agree on state, but the peer's cached view of
/// our summary is stale (`None` or out of date) so their send path
/// fell back to full-state at `broadcast_queue.rs:352-356` instead of
/// hitting the summaries-equal fast-path skip.
///
/// The gate on `sender_summary_bytes == our_summary` is load-bearing.
/// Without it, when our state is strictly ahead of the peer's and we
/// reject their stale broadcast, the `InterestMessage::Summaries`
/// handler at `node.rs:1791-1839` detects a mismatch and pushes the
/// peer's state back via `SyncStateToPeer` — which we then reject
/// again, creating a tight reject→summary→resync→reject loop bounded
/// only by the 60 s `BroadcastDedupCache` (and defeated entirely by
/// payload-byte variation). Restricting to the matching-summary case
/// makes this helper a pure convergence nudge: the `Summaries`
/// receiver's stale-detector returns `is_stale = false`, no
/// `SyncStateToPeer` fires, the only outcome is the sender's
/// peer-summary cache of us flipping from `None` → `Some(our_summary)`
/// so its next broadcast takes the fast-path skip.
///
/// Observed in production (nova, vega): ~80–130 rejections/hour per
/// gateway, all with `incoming_state_size == local_state_size` and
/// "version N == N" — i.e. exactly the same-summary case this helper
/// targets. The 5-min `Interests`/`Summaries` heartbeat eventually
/// populates the sender's cache; this shortcut closes the loop in one
/// round-trip instead of waiting for the next heartbeat tick.
///
/// Unlike `send_proactive_summary_notification` (success path, fans
/// out to all interested peers, explicitly excludes the sender), this
/// targets the single sender — on rejection, the sender is the ONE
/// peer whose cache we know is wrong about us.
///
/// Call sites MUST gate this on `err.is_invalid_update_rejection()`
/// (not the broader `is_contract_exec_rejection`): the stricter
/// predicate matches ONLY the benign "new state version not higher"
/// case, not OOG / `MaxComputeTimeExceeded` / WASM traps / validation
/// failures — which are attacker-inducible and shouldn't amplify into
/// extra messages.
pub(crate) async fn send_summary_back_on_rejection(
    op_manager: &OpManager,
    key: &ContractKey,
    target_addr: SocketAddr,
    sender_summary_bytes: Vec<u8>,
) {
    use crate::message::{InterestMessage, SummaryEntry};
    use crate::ring::interest::contract_hash;

    // Throttle BEFORE the WASM `summarize_state` call. Even with call
    // sites gated on `is_invalid_update_rejection`, a flood of
    // crafted-payload broadcasts that all produce benign rejections
    // would otherwise force one `summarize_state` call per rejection.
    // `should_send_summary_notification` caps at ~10 calls/sec/contract.
    //
    // Sharing the throttle map with `send_proactive_summary_notification`
    // is intentional: both paths emit the same `InterestMessage::Summaries`
    // shape, and the existing throttle's purpose (don't re-spam summary
    // updates for the same contract in bursts) applies to both callers.
    if !op_manager
        .interest_manager
        .should_send_summary_notification(key)
    {
        return;
    }

    let Some(our_summary) = op_manager
        .interest_manager
        .get_contract_summary(op_manager, key)
        .await
    else {
        tracing::debug!(
            contract = %key,
            peer = %target_addr,
            "Skipping summary-back on rejection — no local summary available"
        );
        return;
    };

    // Critical: only proceed when summaries match. See function docs for
    // the SyncStateToPeer-loop rationale. A differing summary means the
    // peer is genuinely out of sync, in which case the 5-min heartbeat
    // is the right convergence mechanism — firing Summaries here would
    // escalate into a per-rejection state ping-pong.
    if our_summary.as_ref() != sender_summary_bytes.as_slice() {
        tracing::debug!(
            contract = %key,
            peer = %target_addr,
            "Skipping summary-back on rejection — sender's summary differs \
             from ours (peer is genuinely out of sync; heartbeat will converge)"
        );
        return;
    }

    let hash = contract_hash(key);
    let message = InterestMessage::Summaries {
        entries: vec![SummaryEntry::from_summary(hash, Some(&our_summary))],
    };

    if let Err(e) = op_manager
        .notify_node_event(NodeEvent::SendInterestMessage {
            target: target_addr,
            message,
        })
        .await
    {
        // info! not debug! — debug! is stripped in release builds via
        // `tracing_max_level_info`, so a saturated event-loop channel
        // would fail silently in production.
        tracing::info!(
            contract = %key,
            peer = %target_addr,
            error = %e,
            "Failed to send summary-back after broadcast rejection"
        );
    }
}

mod messages {
    use std::fmt::Display;

    use freenet_stdlib::prelude::{ContractKey, RelatedContracts, WrappedState};
    use serde::{Deserialize, Serialize};

    use crate::{
        message::{InnerMessage, Transaction},
        ring::Location,
        transport::peer_connection::StreamId,
    };

    /// Payload for streaming UPDATE requests.
    ///
    /// Contains the same data as RequestUpdate but serialized for streaming.
    /// The metadata (key, stream_id, total_size) is sent via RequestUpdateStreaming message.
    #[derive(Debug, Serialize, Deserialize)]
    pub(crate) struct UpdateStreamingPayload {
        #[serde(deserialize_with = "RelatedContracts::deser_related_contracts")]
        pub related_contracts: RelatedContracts<'static>,
        pub value: WrappedState,
    }

    /// Payload for streaming broadcast updates.
    ///
    /// Contains full state for broadcasting to subscribers via streaming.
    /// Used when the full state is large (>streaming_threshold).
    #[derive(Debug, Serialize, Deserialize)]
    pub(crate) struct BroadcastStreamingPayload {
        /// Full contract state bytes
        pub state_bytes: Vec<u8>,
        /// Sender's current state summary bytes
        pub sender_summary_bytes: Vec<u8>,
    }

    #[derive(Debug, Serialize, Deserialize, Clone)]
    /// Update operation messages.
    ///
    /// Uses hop-by-hop routing for request forwarding. Broadcasting to subscribers
    /// uses explicit addresses since there are multiple targets.
    pub(crate) enum UpdateMsg {
        /// Request to update a contract state. Forwarded hop-by-hop toward contract location.
        RequestUpdate {
            id: Transaction,
            key: ContractKey,
            #[serde(deserialize_with = "RelatedContracts::deser_related_contracts")]
            related_contracts: RelatedContracts<'static>,
            value: WrappedState,
        },
        /// Broadcasting a change to a specific subscriber.
        ///
        /// Supports delta-based synchronization: when we know the peer's state summary,
        /// we send a delta instead of full state to reduce bandwidth.
        BroadcastTo {
            id: Transaction,
            key: ContractKey,
            /// The payload: either a delta (if we know peer's summary) or full state.
            payload: crate::message::DeltaOrFullState,
            /// Sender's current state summary bytes, so receiver can update their tracking.
            /// Use `StateSummary::from(sender_summary_bytes.clone())` to convert.
            sender_summary_bytes: Vec<u8>,
        },

        // ---- Streaming variants ----
        /// Streaming variant of RequestUpdate for large state updates.
        ///
        /// Used when the state size exceeds the streaming threshold (default 64KB).
        /// The actual state data is sent via a separate stream identified by stream_id.
        RequestUpdateStreaming {
            id: Transaction,
            /// Identifies the stream carrying the update payload
            stream_id: StreamId,
            /// Contract key being updated
            key: ContractKey,
            /// Total size of the streamed payload in bytes
            total_size: u64,
        },

        /// Streaming variant of BroadcastTo for large full state broadcasts.
        ///
        /// Used when broadcasting full state (not delta) and the state size exceeds
        /// the streaming threshold. Deltas are typically small and use regular BroadcastTo.
        BroadcastToStreaming {
            id: Transaction,
            /// Identifies the stream carrying the broadcast payload
            stream_id: StreamId,
            /// Contract key being broadcast
            key: ContractKey,
            /// Total size of the streamed payload in bytes
            total_size: u64,
        },
    }

    impl InnerMessage for UpdateMsg {
        fn id(&self) -> &Transaction {
            match self {
                UpdateMsg::RequestUpdate { id, .. }
                | UpdateMsg::BroadcastTo { id, .. }
                | UpdateMsg::RequestUpdateStreaming { id, .. }
                | UpdateMsg::BroadcastToStreaming { id, .. } => id,
            }
        }

        fn requested_location(&self) -> Option<crate::ring::Location> {
            match self {
                UpdateMsg::RequestUpdate { key, .. }
                | UpdateMsg::BroadcastTo { key, .. }
                | UpdateMsg::RequestUpdateStreaming { key, .. }
                | UpdateMsg::BroadcastToStreaming { key, .. } => Some(Location::from(key.id())),
            }
        }
    }

    impl Display for UpdateMsg {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            match self {
                UpdateMsg::RequestUpdate { id, .. } => write!(f, "RequestUpdate(id: {id})"),
                UpdateMsg::BroadcastTo { id, .. } => write!(f, "BroadcastTo(id: {id})"),
                UpdateMsg::RequestUpdateStreaming { id, stream_id, .. } => {
                    write!(f, "RequestUpdateStreaming(id: {id}, stream: {stream_id})")
                }
                UpdateMsg::BroadcastToStreaming { id, stream_id, .. } => {
                    write!(f, "BroadcastToStreaming(id: {id}, stream: {stream_id})")
                }
            }
        }
    }
}

// State machine for UPDATE operations.
//
// # Important: Updates are Fire-and-Forget
//
// Updates spread through the subscription tree like a virus - there is no acknowledgment
// or completion signal that propagates back. When a node receives an UPDATE:
// 1. It applies the update locally
// 2. It broadcasts to its downstream subscribers (if any)
// 3. It does NOT wait for or expect any response from downstream
//
// ── Type-state data structs ──────────────────────────────────────────────
//
// Each state in the UPDATE state machine is a named struct with typed
// transition methods.  This gives compile-time guarantees:
//
//   ReceivedRequest ── (next state depends on whether contract is found locally)
//   PrepareRequest ── into_finished()  ──► Finished
//
// Invalid transitions (e.g. Finished → ReceivedRequest) are unrepresentable
// because the target type simply has no such method.

/// Data for the Finished state: update was applied locally.
///
/// The `Finished` state only indicates that the LOCAL update was applied successfully.
/// It does NOT mean all subscribers have received the update - that's unknowable.
#[allow(dead_code)]
#[derive(Debug)]
pub struct FinishedData {
    pub key: ContractKey,
    pub summary: StateSummary<'static>,
}

// `PrepareRequestData` and `UpdateState::PrepareRequest` were retired in
// #1454 phase 5 final together with `start_op` / `request_update`.
//
// `UpdateState::ReceivedRequest` and `UpdateState::Finished` survive
// because the legacy `UpdateOp` carrier (kept until commit 4 deletes
// `OpEnum::Update`) references them via `is_completed()` / `outcome()`
// / `to_host_result()` and the inline test helpers.

// ── State enum (wraps the typed structs) ─────────────────────────────────

#[allow(dead_code)]
#[derive(Debug)]
pub enum UpdateState {
    /// Initial state when receiving an update request from another peer.
    ReceivedRequest,
    /// The update was applied locally.
    Finished(FinishedData),
}

#[cfg(test)]
#[allow(clippy::wildcard_enum_match_arm)]
mod tests {
    use super::*;
    use crate::operations::OpOutcome;
    use crate::operations::test_utils::make_contract_key;

    fn make_update_op(state: Option<UpdateState>, stats: Option<UpdateStats>) -> UpdateOp {
        UpdateOp {
            id: Transaction::new::<UpdateMsg>(),
            state,
            stats,
            upstream_addr: None,
        }
    }

    #[test]
    fn is_client_initiated_true_when_no_upstream() {
        let op = make_update_op(None, None);
        assert!(op.is_client_initiated());
    }

    #[test]
    fn is_client_initiated_false_when_forwarded() {
        let op = UpdateOp {
            id: Transaction::new::<UpdateMsg>(),
            state: None,
            stats: None,
            upstream_addr: Some("127.0.0.1:12345".parse().unwrap()),
        };
        assert!(!op.is_client_initiated());
    }

    #[test]
    fn update_op_outcome_success_untimed_when_finalized_with_full_stats() {
        let target = PeerKeyLocation::random();
        let loc = Location::random();
        let op = make_update_op(
            Some(UpdateState::Finished(FinishedData {
                key: make_contract_key(1),
                summary: StateSummary::from(vec![1u8]),
            })),
            Some(UpdateStats {
                target: Some(target.clone()),
                contract_location: Some(loc),
            }),
        );
        match op.outcome() {
            OpOutcome::ContractOpSuccessUntimed {
                target_peer,
                contract_location,
            } => {
                assert_eq!(*target_peer, target);
                assert_eq!(contract_location, loc);
            }
            OpOutcome::ContractOpSuccess { .. }
            | OpOutcome::ContractOpFailure { .. }
            | OpOutcome::Incomplete
            | OpOutcome::Irrelevant => {
                panic!("Expected ContractOpSuccessUntimed for finalized update with full stats")
            }
        }
    }

    #[test]
    fn update_op_outcome_irrelevant_when_finalized_without_stats() {
        let op = make_update_op(
            Some(UpdateState::Finished(FinishedData {
                key: make_contract_key(1),
                summary: StateSummary::from(vec![1u8]),
            })),
            None,
        );
        assert!(matches!(op.outcome(), OpOutcome::Irrelevant));
    }

    #[test]
    fn update_op_outcome_irrelevant_when_finalized_with_partial_stats() {
        // target is None — should fall through to Irrelevant
        let op = make_update_op(
            Some(UpdateState::Finished(FinishedData {
                key: make_contract_key(1),
                summary: StateSummary::from(vec![1u8]),
            })),
            Some(UpdateStats {
                target: None,
                contract_location: Some(Location::random()),
            }),
        );
        assert!(matches!(op.outcome(), OpOutcome::Irrelevant));
    }

    #[test]
    fn update_op_outcome_failure_when_not_finalized_with_full_stats() {
        let target = PeerKeyLocation::random();
        let loc = Location::random();
        let op = make_update_op(
            Some(UpdateState::ReceivedRequest),
            Some(UpdateStats {
                target: Some(target.clone()),
                contract_location: Some(loc),
            }),
        );
        match op.outcome() {
            OpOutcome::ContractOpFailure {
                target_peer,
                contract_location,
            } => {
                assert_eq!(*target_peer, target);
                assert_eq!(contract_location, loc);
            }
            OpOutcome::ContractOpSuccess { .. }
            | OpOutcome::ContractOpSuccessUntimed { .. }
            | OpOutcome::Incomplete
            | OpOutcome::Irrelevant => {
                panic!("Expected ContractOpFailure for non-finalized update with full stats")
            }
        }
    }

    #[test]
    fn update_op_outcome_incomplete_when_not_finalized_without_stats() {
        let op = make_update_op(Some(UpdateState::ReceivedRequest), None);
        assert!(matches!(op.outcome(), OpOutcome::Incomplete));
    }

    #[test]
    fn update_op_failure_routing_info_with_full_stats() {
        let target = PeerKeyLocation::random();
        let loc = Location::random();
        let op = make_update_op(
            None,
            Some(UpdateStats {
                target: Some(target.clone()),
                contract_location: Some(loc),
            }),
        );
        let info = op.failure_routing_info().expect("should have routing info");
        assert_eq!(info.0, target);
        assert_eq!(info.1, loc);
    }

    #[test]
    fn update_op_failure_routing_info_without_stats() {
        let op = make_update_op(None, None);
        assert!(op.failure_routing_info().is_none());
    }

    #[test]
    fn update_op_failure_routing_info_with_partial_stats() {
        let op = make_update_op(
            None,
            Some(UpdateStats {
                target: None,
                contract_location: Some(Location::random()),
            }),
        );
        assert!(op.failure_routing_info().is_none());
    }

    // `start_op_creates_update_with_partial_stats` tested the deleted
    // `start_op` constructor (#1454 phase 5 final). Replacement
    // coverage lives in `update_op_stats_lifecycle_*` below, which
    // exercises the same partial-stats → finalized transition without
    // depending on the legacy constructor.

    /// Simulate the update operation lifecycle: partial stats → full stats → finished.
    /// This mirrors what happens when an update operation finds a forwarding target.
    #[test]
    fn update_op_stats_lifecycle_from_partial_to_complete() {
        let target = PeerKeyLocation::random();
        let loc = Location::random();

        // Step 1: Created with partial stats (no target yet)
        let mut op = make_update_op(
            Some(UpdateState::ReceivedRequest),
            Some(UpdateStats {
                target: None,
                contract_location: Some(loc),
            }),
        );
        assert!(matches!(op.outcome(), OpOutcome::Incomplete));
        assert!(op.failure_routing_info().is_none());

        // Step 2: Target found during forwarding
        op.stats = Some(UpdateStats {
            target: Some(target.clone()),
            contract_location: Some(loc),
        });
        // Still not finalized → ContractOpFailure (has full stats but not done)
        match op.outcome() {
            OpOutcome::ContractOpFailure {
                target_peer,
                contract_location,
            } => {
                assert_eq!(*target_peer, target);
                assert_eq!(contract_location, loc);
            }
            OpOutcome::ContractOpSuccess { .. }
            | OpOutcome::ContractOpSuccessUntimed { .. }
            | OpOutcome::Incomplete
            | OpOutcome::Irrelevant => {
                panic!("Expected ContractOpFailure for in-progress update with stats")
            }
        }
        assert!(op.failure_routing_info().is_some());

        // Step 3: Operation finishes
        op.state = Some(UpdateState::Finished(FinishedData {
            key: make_contract_key(1),
            summary: StateSummary::from(vec![1u8]),
        }));
        match op.outcome() {
            OpOutcome::ContractOpSuccessUntimed {
                target_peer,
                contract_location,
            } => {
                assert_eq!(*target_peer, target);
                assert_eq!(contract_location, loc);
            }
            OpOutcome::ContractOpSuccess { .. }
            | OpOutcome::ContractOpFailure { .. }
            | OpOutcome::Incomplete
            | OpOutcome::Irrelevant => {
                panic!("Expected ContractOpSuccessUntimed for finished update with stats")
            }
        }
    }

    /// Verify that contract_location=None in UpdateStats causes Incomplete/Irrelevant
    /// outcomes even when a target is set.
    #[test]
    fn update_op_outcome_with_target_but_no_contract_location() {
        let target = PeerKeyLocation::random();

        // Not finalized with target but no location → Incomplete
        let op = make_update_op(
            Some(UpdateState::ReceivedRequest),
            Some(UpdateStats {
                target: Some(target.clone()),
                contract_location: None,
            }),
        );
        assert!(matches!(op.outcome(), OpOutcome::Incomplete));
        assert!(op.failure_routing_info().is_none());

        // Finalized with target but no location → Irrelevant
        let op = make_update_op(
            Some(UpdateState::Finished(FinishedData {
                key: make_contract_key(1),
                summary: StateSummary::from(vec![1u8]),
            })),
            Some(UpdateStats {
                target: Some(target),
                contract_location: None,
            }),
        );
        assert!(matches!(op.outcome(), OpOutcome::Irrelevant));
    }

    // ── Intermediate node stats tracking tests (#3527) ─────────────────────

    use crate::operations::test_utils::make_peer;

    /// Non-finalized UPDATE with stats reports ContractOpFailure on timeout.
    #[test]
    fn test_update_failure_outcome_with_stats() {
        let target = make_peer(9001);
        let contract_location = Location::from(&make_contract_key(42));

        let op = make_update_op(
            Some(UpdateState::ReceivedRequest),
            Some(UpdateStats {
                target: Some(target.clone()),
                contract_location: Some(contract_location),
            }),
        );

        assert!(!op.finalized());
        match op.outcome() {
            OpOutcome::ContractOpFailure {
                target_peer,
                contract_location: loc,
            } => {
                assert_eq!(target_peer, &target);
                assert_eq!(loc, contract_location);
            }
            other => panic!("Expected ContractOpFailure, got {other:?}"),
        }
    }

    /// Non-finalized UPDATE without stats reports Incomplete.
    #[test]
    fn test_update_failure_outcome_without_stats() {
        let op = make_update_op(Some(UpdateState::ReceivedRequest), None);

        assert!(!op.finalized());
        assert!(
            matches!(op.outcome(), OpOutcome::Incomplete),
            "UPDATE without stats should return Incomplete"
        );
    }

    /// failure_routing_info() returns correct peer and location from stats.
    #[test]
    fn test_update_failure_routing_info() {
        let target = make_peer(9002);
        let contract_location = Location::from(&make_contract_key(42));

        let op = make_update_op(
            Some(UpdateState::ReceivedRequest),
            Some(UpdateStats {
                target: Some(target.clone()),
                contract_location: Some(contract_location),
            }),
        );

        let (peer, loc) = op.failure_routing_info().expect("should have routing info");
        assert_eq!(peer, target);
        assert_eq!(loc, contract_location);
    }

    /// Regression tests for issue #3914: misleading ERROR/WARN log noise from
    /// benign WASM rejections of stale broadcast UPDATEs. The contract correctly
    /// rejects an incoming state at a version we already hold (a re-broadcast
    /// the dedup cache missed). On production gateways this generated 80-130
    /// ERROR-level lines per hour per gateway. The tests below pin both that
    /// the benign case is now INFO AND that real WASM failures (out-of-gas,
    /// max-compute-time, traps) stay at ERROR/WARN, so the predicate cannot
    /// silently broaden and hide real failures.
    mod log_severity {
        use super::*;
        use crate::contract::ExecutorError;
        use crate::test_utils::TestLogger;
        use freenet_stdlib::client_api::{ContractError as StdContractError, RequestError};

        // Constructors use `From<RequestError> for ExecutorError` (a public
        // impl) rather than the module-private `ExecutorError::request`, so
        // these helpers don't require widening any visibility for tests.
        fn invalid_update_rejection() -> ExecutorError {
            // Mirrors the production cause string exactly: stdlib's
            // `update_exec_error` prefixes "execution error: " and the
            // contract WASM's `InvalidUpdateWithInfo` Display produces
            // "invalid contract update, reason: ...".
            let req: RequestError = StdContractError::update_exec_error(
                make_contract_key(1),
                "invalid contract update, reason: New state version 100 must be higher than current version 100",
            )
            .into();
            req.into()
        }

        fn out_of_gas_failure() -> ExecutorError {
            // Real WASM fault: contract ran out of gas. Same `update_exec_error`
            // wrapper as the benign case (so the loose `is_contract_exec_rejection`
            // predicate matches both), but the cause string starts with
            // "execution error: The operation ran out of gas..." which the
            // tighter `is_invalid_update_rejection` predicate must REJECT.
            let req: RequestError = StdContractError::update_exec_error(
                make_contract_key(1),
                "The operation ran out of gas. This might be caused by an infinite loop or an inefficient computation.",
            )
            .into();
            req.into()
        }

        fn missing_parameters_failure() -> ExecutorError {
            // Real failure case: contract not ready locally, auto-fetch needed.
            let req: RequestError = StdContractError::Update {
                key: make_contract_key(2),
                cause: "missing contract parameters".into(),
            }
            .into();
            req.into()
        }

        #[test]
        fn update_contract_failure_logs_info_for_invalid_update_rejection() {
            let logger = TestLogger::new().capture_logs().with_level("info").init();

            log_update_contract_failure(&make_contract_key(1), &invalid_update_rejection());

            assert!(
                logger.contains("merge_rejected_invalid_update"),
                "expected event=merge_rejected_invalid_update in logs, got: {:?}",
                logger.logs()
            );
            assert!(
                logger.contains("INFO"),
                "expected INFO-level log for invalid-update rejection, got: {:?}",
                logger.logs()
            );
            assert!(
                !logger.logs().iter().any(|l| l.contains("ERROR")),
                "invalid-update rejection must not produce ERROR-level logs, got: {:?}",
                logger.logs()
            );
        }

        #[test]
        fn update_contract_failure_logs_error_for_real_failure() {
            let logger = TestLogger::new().capture_logs().with_level("info").init();

            log_update_contract_failure(&make_contract_key(2), &missing_parameters_failure());

            assert!(
                logger.contains("ERROR"),
                "real failures must remain ERROR-level, got: {:?}",
                logger.logs()
            );
            assert!(
                logger.contains("Failed to update contract value"),
                "expected ERROR message text, got: {:?}",
                logger.logs()
            );
        }

        /// CRITICAL: out-of-gas comes through the same `update_exec_error`
        /// wrapper as the benign rejection, so the loose
        /// `is_contract_exec_rejection` predicate matches it (used for the
        /// auto-fetch gate, where this is correct: contract code IS present).
        /// But for log severity, OOG is a real bug operators must see and
        /// MUST stay at ERROR. This test pins that.
        #[test]
        fn update_contract_failure_logs_error_for_out_of_gas() {
            let logger = TestLogger::new().capture_logs().with_level("info").init();

            log_update_contract_failure(&make_contract_key(1), &out_of_gas_failure());

            assert!(
                logger.contains("ERROR"),
                "out-of-gas must remain ERROR-level (real WASM fault), got: {:?}",
                logger.logs()
            );
            assert!(
                !logger
                    .logs()
                    .iter()
                    .any(|l| l.contains("merge_rejected_invalid_update")),
                "out-of-gas must NOT be classified as a benign rejection, got: {:?}",
                logger.logs()
            );
        }

        #[test]
        fn broadcast_to_streaming_failure_logs_info_and_skips_auto_fetch_for_invalid_update() {
            let logger = TestLogger::new().capture_logs().with_level("info").init();
            let tx = Transaction::new::<UpdateMsg>();
            let err: OpError = invalid_update_rejection().into();

            let needs_auto_fetch =
                log_broadcast_to_streaming_failure(&tx, &make_contract_key(1), &err);

            assert!(
                !needs_auto_fetch,
                "invalid-update rejection must NOT trigger self-heal auto-fetch (contract code is present)"
            );
            assert!(
                logger.contains("merge_rejected_invalid_update"),
                "expected event=merge_rejected_invalid_update in logs, got: {:?}",
                logger.logs()
            );
            assert!(
                !logger.logs().iter().any(|l| l.contains("WARN")),
                "invalid-update rejection must not produce WARN-level logs (the old misleading 'contract not ready locally' line), got: {:?}",
                logger.logs()
            );
            assert!(
                !logger
                    .logs()
                    .iter()
                    .any(|l| l.contains("contract not ready locally")),
                "the misleading 'contract not ready locally' message must not appear for invalid-update rejections, got: {:?}",
                logger.logs()
            );
        }

        #[test]
        fn broadcast_to_streaming_failure_logs_warn_and_triggers_auto_fetch_for_real_failure() {
            let logger = TestLogger::new().capture_logs().with_level("info").init();
            let tx = Transaction::new::<UpdateMsg>();
            let err: OpError = missing_parameters_failure().into();

            let needs_auto_fetch =
                log_broadcast_to_streaming_failure(&tx, &make_contract_key(2), &err);

            assert!(
                needs_auto_fetch,
                "real failures must trigger self-heal auto-fetch"
            );
            assert!(
                logger.contains("WARN"),
                "real failures remain WARN-level for the streaming branch, got: {:?}",
                logger.logs()
            );
            assert!(
                logger.contains("contract not ready locally"),
                "expected the WARN message text for real failure, got: {:?}",
                logger.logs()
            );
        }

        /// Mirror of `update_contract_failure_logs_error_for_out_of_gas` for
        /// the streaming branch: OOG must stay at WARN, AND auto-fetch must
        /// be SKIPPED because the contract code is present (broader predicate
        /// `is_contract_exec_rejection` correctly catches this case). The two
        /// decisions are deliberately decoupled by the helper.
        #[test]
        fn broadcast_to_streaming_failure_logs_warn_and_skips_auto_fetch_for_out_of_gas() {
            let logger = TestLogger::new().capture_logs().with_level("info").init();
            let tx = Transaction::new::<UpdateMsg>();
            let err: OpError = out_of_gas_failure().into();

            let needs_auto_fetch =
                log_broadcast_to_streaming_failure(&tx, &make_contract_key(1), &err);

            assert!(
                logger.contains("WARN"),
                "out-of-gas must remain WARN-level for the streaming branch, got: {:?}",
                logger.logs()
            );
            assert!(
                !logger
                    .logs()
                    .iter()
                    .any(|l| l.contains("merge_rejected_invalid_update")),
                "out-of-gas must NOT be classified as a benign rejection, got: {:?}",
                logger.logs()
            );
            assert!(
                !needs_auto_fetch,
                "out-of-gas must NOT trigger self-heal auto-fetch (contract code is present locally; the broader is_contract_exec_rejection predicate catches this case independently of log severity)"
            );
        }
    }

    /// Pin: `UpdateMsg::BroadcastTo` handler in `process_message` must spawn
    /// `send_summary_back_on_rejection` when the WASM merge rejects as an
    /// invalid-update rejection (stale version). Skipping this causes the
    /// sender to repeatedly full-state-broadcast identical content; see
    /// PR description for production evidence.
    #[test]
    fn legacy_broadcast_to_sends_summary_back_on_rejection() {
        let src = include_str!("update.rs");
        let bcast_arm_pos = src
            .find("UpdateMsg::BroadcastTo {")
            .expect("UpdateMsg::BroadcastTo match arm not found");
        // Scope to the Err(err) branch of this arm.
        let err_branch_start = src[bcast_arm_pos..]
            .find("Err(err) =>")
            .map(|p| bcast_arm_pos + p)
            .expect("BroadcastTo Err(err) branch not found");
        let branch = &src[err_branch_start..err_branch_start + 3500];

        assert!(
            branch.contains("err.is_invalid_update_rejection()"),
            "BroadcastTo rejection branch MUST gate summary-back on \
             is_invalid_update_rejection (not is_contract_exec_rejection): \
             the broader predicate matches OOG/traps which are \
             attacker-inducible and must not amplify into summary-back"
        );
        assert!(
            branch.contains("send_summary_back_on_rejection"),
            "send_summary_back_on_rejection call missing — stale-version \
             rejections in legacy BroadcastTo will keep amplifying"
        );
    }

    /// Pin: `UpdateMsg::BroadcastToStreaming` handler must spawn
    /// `send_summary_back_on_rejection` on the invalid-update rejection
    /// branch (the non-`try_auto_fetch_contract` branch).
    #[test]
    fn legacy_broadcast_to_streaming_sends_summary_back_on_rejection() {
        let src = include_str!("update.rs");
        let stream_arm_pos = src
            .find("UpdateMsg::BroadcastToStreaming")
            .expect("BroadcastToStreaming arm not found");
        let err_branch_start = src[stream_arm_pos..]
            .find("Err(err) =>")
            .map(|p| stream_arm_pos + p)
            .expect("BroadcastToStreaming Err(err) branch not found");
        let branch = &src[err_branch_start..err_branch_start + 3000];

        assert!(
            branch.contains("err.is_invalid_update_rejection()"),
            "BroadcastToStreaming Err branch must gate summary-back on \
             is_invalid_update_rejection"
        );
        assert!(
            branch.contains("send_summary_back_on_rejection"),
            "BroadcastToStreaming Err branch must spawn \
             send_summary_back_on_rejection on invalid-update rejection"
        );
        assert!(
            branch.contains("try_auto_fetch_contract"),
            "BroadcastToStreaming Err branch must still call \
             try_auto_fetch_contract on the non-rejection (missing-contract) path"
        );
    }

    /// Pin: `send_summary_back_on_rejection` helper MUST gate on
    /// `sender_summary_bytes` matching our current summary before sending
    /// `InterestMessage::Summaries`. Without this gate, a mismatch triggers
    /// `SyncStateToPeer` at `node.rs:1791-1839` which re-sends the sender's
    /// stale state back to us, creating a reject→summary→resync→reject
    /// loop. ALSO pins the throttle-before-WASM ordering to prevent
    /// attacker-induced `summarize_state` amplification under rejection
    /// floods. See the helper's docs and PR description.
    #[test]
    fn summary_back_helper_gates_on_summary_equality() {
        let src = include_str!("update.rs");
        let fn_start = src
            .find("pub(crate) async fn send_summary_back_on_rejection(")
            .expect("send_summary_back_on_rejection fn not found");
        let fn_end_offset = src[fn_start..]
            .find("\n}\n")
            .expect("send_summary_back_on_rejection fn close not found");
        let fn_body = &src[fn_start..fn_start + fn_end_offset];

        assert!(
            fn_body.contains("sender_summary_bytes"),
            "send_summary_back_on_rejection must take sender_summary_bytes \
             as a parameter"
        );

        // Pin throttle-before-WASM ordering. Without this, an attacker
        // inducing rejections can force one `summarize_state` WASM call per
        // rejection — the equality gate below guards only the outgoing send.
        let throttle_pos = fn_body
            .find("should_send_summary_notification")
            .expect("helper MUST call should_send_summary_notification (throttle)");
        let wasm_call_pos = fn_body
            .find("get_contract_summary")
            .expect("helper must call get_contract_summary to compute our_summary");
        assert!(
            throttle_pos < wasm_call_pos,
            "should_send_summary_notification MUST run before get_contract_summary \
             — otherwise attacker-induced rejections force unbounded WASM amplification"
        );

        // Pin the EXACT inequality direction: on difference → early return.
        // If the direction is reversed (== → early return, meaning "send only
        // when they differ"), the helper reintroduces the SyncStateToPeer
        // loop that this PR fixes.
        let inequality_check_pos = fn_body
            .find("our_summary.as_ref() != sender_summary_bytes.as_slice()")
            .expect(
                "helper MUST use `our_summary.as_ref() != sender_summary_bytes.as_slice()` \
                 as the gate condition (reversed direction reintroduces the SyncStateToPeer \
                 loop — see node.rs:1791-1839)",
            );
        let after_check = &fn_body[inequality_check_pos..];
        let return_pos = after_check.find("return;").expect(
            "gate must early-return on inequality (reversed direction would bypass \
             the SyncStateToPeer safeguard)",
        );
        let notify_pos = after_check
            .find("notify_node_event")
            .expect("helper must still call notify_node_event on the equality path");
        assert!(
            return_pos < notify_pos,
            "early return on inequality MUST precede the notify_node_event \
             send — otherwise mismatched summaries would still be transmitted"
        );
    }
}