freenet 0.2.53

#[cfg(debug_assertions)]
use std::backtrace::Backtrace as StdTrace;
use std::{pin::Pin, time::Duration};

use freenet_stdlib::prelude::{ContractInstanceId, ContractKey};
use futures::Future;
use tokio::sync::mpsc::error::SendError;

use std::net::SocketAddr;

use crate::{
    client_events::HostResult,
    config::GlobalExecutor,
    contract::{ContractError, ExecutorError},
    message::{InnerMessage, MessageStats, NetMessage, NetMessageV1, Transaction, TransactionType},
    node::{ConnectionError, NetworkBridge, OpManager, OpNotAvailable},
    ring::{Location, PeerKeyLocation, RingError},
};

pub(crate) mod connect;
pub(crate) mod get;
pub(crate) mod op_ctx;
pub(crate) mod orphan_streams;
pub(crate) mod put;
pub(crate) mod subscribe;
#[cfg(test)]
pub(crate) mod test_utils;
pub(crate) mod update;
pub(crate) mod visited_peers;

pub(crate) use op_ctx::OpCtx;
pub(crate) use visited_peers::VisitedPeers;

pub(crate) trait Operation
where
    Self: Sized,
{
    type Message: InnerMessage + std::fmt::Display;

    type Result;

    fn load_or_init<'a>(
        op_manager: &'a OpManager,
        msg: &'a Self::Message,
        source_addr: Option<SocketAddr>,
    ) -> impl Future<Output = Result<OpInitialization<Self>, OpError>> + 'a;

    fn id(&self) -> &Transaction;

    #[allow(clippy::type_complexity)]
    fn process_message<'a, CB: NetworkBridge>(
        self,
        conn_manager: &'a mut CB,
        op_manager: &'a OpManager,
        input: &'a Self::Message,
        source_addr: Option<SocketAddr>,
    ) -> Pin<Box<dyn Future<Output = Result<OperationResult, OpError>> + Send + 'a>>;
}

/// Result of processing an operation message.
///
/// This enum encodes the *only* valid result combinations at the type level,
/// replacing the previous struct-of-Options where illegal combinations (e.g.
/// `stream_data: Some` but `return_msg: None`) were representable but
/// meaningless. Each variant documents exactly what the operation layer
/// expects the caller to do next.
///
/// See `handle_op_result` for how each variant is dispatched.
#[must_use]
#[allow(clippy::large_enum_variant)] // Hot path — boxing `OpEnum` would add indirection overhead
pub(crate) enum OperationResult {
    /// Operation is fully complete on this node — nothing to send, no state to keep.
    Completed,

    /// Keep processing: save state but don't send any message to peers.
    /// Used for intermediate states and locally-finalized operations.
    ContinueOp(OpEnum),

    /// Send a message to a peer AND keep the operation state.
    /// Used for forwarding requests and sending responses while the operation
    /// continues (e.g., awaiting sub-operations).
    SendAndContinue {
        msg: NetMessage,
        /// Target peer. `None` for operations that handle their own routing
        /// (connect) or that re-queue locally (update broadcasting).
        next_hop: Option<SocketAddr>,
        state: OpEnum,
        /// Optional stream payload to send alongside the message.
        stream_data: Option<(crate::transport::peer_connection::StreamId, bytes::Bytes)>,
    },

    /// Send a final message and complete the operation on this node.
    /// Used for sending responses upstream when this node is done.
    SendAndComplete {
        msg: NetMessage,
        /// Target peer. `None` for operations that handle their own routing.
        next_hop: Option<SocketAddr>,
        /// Optional stream payload to send alongside the message.
        stream_data: Option<(crate::transport::peer_connection::StreamId, bytes::Bytes)>,
    },
}

pub(crate) struct OpInitialization<Op> {
    /// The source address of the peer that sent this message.
    /// Used for sending error responses (Aborted) and as upstream_addr.
    #[allow(dead_code)]
    pub source_addr: Option<SocketAddr>,
    pub op: Op,
}

pub(crate) async fn handle_op_request<Op, NB>(
    op_manager: &OpManager,
    network_bridge: &mut NB,
    msg: &Op::Message,
    source_addr: Option<SocketAddr>,
) -> Result<Option<OpEnum>, OpError>
where
    Op: Operation,
    NB: NetworkBridge,
{
    let tx = *msg.id();
    let result = {
        let OpInitialization { source_addr: _, op } =
            Op::load_or_init(op_manager, msg, source_addr).await?;
        op.process_message(network_bridge, op_manager, msg, source_addr)
            .await
    };

    handle_op_result(op_manager, network_bridge, result, tx, source_addr).await
}

#[inline(always)]
async fn handle_op_result<CB>(
    op_manager: &OpManager,
    network_bridge: &mut CB,
    result: Result<OperationResult, OpError>,
    tx_id: Transaction,
    source_addr: Option<SocketAddr>,
) -> Result<Option<OpEnum>, OpError>
where
    CB: NetworkBridge,
{
    match result {
        Err(OpError::StatePushed) => {
            // do nothing and continue, the operation will just continue later on
            tracing::debug!("entered in state pushed to continue with op");
            return Ok(None);
        }
        Err(OpError::OpNotPresent(tx)) => {
            // OpNotPresent is benign — it means a duplicate message arrived for an
            // operation that was already completed or claimed (e.g., duplicate metadata
            // from embedded fragment #1 + separate message). Do NOT send Aborted, as
            // the primary processing path is still active and will complete normally.
            tracing::debug!(
                tx = %tx,
                "Ignoring duplicate message for already-handled operation"
            );
            return Ok(None);
        }
        Err(err) => {
            tracing::error!(
                tx = %tx_id,
                error = %err,
                error_debug = ?err,
                source = ?source_addr,
                "handle_op_result: sending Aborted due to operation error"
            );
            if let Some(addr) = source_addr {
                network_bridge
                    .send(addr, NetMessage::V1(NetMessageV1::Aborted(tx_id)))
                    .await?;
            }
            return Err(err);
        }

        // ── No message, no state → fully done ──────────────────────────
        Ok(OperationResult::Completed) => {
            op_manager.completed(tx_id);
        }

        // ── No message, has state → keep processing ────────────────────
        Ok(OperationResult::ContinueOp(state)) => {
            if state.finalized() {
                tracing::debug!(%tx_id, "operation complete");
                op_manager.completed(tx_id);
                return Ok(Some(state));
            } else {
                // Non-finalized: push state for later processing.
                let id = *state.id();
                op_manager.push(id, state).await?;
            }
        }

        // ── Has message + state → send and keep processing ─────────────
        Ok(OperationResult::SendAndContinue {
            msg,
            next_hop,
            state: updated_state,
            stream_data,
        }) => {
            if updated_state.finalized() {
                let id = *msg.id();
                tracing::debug!(%id, "operation finalized with outgoing message");
                op_manager.completed(id);
                if let Some(target) = next_hop {
                    tracing::debug!(%id, ?target, "sending final message to target");
                    send_with_stream(network_bridge, target, msg, stream_data).await?;
                }
                return Ok(Some(updated_state));
            } else {
                let id = *msg.id();
                tracing::debug!(%id, "operation in progress");
                let target = next_hop.ok_or_else(|| {
                    // Only UPDATE's deprecated `Broadcasting` variant produced
                    // SendAndContinue without a next_hop, and it has been
                    // removed. Reaching this branch indicates a bug in an
                    // operation's process_message implementation.
                    OpError::UnexpectedOpState
                })?;
                tracing::debug!(%id, ?target, "sending updated op state");
                // IMPORTANT: Push state BEFORE sending message to avoid race condition.
                // If we send first, a fast response might arrive before the state is saved,
                // causing load_or_init to fail to find the operation.
                op_manager.push(id, updated_state).await?;
                send_with_stream(network_bridge, target, msg, stream_data).await?;
            }
        }

        // ── Has message, no state → send final response and complete ───
        // Complete AFTER send to avoid response-lost: if send fails (peer
        // disconnected), the op stays in under_progress for GC retry (#3590).
        Ok(OperationResult::SendAndComplete {
            msg,
            next_hop,
            stream_data,
        }) => {
            if let Some(target) = next_hop {
                tracing::debug!(%tx_id, ?target, "sending back message to target");
                match send_with_stream(network_bridge, target, msg, stream_data).await {
                    Ok(()) => {
                        op_manager.completed(tx_id);
                    }
                    Err(e) => {
                        // Return directly — bypasses the Aborted-sending error
                        // handler at the top of this function intentionally.
                        // For relay nodes: the op state was already consumed by
                        // process_message, so the tx sits in under_progress until
                        // the 5× TTL cutoff cleans it up. The originator recovers
                        // independently via its own speculative retry (ACK_TIMEOUT
                        // or PROGRESS_TIMEOUT in the GC task).
                        tracing::warn!(
                            %tx_id, %target, error = %e,
                            "Response send failed — originator will retry via speculative path"
                        );
                        return Err(e);
                    }
                }
            } else {
                op_manager.completed(tx_id);
            }
        }
    }
    Ok(None)
}

/// Send a message to a peer, optionally followed by stream data.
///
/// Extracts the repeated send-with-optional-stream pattern from `handle_op_result`.
async fn send_with_stream<CB: NetworkBridge>(
    network_bridge: &mut CB,
    target: SocketAddr,
    msg: NetMessage,
    stream_data: Option<(crate::transport::peer_connection::StreamId, bytes::Bytes)>,
) -> Result<(), OpError> {
    let id = *msg.id();
    // Serialize metadata for embedding in fragment #1 (fix #2757)
    let metadata = if stream_data.is_some() {
        match bincode::serialize(&msg) {
            Ok(bytes) => Some(bytes::Bytes::from(bytes)),
            Err(e) => {
                tracing::warn!(%id, error = %e, "Failed to serialize metadata for embedding");
                None
            }
        }
    } else {
        None
    };
    network_bridge.send(target, msg).await?;
    if let Some((stream_id, data)) = stream_data {
        tracing::debug!(%id, %stream_id, ?target, "sending stream data");
        network_bridge
            .send_stream(target, stream_id, data, metadata)
            .await?;
    }
    Ok(())
}

#[must_use]
#[allow(clippy::large_enum_variant)]
pub(crate) enum OpEnum {
    Connect(Box<connect::ConnectOp>),
    Put(put::PutOp),
    Get(get::GetOp),
    Subscribe(subscribe::SubscribeOp),
    Update(update::UpdateOp),
}

impl OpEnum {
    delegate::delegate! {
        to match self {
            OpEnum::Connect(op) => op,
            OpEnum::Put(op) => op,
            OpEnum::Get(op) => op,
            OpEnum::Subscribe(op) => op,
            OpEnum::Update(op) => op,
        } {
            pub fn id(&self) -> &Transaction;
            pub fn outcome(&self) -> OpOutcome<'_>;
            pub fn finalized(&self) -> bool;
            pub fn to_host_result(&self) -> HostResult;
        }
    }

    /// Returns true if this is a subscription renewal (node-internal operation
    /// with no client waiting for the result).
    pub fn is_subscription_renewal(&self) -> bool {
        matches!(self, OpEnum::Subscribe(op) if op.is_renewal())
    }
}

macro_rules! try_from_op_enum {
    ($op_enum:path, $op_type:ty, $transaction_type:expr) => {
        impl TryFrom<OpEnum> for $op_type {
            type Error = OpError;

            fn try_from(value: OpEnum) -> Result<Self, Self::Error> {
                match value {
                    $op_enum(op) => Ok(op),
                    other => Err(OpError::IncorrectTxType(
                        $transaction_type,
                        other.id().transaction_type(),
                    )),
                }
            }
        }
    };
}

try_from_op_enum!(OpEnum::Put, put::PutOp, TransactionType::Put);
try_from_op_enum!(OpEnum::Get, get::GetOp, TransactionType::Get);
try_from_op_enum!(
    OpEnum::Subscribe,
    subscribe::SubscribeOp,
    TransactionType::Subscribe
);
try_from_op_enum!(OpEnum::Update, update::UpdateOp, TransactionType::Update);

#[derive(Debug)]
pub(crate) enum OpOutcome<'a> {
    /// An op which involves a contract completed successfully.
    ContractOpSuccess {
        target_peer: &'a PeerKeyLocation,
        contract_location: Location,
        /// Time the operation took to initiate.
        first_response_time: Duration,
        /// Size of the payload (contract, state, etc.) in bytes.
        payload_size: usize,
        /// Transfer time of the payload.
        payload_transfer_time: Duration,
    },
    /// An op which involves a contract completed successfully but has no timing data
    /// (put, update). Feeds only the failure estimator.
    ContractOpSuccessUntimed {
        target_peer: &'a PeerKeyLocation,
        contract_location: Location,
    },
    /// An op which involves a contract completed unsuccessfully.
    ContractOpFailure {
        target_peer: &'a PeerKeyLocation,
        contract_location: Location,
    },
    /// In transit contract operation.
    Incomplete,
    /// This operation stats are not relevant for this peer.
    Irrelevant,
}

#[derive(Debug, thiserror::Error)]
pub(crate) enum OpError {
    #[error(transparent)]
    ConnError(#[from] ConnectionError),
    #[error(transparent)]
    RingError(#[from] RingError),
    #[error(transparent)]
    ContractError(#[from] ContractError),
    #[error(transparent)]
    ExecutorError(#[from] ExecutorError),

    #[error("unexpected operation state")]
    UnexpectedOpState,
    #[error(
        "cannot perform a state transition from the current state with the provided input (tx: {tx})"
    )]
    InvalidStateTransition {
        tx: Transaction,
        #[cfg(debug_assertions)]
        state: Option<Box<dyn std::fmt::Debug + Send + Sync>>,
        #[cfg(debug_assertions)]
        trace: StdTrace,
    },
    #[error("failed notifying, channel closed")]
    NotificationError,
    #[error("notification channel error: {0}")]
    NotificationChannelError(String),
    #[error("unspected transaction type, trying to get a {0:?} from a {1:?}")]
    IncorrectTxType(TransactionType, TransactionType),
    #[error("op not present: {0}")]
    OpNotPresent(Transaction),
    #[error("op not available")]
    OpNotAvailable(#[from] OpNotAvailable),

    // Streaming-related errors
    #[error("stream was cancelled")]
    StreamCancelled,
    #[error("failed to claim orphan stream")]
    OrphanStreamClaimFailed,

    // used for control flow
    /// This is used as an early interrumpt of an op update when an op
    /// was sent throught the fast path back to the storage.
    #[error("early push of state into the op stack")]
    StatePushed,
}

impl OpError {
    pub fn invalid_transition(tx: Transaction) -> Self {
        Self::InvalidStateTransition {
            tx,
            #[cfg(debug_assertions)]
            state: None,
            #[cfg(debug_assertions)]
            trace: StdTrace::force_capture(),
        }
    }

    pub fn invalid_transition_with_state(
        tx: Transaction,
        state: Box<dyn std::fmt::Debug + Send + Sync>,
    ) -> Self {
        #[cfg(not(debug_assertions))]
        {
            drop(state);
        }
        Self::InvalidStateTransition {
            tx,
            #[cfg(debug_assertions)]
            state: Some(state),
            #[cfg(debug_assertions)]
            trace: StdTrace::force_capture(),
        }
    }

    /// Returns true if this error indicates a contract's WASM merge function
    /// ran and rejected the update. When true, the contract code is present
    /// locally and auto-fetching would be unnecessary.
    ///
    /// BROADER than `is_invalid_update_rejection` : includes runtime failures
    /// like OOG/timeout/traps. Use this for auto-fetch decisions, NOT for log
    /// severity. See `ExecutorError::is_contract_exec_rejection`.
    pub fn is_contract_exec_rejection(&self) -> bool {
        matches!(self, Self::ExecutorError(e) if e.is_contract_exec_rejection())
    }

    /// Returns true ONLY when the contract WASM merge function rejected the
    /// update with a typed `InvalidUpdate{,WithInfo}` error (the benign
    /// stale-state case from issue #3914). Use this for log-severity
    /// decisions: real WASM faults (OOG, traps, timeouts) return false here
    /// and stay at ERROR/WARN. See `ExecutorError::is_invalid_update_rejection`.
    pub fn is_invalid_update_rejection(&self) -> bool {
        matches!(self, Self::ExecutorError(e) if e.is_invalid_update_rejection())
    }
}

impl<T> From<SendError<T>> for OpError {
    fn from(_: SendError<T>) -> OpError {
        OpError::NotificationError
    }
}

/// Announces to neighbors that we're hosting a contract.
/// This broadcasts to all connected peers so they know to forward UPDATEs to us.
pub(crate) async fn announce_contract_hosted(op_manager: &OpManager, key: &ContractKey) {
    if let Some(announcement) = op_manager.neighbor_hosting.on_contract_hosted(key) {
        tracing::debug!(
            %key,
            "NEIGHBOR_HOSTING: Announcing contract hosted to neighbors"
        );
        if let Err(err) = op_manager
            .notify_node_event(crate::message::NodeEvent::BroadcastHostingUpdate {
                message: announcement,
            })
            .await
        {
            tracing::warn!(
                contract = %key,
                error = %err,
                phase = "error",
                "NEIGHBOR_HOSTING: Failed to broadcast hosting announcement"
            );
        }
    }
}

/// Set up subscription forwarding at a relay node during GET response propagation.
///
/// Relay nodes only set up forwarding (upstream/downstream registration) -- they
/// do NOT call `ring.subscribe()` or `announce_contract_hosted()`, which would
/// cause a subscription storm.
///
/// Registers:
/// 1. Upstream peer (response sender) as interest source for Unsubscribe routing
/// 2. Downstream peer (GET requester) as downstream subscriber for UPDATE propagation
pub(crate) async fn setup_subscription_forwarding_at_relay(
    op_manager: &OpManager,
    key: &ContractKey,
    tx: &crate::message::Transaction,
    upstream_response_addr: std::net::SocketAddr,
    downstream_requester_addr: std::net::SocketAddr,
) {
    // Register upstream peer (response sender) as interest source
    if let Some(upstream_pkl) = op_manager
        .ring
        .connection_manager
        .get_peer_by_addr(upstream_response_addr)
    {
        let peer_key = crate::ring::interest::PeerKey::from(upstream_pkl.pub_key.clone());
        op_manager
            .interest_manager
            .register_peer_interest(key, peer_key, None, true);
    } else {
        tracing::debug!(
            tx = %tx,
            contract = %key,
            upstream = %upstream_response_addr,
            "Piggyback relay: upstream peer not in ring, skipping interest registration"
        );
    }

    // Register downstream peer (GET requester) as downstream subscriber
    subscribe::register_downstream_subscriber(
        op_manager,
        key,
        downstream_requester_addr,
        None,
        None,
        tx,
        " (relay, piggybacked on GET response)",
    )
    .await;

    tracing::debug!(
        tx = %tx,
        contract = %key,
        upstream = %upstream_response_addr,
        downstream = %downstream_requester_addr,
        "Set up subscription forwarding at relay via GET piggyback"
    );
}

/// Complete subscription at the originator node via GET piggyback.
///
/// The subscription tree was built by relay nodes during GET response propagation.
/// This function performs the local registration at the originator:
/// 1. Mark subscribed (lease in active_subscriptions)
/// 2. Register the upstream peer as an interest source
/// 3. Register local interest so ChangeInterests from peers are processed
/// 4. Announce contract hosted so neighbors send UPDATEs
pub(crate) async fn complete_piggyback_subscription(
    op_manager: &OpManager,
    key: &ContractKey,
    tx: &crate::message::Transaction,
    sender_from_addr: &Option<crate::ring::PeerKeyLocation>,
) {
    op_manager.ring.subscribe(*key);
    op_manager.ring.complete_subscription_request(key, true);

    // Register local interest so ChangeInterests from peers get properly processed.
    let became_interested = op_manager.interest_manager.add_local_client(key);
    if became_interested {
        broadcast_change_interests(op_manager, vec![*key], vec![]).await;
    }

    // Announce that we host this contract so neighbors include us in broadcast targets.
    announce_contract_hosted(op_manager, key).await;

    if let Some(upstream_pkl) = sender_from_addr.as_ref() {
        let peer_key = crate::ring::interest::PeerKey::from(upstream_pkl.pub_key.clone());
        op_manager
            .interest_manager
            .register_peer_interest(key, peer_key, None, true);
        tracing::debug!(tx = %tx, contract = %key, "Subscription completed via GET piggyback");
    } else {
        // sender_from_addr can be None for transient connections not yet in the ring.
        // Subscription is still marked locally; the 2-minute renewal cycle will
        // establish a full subscription tree via a standard SUBSCRIBE operation.
        tracing::warn!(
            tx = %tx,
            contract = %key,
            "GET piggyback: upstream peer not in ring, subscription tree incomplete -- renewal will heal"
        );
    }
}

/// Auto-subscribe at the originator: use piggybacked subscription if available,
/// otherwise fall back to a separate SUBSCRIBE operation.
///
/// Called from GET response handling when `AUTO_SUBSCRIBE_ON_GET` is enabled and
/// the originator is not yet subscribed. Consolidates the piggyback-or-fallback
/// logic that appears in both streaming and non-streaming response paths.
pub(crate) async fn auto_subscribe_on_get_response(
    op_manager: &OpManager,
    key: &ContractKey,
    tx: &crate::message::Transaction,
    sender_from_addr: &Option<crate::ring::PeerKeyLocation>,
    subscribe_requested: bool,
    blocking_sub: bool,
    path_label: &str,
) {
    if subscribe_requested {
        complete_piggyback_subscription(op_manager, key, tx, sender_from_addr).await;
    } else {
        let child_tx = start_subscription_request(op_manager, *tx, *key, blocking_sub);
        tracing::debug!(tx = %tx, %child_tx, blocking = %blocking_sub, "started subscription ({path_label}, fallback)");
    }
}

/// Broadcast ChangeInterests message to all connected peers.
///
/// Called when local interest in contracts changes (gained or lost).
pub(crate) async fn broadcast_change_interests(
    op_manager: &OpManager,
    added: Vec<ContractKey>,
    removed: Vec<ContractKey>,
) {
    use crate::ring::interest::contract_hash;

    if added.is_empty() && removed.is_empty() {
        return;
    }

    let added_hashes: Vec<u32> = added.iter().map(contract_hash).collect();
    let removed_hashes: Vec<u32> = removed.iter().map(contract_hash).collect();

    tracing::debug!(
        added_count = added_hashes.len(),
        removed_count = removed_hashes.len(),
        "Broadcasting ChangeInterests to neighbors"
    );

    if let Err(err) = op_manager
        .notify_node_event(crate::message::NodeEvent::BroadcastChangeInterests {
            added: added_hashes,
            removed: removed_hashes,
        })
        .await
    {
        tracing::warn!(
            error = %err,
            "Failed to broadcast ChangeInterests"
        );
    }
}

/// Initiates a subscription after a PUT or GET, routing through the
/// task-per-tx subscribe driver.
///
/// `blocking` is accepted for API stability (callers may inspect it for
/// telemetry), but post-#1454 sub-op SUBSCRIBE migration it has no
/// behavioral effect: the subscribe runs as a fire-and-forget background
/// task in either case. The legacy `SubOperationTracker`-based blocking
/// semantics (parent waits for child via tracker DashMap) are retired —
/// task-per-tx PUT/GET drivers handle blocking by `await`ing
/// `subscribe::run_client_subscribe` inline (see `maybe_subscribe_child`
/// in `put/op_ctx_task.rs` and `get/op_ctx_task.rs`).
pub(super) fn start_subscription_request(
    op_manager: &OpManager,
    parent_tx: Transaction,
    key: ContractKey,
    blocking: bool,
) -> Transaction {
    let child_tx = Transaction::new_child_of::<subscribe::SubscribeMsg>(&parent_tx);

    tracing::debug!(
        %parent_tx,
        %child_tx,
        %key,
        blocking,
        "spawning child subscription operation (task-per-tx driver)"
    );

    // `run_client_subscribe` requires `Arc<OpManager>`. Callers on
    // legacy `process_message` paths only have `&OpManager`, so we
    // wrap a single clone here. PUT/GET task drivers that already
    // hold `&Arc<OpManager>` route through their own
    // `maybe_subscribe_child` helper (see
    // `put/op_ctx_task.rs::maybe_subscribe_child` and the GET
    // counterpart) and don't pay this cost.
    let op_manager_arc = std::sync::Arc::new(op_manager.clone());
    let instance_id = *key.id();
    GlobalExecutor::spawn(async move {
        subscribe::run_client_subscribe(op_manager_arc, instance_id, child_tx).await;
    });

    child_tx
}

async fn has_contract(
    op_manager: &OpManager,
    instance_id: ContractInstanceId,
) -> Result<Option<ContractKey>, OpError> {
    match op_manager
        .notify_contract_handler(crate::contract::ContractHandlerEvent::GetQuery {
            instance_id,
            return_contract_code: false,
        })
        .await?
    {
        crate::contract::ContractHandlerEvent::GetResponse {
            key,
            response: Ok(crate::contract::StoreResponse { state: Some(_), .. }),
        } => Ok(key),
        crate::contract::ContractHandlerEvent::DelegateRequest { .. }
        | crate::contract::ContractHandlerEvent::DelegateResponse(_)
        | crate::contract::ContractHandlerEvent::PutQuery { .. }
        | crate::contract::ContractHandlerEvent::PutResponse { .. }
        | crate::contract::ContractHandlerEvent::GetQuery { .. }
        | crate::contract::ContractHandlerEvent::GetResponse { .. }
        | crate::contract::ContractHandlerEvent::UpdateQuery { .. }
        | crate::contract::ContractHandlerEvent::UpdateResponse { .. }
        | crate::contract::ContractHandlerEvent::UpdateNoChange { .. }
        | crate::contract::ContractHandlerEvent::RegisterSubscriberListener { .. }
        | crate::contract::ContractHandlerEvent::RegisterSubscriberListenerResponse
        | crate::contract::ContractHandlerEvent::QuerySubscriptions { .. }
        | crate::contract::ContractHandlerEvent::QuerySubscriptionsResponse
        | crate::contract::ContractHandlerEvent::GetSummaryQuery { .. }
        | crate::contract::ContractHandlerEvent::GetSummaryResponse { .. }
        | crate::contract::ContractHandlerEvent::GetDeltaQuery { .. }
        | crate::contract::ContractHandlerEvent::GetDeltaResponse { .. }
        | crate::contract::ContractHandlerEvent::NotifySubscriptionError { .. }
        | crate::contract::ContractHandlerEvent::NotifySubscriptionErrorResponse
        | crate::contract::ContractHandlerEvent::ClientDisconnect { .. } => Ok(None),
    }
}

/// Determines if streaming transport should be used for a payload of the given size.
///
/// Returns `true` if the payload size exceeds the streaming threshold (default: 64KB).
///
/// # Arguments
/// * `streaming_threshold` - Size threshold above which streaming is used (exclusive)
/// * `payload_size` - Size of the payload in bytes
///
/// # Note
/// The threshold comparison is exclusive (`>`), meaning payloads exactly at the
/// threshold will NOT use streaming. This is intentional: the threshold represents
/// "the maximum size for non-streaming transfers", so payloads must exceed it.
pub(crate) fn should_use_streaming(streaming_threshold: usize, payload_size: usize) -> bool {
    payload_size > streaming_threshold
}

/// Conservative effective throughput floor for streaming transfers (bytes/sec).
///
/// Used to scale the per-attempt timeout for streaming PUTs. Set to 20 KiB/s
/// so that even a slow link (or congested gateway) has time to drain a large
/// payload before the retry loop fires. Real-world end-to-end throughput
/// observed for the freenet.org website upload (2.4 MB in ~62 s) is ~40 KiB/s,
/// so 20 KiB/s gives ~2x safety margin.
const STREAMING_THROUGHPUT_FLOOR_BPS: usize = 20 * 1024;

/// Minimum drain budget added to streaming timeouts on top of `OPERATION_TTL`.
///
/// Without this floor, payloads just above `streaming_threshold` (where
/// `payload_size / STREAMING_THROUGHPUT_FLOOR_BPS` rounds down to 0) would
/// fall back to the unscaled `OPERATION_TTL` even though `process_message`
/// chose to stream them. That's exactly the #4001 bug — so guarantee at
/// least an extra 30 s of headroom for *every* streaming-eligible payload.
/// 30 s covers stream handshake + first chunk RTT + brief congestion.
const STREAMING_MIN_DRAIN_SECS: u64 = 30;

/// Hard ceiling on the per-attempt timeout for streaming PUTs.
///
/// Even at the throughput floor, a 25 MB payload would only need ~21 minutes,
/// but capping at 10 minutes prevents pathological cases (a wedged remote that
/// never errors) from holding the driver hostage indefinitely. The retry loop
/// can still recover by advancing to a different peer when this fires.
const STREAMING_ATTEMPT_TIMEOUT_CAP: std::time::Duration = std::time::Duration::from_secs(600);

/// Compute the per-attempt timeout for an operation whose payload may use
/// streaming transport.
///
/// For non-streaming payloads (size <= `streaming_threshold`), returns
/// [`crate::config::OPERATION_TTL`] (60 s) — there is no per-fragment progress
/// to wait on, so the standard timeout applies.
///
/// For streaming payloads, returns `OPERATION_TTL` (handshake / k-closest /
/// downstream relays) plus `max(STREAMING_MIN_DRAIN_SECS, payload_size /
/// STREAMING_THROUGHPUT_FLOOR_BPS)` seconds to give the streaming layer time
/// to drain the bytes, capped at [`STREAMING_ATTEMPT_TIMEOUT_CAP`] (10 min).
/// The `STREAMING_MIN_DRAIN_SECS` floor ensures payloads just above the
/// threshold still escape the unscaled `OPERATION_TTL` (integer truncation
/// would otherwise reduce the drain term to zero — re-introducing the
/// #4001 bug for payloads of size `(threshold, threshold + floor_bps)`).
///
/// This is a heuristic: it relocates the cliff at which `drive_retry_loop`
/// fires retries while the original streaming op is still in flight, but does
/// not eliminate it. Issue #4001 has a follow-up design to replace this with
/// a true stream-inactivity timeout that observes per-fragment progress.
pub(crate) fn streaming_aware_attempt_timeout(
    streaming_threshold: usize,
    payload_size: usize,
) -> std::time::Duration {
    if !should_use_streaming(streaming_threshold, payload_size) {
        return crate::config::OPERATION_TTL;
    }
    let drain_secs =
        ((payload_size / STREAMING_THROUGHPUT_FLOOR_BPS) as u64).max(STREAMING_MIN_DRAIN_SECS);
    let total = crate::config::OPERATION_TTL + std::time::Duration::from_secs(drain_secs);
    total.min(STREAMING_ATTEMPT_TIMEOUT_CAP)
}

#[cfg(test)]
mod ordering_invariant_tests {
    //! Tests documenting critical ordering invariants in the operations module.
    //!
    //! These tests don't reproduce actual race conditions (which would require
    //! non-deterministic timing), but document the design decisions and invariants
    //! that prevent them.
    //!
    //! # Push-Before-Send Invariant
    //!
    //! The `handle_op_result` function (lines 178-182) maintains a critical invariant:
    //!
    //! ```text
    //! op_manager.push(id, updated_state).await?;  // FIRST
    //! network_bridge.send(target, msg).await?;    // SECOND
    //! ```
    //!
    //! ## Why This Ordering Matters
    //!
    //! If the order were reversed:
    //! 1. Message is sent to peer
    //! 2. Peer processes and responds FAST
    //! 3. Response arrives at origin
    //! 4. `load_or_init` tries to find operation in storage
    //! 5. **RACE**: `push` hasn't happened yet → operation not found → error
    //!
    //! ## The Invariant
    //!
    //! By pushing state BEFORE sending, we guarantee that when a response
    //! arrives (no matter how fast), the operation state is already in storage.
    //!
    //! ## Why We Can't Easily Test This
    //!
    //! Testing the race would require:
    //! 1. Intercepting between `push` and `send` calls
    //! 2. Simulating an instant response arrival
    //! 3. Verifying `load_or_init` finds the state
    //!
    //! This would require modifying production code to accept test hooks,
    //! which adds complexity for minimal benefit since the invariant is
    //! clear and the code correctly implements it.
    //!
    //! Instead, we document the invariant here and verify the building blocks work.

    use super::test_utils::MockNetworkBridge;
    use crate::message::{NetMessage, NetMessageV1, Transaction};
    use crate::node::NetworkBridge;
    use crate::operations::connect::ConnectMsg;
    use std::net::{IpAddr, Ipv4Addr, SocketAddr};

    /// Verify that MockNetworkBridge correctly records send ordering.
    ///
    /// This is a building block for any future ordering tests.
    #[tokio::test]
    async fn mock_network_bridge_records_send_ordering() {
        let bridge = MockNetworkBridge::new();
        let addr1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5000);
        let addr2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5001);

        let tx1 = Transaction::new::<ConnectMsg>();
        let tx2 = Transaction::new::<ConnectMsg>();

        // Send in specific order
        bridge
            .send(addr1, NetMessage::V1(NetMessageV1::Aborted(tx1)))
            .await
            .unwrap();
        bridge
            .send(addr2, NetMessage::V1(NetMessageV1::Aborted(tx2)))
            .await
            .unwrap();

        // Verify ordering is preserved in recording
        let sent = bridge.sent_messages();
        assert_eq!(sent.len(), 2);
        assert_eq!(sent[0].0, addr1, "First send should be to addr1");
        assert_eq!(sent[1].0, addr2, "Second send should be to addr2");
    }

    /// Document that push-before-send is intentional via code comment verification.
    ///
    /// This test serves as a tripwire: if someone removes the comment explaining
    /// the invariant, this test name will remind them of its importance.
    #[test]
    fn push_before_send_invariant_is_documented() {
        // The invariant is documented at operations.rs lines 178-182:
        //
        // ```rust
        // // IMPORTANT: Push state BEFORE sending message to avoid race condition.
        // // If we send first, a fast response might arrive before the state is saved,
        // // causing load_or_init to fail to find the operation.
        // op_manager.push(id, updated_state).await?;
        // network_bridge.send(target, msg).await?;
        // ```
        //
        // This test documents that the invariant exists and is intentional.
        // If refactoring this code, maintain the push-before-send ordering!
    }
}

#[cfg(test)]
mod streaming_tests {
    use super::{
        STREAMING_ATTEMPT_TIMEOUT_CAP, should_use_streaming, streaming_aware_attempt_timeout,
    };
    use crate::config::OPERATION_TTL;
    use std::time::Duration;

    const DEFAULT_THRESHOLD: usize = 64 * 1024; // 64KB

    #[test]
    fn test_streaming_respects_threshold() {
        assert!(!should_use_streaming(DEFAULT_THRESHOLD, 0));
        assert!(!should_use_streaming(DEFAULT_THRESHOLD, 1000));
        assert!(!should_use_streaming(DEFAULT_THRESHOLD, DEFAULT_THRESHOLD)); // exactly at threshold
        assert!(should_use_streaming(
            DEFAULT_THRESHOLD,
            DEFAULT_THRESHOLD + 1
        )); // just above
        assert!(should_use_streaming(DEFAULT_THRESHOLD, 1024 * 1024)); // 1MB
    }

    #[test]
    fn test_streaming_custom_threshold() {
        let custom_threshold = 128 * 1024; // 128KB
        assert!(!should_use_streaming(custom_threshold, 64 * 1024));
        assert!(!should_use_streaming(custom_threshold, custom_threshold));
        assert!(should_use_streaming(custom_threshold, custom_threshold + 1));
    }

    #[test]
    fn test_streaming_zero_threshold() {
        // With threshold of 0, any non-zero payload uses streaming
        assert!(!should_use_streaming(0, 0));
        assert!(should_use_streaming(0, 1));
        assert!(should_use_streaming(0, 100));
    }

    /// Non-streaming payloads (at or below the threshold) get the standard
    /// `OPERATION_TTL`. Crossing the threshold is what triggers scaling.
    #[test]
    fn non_streaming_payload_uses_operation_ttl() {
        assert_eq!(
            streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, 0),
            OPERATION_TTL
        );
        assert_eq!(
            streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, 1024),
            OPERATION_TTL
        );
        assert_eq!(
            streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, DEFAULT_THRESHOLD),
            OPERATION_TTL
        );
    }

    /// Regression test for #4001: a 2.4 MB payload (the freenet.org website
    /// case) MUST get a per-attempt timeout that exceeds the observed
    /// end-to-end completion time of ~62 s. With the old hard-coded 60 s
    /// `OPERATION_TTL`, the retry loop fired three retries while the
    /// original streaming PUT was still in flight, causing version-conflict
    /// failures on every push to `freenet/web`.
    #[test]
    fn website_payload_attempt_timeout_exceeds_observed_completion() {
        let website_payload_size = 2_460_242; // bytes, from freenet/web 2026-05-01 logs
        let timeout = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, website_payload_size);
        let observed_completion = Duration::from_secs(63); // log: elapsed_ms=62335

        assert!(
            timeout > observed_completion,
            "streaming-aware timeout ({timeout:?}) must exceed observed \
             completion time ({observed_completion:?}) so the retry loop \
             does not fire while the original streaming PUT is still in \
             flight (issue #4001)"
        );
        assert!(
            timeout > OPERATION_TTL,
            "streaming-aware timeout ({timeout:?}) must exceed OPERATION_TTL \
             ({OPERATION_TTL:?}); otherwise the fix is a no-op for the bug \
             reported in #4001"
        );
    }

    /// Streaming timeouts grow with payload size — a 10 MB payload gets a
    /// strictly larger timeout than a 1 MB payload, so the cliff scales.
    #[test]
    fn streaming_timeout_scales_with_payload_size() {
        let small_streaming = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, 1_000_000);
        let medium_streaming = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, 10_000_000);

        assert!(
            small_streaming < medium_streaming,
            "1 MB timeout ({small_streaming:?}) must be smaller than \
             10 MB timeout ({medium_streaming:?})"
        );
        assert!(small_streaming > OPERATION_TTL);
        assert!(medium_streaming > OPERATION_TTL);
    }

    /// Pathological payloads cannot push the per-attempt timeout above the
    /// hard ceiling. Without this, a wedged remote could hold the driver
    /// hostage indefinitely.
    #[test]
    fn streaming_timeout_capped_at_ceiling() {
        // 1 GB — far beyond the per-attempt ceiling at 20 KB/s floor.
        let huge = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, 1024 * 1024 * 1024);
        assert_eq!(
            huge, STREAMING_ATTEMPT_TIMEOUT_CAP,
            "huge payloads must clamp to the cap"
        );
    }

    /// Boundary: the threshold itself is non-streaming, but `threshold + 1`
    /// crosses into streaming territory. The timeout MUST jump *strictly*
    /// above `OPERATION_TTL` at the crossing — otherwise streaming PUTs
    /// just over the threshold inherit the unscaled 60 s timeout that this
    /// fix is trying to escape.
    ///
    /// Without [`super::STREAMING_MIN_DRAIN_SECS`], integer division
    /// would truncate `1 / 20 KiB` to 0 s of drain budget for any payload
    /// of size `(threshold, threshold + STREAMING_THROUGHPUT_FLOOR_BPS)`,
    /// silently re-introducing the #4001 bug. Pin both ends of that gap.
    #[test]
    fn streaming_timeout_jumps_above_threshold_boundary() {
        let at_threshold = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, DEFAULT_THRESHOLD);
        let just_above = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, DEFAULT_THRESHOLD + 1);
        // 19 KiB above threshold — would be `0 s drain` under naive
        // truncation, but the floor saves us.
        let in_truncation_gap =
            streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, DEFAULT_THRESHOLD + 19 * 1024);
        assert_eq!(at_threshold, OPERATION_TTL);
        assert!(
            just_above > OPERATION_TTL,
            "just-above-threshold timeout {just_above:?} must STRICTLY \
             exceed OPERATION_TTL ({OPERATION_TTL:?}); a fix that lets \
             this equal OPERATION_TTL is a no-op for the size range \
             (threshold, threshold + 20 KiB) — exactly the truncation \
             gap STREAMING_MIN_DRAIN_SECS exists to close (#4001 \
             skeptical review)"
        );
        assert!(
            in_truncation_gap > OPERATION_TTL,
            "payload in the truncation gap (threshold + 19 KiB) must \
             exceed OPERATION_TTL — STREAMING_MIN_DRAIN_SECS guarantees it"
        );
    }

    /// The minimum-drain floor applies to every streaming-eligible payload.
    /// Pin the exact value so a future tightening of the floor can't
    /// silently reintroduce the truncation gap.
    #[test]
    fn streaming_timeout_min_drain_floor() {
        let just_above = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, DEFAULT_THRESHOLD + 1);
        assert_eq!(
            just_above,
            OPERATION_TTL + Duration::from_secs(super::STREAMING_MIN_DRAIN_SECS),
            "streaming-eligible payloads must get at least \
             OPERATION_TTL + STREAMING_MIN_DRAIN_SECS"
        );
    }

    /// Pin the exact payload size where the timeout stops scaling and
    /// clamps to the 10-min ceiling. With a 60 s base + 20 KiB/s floor +
    /// 600 s cap, scaling stops at exactly `(600 - 60) * 20 KiB`.
    #[test]
    fn streaming_timeout_cap_boundary() {
        const FLOOR_BPS: usize = 20 * 1024;
        let scaling_max_bytes =
            (STREAMING_ATTEMPT_TIMEOUT_CAP - OPERATION_TTL).as_secs() as usize * FLOOR_BPS;
        let just_below_cap = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, scaling_max_bytes);
        let at_cap = streaming_aware_attempt_timeout(DEFAULT_THRESHOLD, scaling_max_bytes + 1);
        assert_eq!(just_below_cap, STREAMING_ATTEMPT_TIMEOUT_CAP);
        assert_eq!(at_cap, STREAMING_ATTEMPT_TIMEOUT_CAP);
    }
}

#[cfg(test)]
mod sub_op_subscribe_migration_pin_tests {
    //! Pin tests for the #1454 sub-op SUBSCRIBE migration.
    //!
    //! These tests scrape the source of `start_subscription_request` to
    //! prove that:
    //! 1. The legacy `subscribe::request_subscribe` driver and the legacy
    //!    `expect_and_register_sub_operation` registration call are gone
    //!    from the production sub-op SUBSCRIBE entry point.
    //! 2. The function spawns the task-per-tx
    //!    `subscribe::run_client_subscribe` driver instead.
    //!
    //! If a future refactor reintroduces the legacy paths, these tests
    //! will fail loudly and force a re-evaluation against the rationale
    //! recorded in `.claude/rules/operations.md`.

    fn extract_start_subscription_request_body() -> &'static str {
        let src = include_str!("operations.rs");
        // Compose runtime-needle to avoid self-trigger (the test file
        // itself contains the literal "fn start_subscription_request").
        let head = ["fn ", "start_subscription_request("].concat();
        let start = src
            .find(&head)
            .expect("`fn start_subscription_request(` must exist in operations.rs");
        // Walk forward from the function signature to the first `{`,
        // then track brace depth until we find the matching close.
        // This is robust against renaming/moving the next function.
        let body_open = src[start..]
            .find('{')
            .map(|off| start + off)
            .expect("expected `{` after function signature");
        let mut depth: i32 = 0;
        let mut end = body_open;
        for (i, ch) in src[body_open..].char_indices() {
            match ch {
                '{' => depth += 1,
                '}' => {
                    depth -= 1;
                    if depth == 0 {
                        end = body_open + i + 1;
                        break;
                    }
                }
                _ => {}
            }
        }
        assert!(
            end > body_open,
            "failed to find matching `}}` for start_subscription_request"
        );
        &src[start..end]
    }

    #[test]
    fn start_subscription_request_does_not_call_legacy_request_subscribe() {
        let body = extract_start_subscription_request_body();
        assert!(
            !body.contains("subscribe::request_subscribe"),
            "`start_subscription_request` must NOT route through the \
             legacy `subscribe::request_subscribe` driver — sub-op \
             SUBSCRIBE migrated to `subscribe::run_client_subscribe` \
             (see #1454 follow-up)."
        );
    }

    #[test]
    fn start_subscription_request_does_not_register_with_sub_op_tracker() {
        let body = extract_start_subscription_request_body();
        assert!(
            !body.contains("expect_and_register_sub_operation"),
            "`start_subscription_request` must NOT call \
             `expect_and_register_sub_operation` — `SubOperationTracker` \
             was deleted in #1454 Phase 3c. Reintroducing it would be a \
             regression."
        );
        assert!(
            !body.contains("sub_operation_failed"),
            "`start_subscription_request` must NOT propagate failures \
             via `sub_operation_failed` — the task-per-tx subscribe \
             driver publishes its own `HostResult::Err`."
        );
    }

    #[test]
    fn start_subscription_request_spawns_task_per_tx_driver() {
        let body = extract_start_subscription_request_body();
        assert!(
            body.contains("subscribe::run_client_subscribe"),
            "`start_subscription_request` must spawn the task-per-tx \
             subscribe driver `subscribe::run_client_subscribe` — \
             matches the `maybe_subscribe_child` pattern in \
             `put/op_ctx_task.rs` and `get/op_ctx_task.rs`."
        );
    }
}