From be4ec2d85bade0611634122c128dbde0d20e7c4d Mon Sep 17 00:00:00 2001 From: Chris O'Neil Date: Wed, 16 Sep 2026 17:59:42 +0100 Subject: [PATCH 1/2] feat: add chunk rpc, upgrade and evm rpc traffic attribution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit V2-834 Part D. Part D of the traffic-accounting ground truth. Three new process-global counter tables, same relaxed-atomic style as the V2-623 replication table, emitted on the existing 300s summary loop. - storage::traffic — the largest attribution gap by volume: chunk RPC bytes were indistinguishable from DHT messaging in `wire_tx_bytes`. Inbound bytes are counted at `ChunkMessage::decode` by request kind (get / put / quote variants / other / decode_error); outbound bytes at send-success in `answer_one_request`, keyed by kind × outcome (get: success/not_found/error; put: success/already_exists/payment_required/ error) via a `traffic_key` carried on `HandledChunkRequest`; failed sends itemised separately. `chunk rpc traffic summary (cumulative)` in two `group`s, target `ant_node::storage::traffic`. - upgrade::traffic — archive / binary / signature / manifest response body sizes across all three reqwest clients; the releases poll now reads `bytes()` then `serde_json::from_slice` so the body length is observable. `upgrade traffic summary (cumulative)`. - payment::traffic — EVM RPC call counts and outcomes at the two on-chain read sites. Bodies are serialised inside alloy behind a fresh per-call evmlib provider, so bytes are not observable from ant-node; EVM RPC is HTTPS/TCP and outside the UDP reconciliation invariant. `evm rpc summary (cumulative)`. Test evidence: `cargo clippy --all-features -- -D clippy::panic -D clippy::unwrap_used -D clippy::expect_used` clean, `cargo clippy --all-targets -- -D warnings` clean apart from the pre-existing `file_store.rs:2005` lint on main, `cargo fmt --check` clean, new `storage::traffic` unit tests pass. Co-Authored-By: Claude Opus 5 (1M context) --- src/node.rs | 14 ++ src/payment/mod.rs | 1 + src/payment/traffic.rs | 57 ++++++++ src/payment/verifier.rs | 32 +++-- src/replication/mod.rs | 5 + src/storage/handler.rs | 59 ++++++--- src/storage/mod.rs | 1 + src/storage/traffic.rs | 278 ++++++++++++++++++++++++++++++++++++++++ src/upgrade/apply.rs | 51 +++++--- src/upgrade/mod.rs | 51 ++++++-- src/upgrade/monitor.rs | 20 ++- src/upgrade/traffic.rs | 76 +++++++++++ 12 files changed, 586 insertions(+), 59 deletions(-) create mode 100644 src/payment/traffic.rs create mode 100644 src/storage/traffic.rs create mode 100644 src/upgrade/traffic.rs diff --git a/src/node.rs b/src/node.rs index 561b7ae2..1c640cb5 100644 --- a/src/node.rs +++ b/src/node.rs @@ -15,6 +15,7 @@ use crate::payment::{ use crate::replication::config::ReplicationConfig; use crate::replication::fresh::FreshWriteEvent; use crate::replication::ReplicationEngine; +use crate::storage::traffic as storage_traffic; use crate::storage::MIB; use crate::storage::{AntProtocol, ChunkRequestContext, ChunkStore, ChunkStoreConfig}; use crate::upgrade::{ @@ -1082,6 +1083,7 @@ impl RunningNode { ) .await; let telemetry = handled.get_telemetry; + let traffic_key = handled.traffic_key; match handled.response { Ok(Some(response)) => { let send_started = Instant::now(); @@ -1091,6 +1093,18 @@ impl RunningNode { if let Some(telemetry) = telemetry { telemetry.finish_send(send_started.elapsed(), send_result.is_ok()); } + // V2-834: attribute response bytes only once the send is + // confirmed; failed sends are itemised separately. + match (&send_result, traffic_key) { + (Ok(()), Some(key)) => storage_traffic::record_tx(key, response.len()), + (Ok(()), None) => { + storage_traffic::record_tx( + storage_traffic::ChunkResponseKey::Other, + response.len(), + ); + } + (Err(_), _) => storage_traffic::record_send_failed(response.len()), + } if let Err(e) = send_result { warn!("Failed to send {data_type} protocol response to {source}: {e}"); } diff --git a/src/payment/mod.rs b/src/payment/mod.rs index d876a06a..c9b2a32d 100644 --- a/src/payment/mod.rs +++ b/src/payment/mod.rs @@ -44,6 +44,7 @@ pub mod pricing; pub mod proof; pub mod quote; pub mod single_node; +pub(crate) mod traffic; mod verifier; pub mod wallet; diff --git a/src/payment/traffic.rs b/src/payment/traffic.rs new file mode 100644 index 00000000..f5e75db7 --- /dev/null +++ b/src/payment/traffic.rs @@ -0,0 +1,57 @@ +//! Cumulative EVM RPC call accounting (V2-834 Part D.3). +//! +//! Both on-chain reads go through `evmlib` → `alloy`, which builds a fresh +//! HTTP provider per call and serialises JSON-RPC inside its own transport, +//! so request/response body sizes are not observable from ant-node without a +//! transport-layer change in `evmlib`. This table therefore records calls and +//! outcomes only. EVM RPC is HTTPS over TCP and sits outside the UDP +//! reconciliation invariant; call counts are what is needed to bound it. + +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Which on-chain read was made. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EvmRpcCall { + /// `IPaymentVault::completedPayments(quote_hash)` (single-node path). + CompletedPayments, + /// `getCompletedMerklePayment(pool_hash)` (merkle batch path). + CompletedMerklePayment, +} + +impl EvmRpcCall { + const N: usize = 2; + + const fn index(self) -> usize { + match self { + Self::CompletedPayments => 0, + Self::CompletedMerklePayment => 1, + } + } +} + +static OK_COUNT: [AtomicU64; EvmRpcCall::N] = [const { AtomicU64::new(0) }; EvmRpcCall::N]; +static ERR_COUNT: [AtomicU64; EvmRpcCall::N] = [const { AtomicU64::new(0) }; EvmRpcCall::N]; + +/// Record the outcome of one on-chain read. +pub fn record(call: EvmRpcCall, ok: bool) { + let table = if ok { &OK_COUNT } else { &ERR_COUNT }; + table[call.index()].fetch_add(1, Ordering::Relaxed); +} + +/// Emit the cumulative EVM RPC call figures as one INFO line, target +/// `ant_node::payment::traffic`. +pub fn log_evm_rpc_summary() { + use EvmRpcCall as C; + + let ok = |c: C| OK_COUNT[c.index()].load(Ordering::Relaxed); + let err = |c: C| ERR_COUNT[c.index()].load(Ordering::Relaxed); + + crate::logging::info!( + target: "ant_node::payment::traffic", + completed_payments_ok_count = ok(C::CompletedPayments), + completed_payments_err_count = err(C::CompletedPayments), + merkle_payment_ok_count = ok(C::CompletedMerklePayment), + merkle_payment_err_count = err(C::CompletedMerklePayment), + "evm rpc summary (cumulative)" + ); +} diff --git a/src/payment/verifier.rs b/src/payment/verifier.rs index 03a8c54d..7252af43 100644 --- a/src/payment/verifier.rs +++ b/src/payment/verifier.rs @@ -1819,11 +1819,14 @@ impl PaymentVerifier { let vault_address = *self.config.evm.network.payment_vault_address(); let contract = payment_vault::interface::IPaymentVault::new(vault_address, provider); - let result = contract - .completedPayments(quote_hash) - .call() - .await - .map_err(|e| Error::Payment(format!("completedPayments lookup failed: {e}")))?; + let result = contract.completedPayments(quote_hash).call().await; + // V2-834: EVM RPC is counted per call (bodies live inside alloy). + super::traffic::record( + super::traffic::EvmRpcCall::CompletedPayments, + result.is_ok(), + ); + let result = + result.map_err(|e| Error::Payment(format!("completedPayments lookup failed: {e}")))?; Ok((Amount::from(result.amount), Some(result.rewardsAddress.0))) } @@ -3254,13 +3257,18 @@ impl PaymentVerifier { // Query on-chain for completed merkle payment let info = payment_vault::get_completed_merkle_payment(&self.config.evm.network, pool_hash) - .await - .map_err(|e| { - let pool_hex = hex::encode(pool_hash); - Error::Payment(format!( - "Failed to query merkle payment info for pool {pool_hex}: {e}" - )) - })?; + .await; + // V2-834: EVM RPC is counted per call (bodies live inside alloy). + super::traffic::record( + super::traffic::EvmRpcCall::CompletedMerklePayment, + info.is_ok(), + ); + let info = info.map_err(|e| { + let pool_hex = hex::encode(pool_hash); + Error::Payment(format!( + "Failed to query merkle payment info for pool {pool_hex}: {e}" + )) + })?; let paid_node_addresses: Vec<_> = info .paidNodeAddresses diff --git a/src/replication/mod.rs b/src/replication/mod.rs index e245bd9d..54498cdc 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -3468,6 +3468,11 @@ impl ReplicationEngine { protocol::log_served_peers_summary(); protocol::log_audit_outcome_summary(); audit_metrics::log_responder_admission_summary(); + // V2-834 Part D: chunk RPC, upgrade and EVM RPC attribution + // ride the same cadence. + crate::storage::traffic::log_chunk_rpc_traffic_summary(); + crate::upgrade::traffic::log_upgrade_traffic_summary(); + crate::payment::traffic::log_evm_rpc_summary(); } } } diff --git a/src/storage/handler.rs b/src/storage/handler.rs index bedffa11..b3c02541 100644 --- a/src/storage/handler.rs +++ b/src/storage/handler.rs @@ -43,6 +43,7 @@ use crate::payment::{PaymentVerifier, QuoteGenerator, VerificationContext}; use crate::replication::admission; use crate::replication::config::K_BUCKET_SIZE; use crate::replication::fresh::FreshWriteEvent; +use crate::storage::traffic::{self, ChunkRequestKind, ChunkResponseKey}; use crate::storage::ChunkStore; use bytes::Bytes; use parking_lot::RwLock; @@ -97,6 +98,9 @@ impl ChunkRequestContext { pub struct HandledChunkRequest { pub(crate) response: Result>, pub(crate) get_telemetry: Option, + /// Kind × outcome of the encoded response, so the router can attribute + /// the bytes once the send is confirmed (V2-834). + pub(crate) traffic_key: Option, } /// Bounded stage timings for one decoded chunk GET. @@ -461,14 +465,21 @@ impl AntProtocol { let message = match ChunkMessage::decode(data) { Ok(message) => message, Err(e) => { + traffic::record_rx(ChunkRequestKind::DecodeError, data.len()); return HandledChunkRequest { response: Err(Error::Protocol(format!("Failed to decode message: {e}"))), get_telemetry: None, + traffic_key: None, }; } }; - let (response, mut get_telemetry) = + // V2-834: attribute inbound bytes by request kind at the decode + // choke point. Only this entry point has the wire length; the browser + // adapter below is handed an already-decoded message. + traffic::record_rx(ChunkRequestKind::of(&message.body), data.len()); + + let (response, mut get_telemetry, traffic_key) = self.handle_message_with_context(message, context).await; let encode_started = Instant::now(); let encoded = response @@ -486,6 +497,7 @@ impl AntProtocol { HandledChunkRequest { response: encoded, get_telemetry, + traffic_key, } } @@ -500,19 +512,27 @@ impl AntProtocol { &self, message: ChunkMessage, context: Option, - ) -> (Option, Option) { + ) -> ( + Option, + Option, + Option, + ) { let request_id = message.request_id; let mut get_telemetry = None; - let response_body = match message.body { + // Each arm yields the response body and its V2-834 traffic key. + let (response_body, traffic_key) = match message.body { ChunkMessageBody::PutRequest(req) => { - ChunkMessageBody::PutResponse(self.handle_put(req).await) + let response = self.handle_put(req).await; + let key = ChunkResponseKey::of_put(&response); + (ChunkMessageBody::PutResponse(response), key) } ChunkMessageBody::GetRequest(req) => { let chunk_address = hex::encode(req.address); let storage_started = Instant::now(); let response = self.handle_get_inner(req).await; let storage_read_ms = duration_ms(storage_started.elapsed()); + let key = ChunkResponseKey::of_get(&response); if let Some(context) = context { get_telemetry = Some(GetRequestTelemetry::from_response( context, @@ -522,26 +542,34 @@ impl AntProtocol { &response, )); } - ChunkMessageBody::GetResponse(response) + (ChunkMessageBody::GetResponse(response), key) } ChunkMessageBody::QuoteRequest(ref req) => { Self::note_unversioned_quote("single_node"); - ChunkMessageBody::QuoteResponse(self.handle_quote(req)) + ( + ChunkMessageBody::QuoteResponse(self.handle_quote(req)), + ChunkResponseKey::Quote, + ) } ChunkMessageBody::MerkleCandidateQuoteRequest(ref req) => { Self::note_unversioned_quote("merkle"); - ChunkMessageBody::MerkleCandidateQuoteResponse( - self.handle_merkle_candidate_quote(req), + ( + ChunkMessageBody::MerkleCandidateQuoteResponse( + self.handle_merkle_candidate_quote(req), + ), + ChunkResponseKey::MerkleQuote, ) } - ChunkMessageBody::QuoteRequestV2(ref req) => { - ChunkMessageBody::QuoteResponse(self.handle_quote_v2(req)) - } - ChunkMessageBody::MerkleCandidateQuoteRequestV2(ref req) => { + ChunkMessageBody::QuoteRequestV2(ref req) => ( + ChunkMessageBody::QuoteResponse(self.handle_quote_v2(req)), + ChunkResponseKey::QuoteV2, + ), + ChunkMessageBody::MerkleCandidateQuoteRequestV2(ref req) => ( ChunkMessageBody::MerkleCandidateQuoteResponse( self.handle_merkle_candidate_quote_v2(req), - ) - } + ), + ChunkResponseKey::MerkleQuoteV2, + ), // Anything else — response messages are handled by client // subscribers (e.g. send_and_await_chunk_response), not by the // protocol handler. Returning None prevents the caller from @@ -554,7 +582,7 @@ impl AntProtocol { // select handshake version-gates peers, so this arm should // only be reached by a misconfigured peer. _ => { - return (None, None); + return (None, None, None); } }; @@ -564,6 +592,7 @@ impl AntProtocol { body: response_body, }), get_telemetry, + Some(traffic_key), ) } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index f27b4733..b60d71c7 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -53,6 +53,7 @@ mod handler; pub(crate) mod lmdb; pub mod migration; pub(crate) mod migration_signal; +pub(crate) mod traffic; pub use crate::ant_protocol::XorName; pub use chunk_store::{ChunkStore, ChunkStoreConfig, VerifyReport, LEGACY_ENV_DIR}; diff --git a/src/storage/traffic.rs b/src/storage/traffic.rs new file mode 100644 index 00000000..dc316f4e --- /dev/null +++ b/src/storage/traffic.rs @@ -0,0 +1,278 @@ +//! Cumulative chunk-RPC traffic accounting (V2-834 Part D.1). +//! +//! Client and community chunk serving flowed through saorsa-core's generic +//! wire counters, indistinguishable from DHT messaging. This module keeps +//! process-global relaxed-atomic tables, in the same style as the replication +//! table in [`crate::replication::protocol`], so "serving user downloads" can +//! be separated from everything else in `wire_tx_bytes`. +//! +//! Requests are counted at decode (by request kind) and responses at +//! send-success (by kind × outcome), so the tx figures are bytes confirmed +//! handed to the transport, not bytes merely encoded. + +use std::sync::atomic::{AtomicU64, Ordering}; + +use crate::ant_protocol::{ChunkGetResponse, ChunkMessageBody, ChunkPutResponse}; + +/// Kind of an inbound chunk message, for the rx table. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ChunkRequestKind { + Get, + Put, + Quote, + MerkleQuote, + QuoteV2, + MerkleQuoteV2, + /// A non-request variant (responses meant for client subscribers) or an + /// unknown future variant. + Other, + /// Bytes that failed `ChunkMessage::decode`. + DecodeError, +} + +impl ChunkRequestKind { + const N: usize = 8; + + const fn index(self) -> usize { + match self { + Self::Get => 0, + Self::Put => 1, + Self::Quote => 2, + Self::MerkleQuote => 3, + Self::QuoteV2 => 4, + Self::MerkleQuoteV2 => 5, + Self::Other => 6, + Self::DecodeError => 7, + } + } +} + +/// Kind × outcome of an outbound chunk response, for the tx table. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ChunkResponseKey { + GetSuccess, + GetNotFound, + GetError, + PutSuccess, + PutAlreadyExists, + PutPaymentRequired, + PutError, + Quote, + MerkleQuote, + QuoteV2, + MerkleQuoteV2, + /// A response variant this table does not itemise (e.g. an `Other` + /// outcome on a `#[non_exhaustive]` enum). + Other, +} + +impl ChunkResponseKey { + const N: usize = 12; + + const fn index(self) -> usize { + match self { + Self::GetSuccess => 0, + Self::GetNotFound => 1, + Self::GetError => 2, + Self::PutSuccess => 3, + Self::PutAlreadyExists => 4, + Self::PutPaymentRequired => 5, + Self::PutError => 6, + Self::Quote => 7, + Self::MerkleQuote => 8, + Self::QuoteV2 => 9, + Self::MerkleQuoteV2 => 10, + Self::Other => 11, + } + } +} + +impl ChunkRequestKind { + /// Classify a decoded inbound message. + pub fn of(body: &ChunkMessageBody) -> Self { + match body { + ChunkMessageBody::GetRequest(_) => Self::Get, + ChunkMessageBody::PutRequest(_) => Self::Put, + ChunkMessageBody::QuoteRequest(_) => Self::Quote, + ChunkMessageBody::MerkleCandidateQuoteRequest(_) => Self::MerkleQuote, + ChunkMessageBody::QuoteRequestV2(_) => Self::QuoteV2, + ChunkMessageBody::MerkleCandidateQuoteRequestV2(_) => Self::MerkleQuoteV2, + _ => Self::Other, + } + } +} + +impl ChunkResponseKey { + /// Classify a GET response by outcome. + pub fn of_get(response: &ChunkGetResponse) -> Self { + match response { + ChunkGetResponse::Success { .. } => Self::GetSuccess, + ChunkGetResponse::NotFound { .. } => Self::GetNotFound, + ChunkGetResponse::Error(_) => Self::GetError, + _ => Self::Other, + } + } + + /// Classify a PUT response by outcome. + pub fn of_put(response: &ChunkPutResponse) -> Self { + match response { + ChunkPutResponse::Success { .. } => Self::PutSuccess, + ChunkPutResponse::AlreadyExists { .. } => Self::PutAlreadyExists, + ChunkPutResponse::PaymentRequired { .. } => Self::PutPaymentRequired, + ChunkPutResponse::Error(_) => Self::PutError, + _ => Self::Other, + } + } +} + +static RX_BYTES: [AtomicU64; ChunkRequestKind::N] = + [const { AtomicU64::new(0) }; ChunkRequestKind::N]; +static RX_COUNT: [AtomicU64; ChunkRequestKind::N] = + [const { AtomicU64::new(0) }; ChunkRequestKind::N]; +static TX_BYTES: [AtomicU64; ChunkResponseKey::N] = + [const { AtomicU64::new(0) }; ChunkResponseKey::N]; +static TX_COUNT: [AtomicU64; ChunkResponseKey::N] = + [const { AtomicU64::new(0) }; ChunkResponseKey::N]; +/// Encoded responses whose transport send failed (not in `TX_*`). +static SEND_FAILED_BYTES: AtomicU64 = AtomicU64::new(0); +static SEND_FAILED_COUNT: AtomicU64 = AtomicU64::new(0); + +/// Record one inbound chunk message at decode time (wire length). +pub fn record_rx(kind: ChunkRequestKind, bytes: usize) { + let i = kind.index(); + RX_BYTES[i].fetch_add(bytes as u64, Ordering::Relaxed); + RX_COUNT[i].fetch_add(1, Ordering::Relaxed); +} + +/// Record one chunk response confirmed handed to the transport. +pub fn record_tx(key: ChunkResponseKey, bytes: usize) { + let i = key.index(); + TX_BYTES[i].fetch_add(bytes as u64, Ordering::Relaxed); + TX_COUNT[i].fetch_add(1, Ordering::Relaxed); +} + +/// Record one encoded chunk response whose send failed. +pub fn record_send_failed(bytes: usize) { + SEND_FAILED_BYTES.fetch_add(bytes as u64, Ordering::Relaxed); + SEND_FAILED_COUNT.fetch_add(1, Ordering::Relaxed); +} + +/// Emit the cumulative chunk-RPC traffic as INFO summary lines, target +/// `ant_node::storage::traffic`. +/// +/// Flat snake-case keys like the replication summary. Two lines sharing the +/// same target and message, distinguished by `group`: rx by request kind +/// (`group = 1`) and tx by kind × outcome (`group = 2`), keeping each under +/// `tracing`'s 32-field cap. +pub fn log_chunk_rpc_traffic_summary() { + use ChunkRequestKind as Q; + use ChunkResponseKey as R; + + let rb = |k: Q| RX_BYTES[k.index()].load(Ordering::Relaxed); + let rc = |k: Q| RX_COUNT[k.index()].load(Ordering::Relaxed); + let tb = |k: R| TX_BYTES[k.index()].load(Ordering::Relaxed); + let tc = |k: R| TX_COUNT[k.index()].load(Ordering::Relaxed); + + crate::logging::info!( + target: "ant_node::storage::traffic", + group = 1, + get_rx_bytes = rb(Q::Get), get_rx_count = rc(Q::Get), + put_rx_bytes = rb(Q::Put), put_rx_count = rc(Q::Put), + quote_rx_bytes = rb(Q::Quote), quote_rx_count = rc(Q::Quote), + merkle_quote_rx_bytes = rb(Q::MerkleQuote), merkle_quote_rx_count = rc(Q::MerkleQuote), + quote_v2_rx_bytes = rb(Q::QuoteV2), quote_v2_rx_count = rc(Q::QuoteV2), + merkle_quote_v2_rx_bytes = rb(Q::MerkleQuoteV2), + merkle_quote_v2_rx_count = rc(Q::MerkleQuoteV2), + other_rx_bytes = rb(Q::Other), other_rx_count = rc(Q::Other), + decode_error_rx_bytes = rb(Q::DecodeError), decode_error_rx_count = rc(Q::DecodeError), + "chunk rpc traffic summary (cumulative)" + ); + + crate::logging::info!( + target: "ant_node::storage::traffic", + group = 2, + get_success_tx_bytes = tb(R::GetSuccess), get_success_tx_count = tc(R::GetSuccess), + get_not_found_tx_bytes = tb(R::GetNotFound), get_not_found_tx_count = tc(R::GetNotFound), + get_error_tx_bytes = tb(R::GetError), get_error_tx_count = tc(R::GetError), + put_success_tx_bytes = tb(R::PutSuccess), put_success_tx_count = tc(R::PutSuccess), + put_already_exists_tx_bytes = tb(R::PutAlreadyExists), + put_already_exists_tx_count = tc(R::PutAlreadyExists), + put_payment_required_tx_bytes = tb(R::PutPaymentRequired), + put_payment_required_tx_count = tc(R::PutPaymentRequired), + put_error_tx_bytes = tb(R::PutError), put_error_tx_count = tc(R::PutError), + quote_tx_bytes = tb(R::Quote), quote_tx_count = tc(R::Quote), + merkle_quote_tx_bytes = tb(R::MerkleQuote), merkle_quote_tx_count = tc(R::MerkleQuote), + quote_v2_tx_bytes = tb(R::QuoteV2), quote_v2_tx_count = tc(R::QuoteV2), + merkle_quote_v2_tx_bytes = tb(R::MerkleQuoteV2), + merkle_quote_v2_tx_count = tc(R::MerkleQuoteV2), + other_tx_bytes = tb(R::Other), other_tx_count = tc(R::Other), + send_failed_tx_bytes = SEND_FAILED_BYTES.load(Ordering::Relaxed), + send_failed_tx_count = SEND_FAILED_COUNT.load(Ordering::Relaxed), + "chunk rpc traffic summary (cumulative)" + ); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn indices_are_distinct_and_in_range() { + let req = [ + ChunkRequestKind::Get, + ChunkRequestKind::Put, + ChunkRequestKind::Quote, + ChunkRequestKind::MerkleQuote, + ChunkRequestKind::QuoteV2, + ChunkRequestKind::MerkleQuoteV2, + ChunkRequestKind::Other, + ChunkRequestKind::DecodeError, + ]; + let mut seen = std::collections::HashSet::new(); + for k in req { + assert!(k.index() < ChunkRequestKind::N); + assert!(seen.insert(k.index())); + } + let resp = [ + ChunkResponseKey::GetSuccess, + ChunkResponseKey::GetNotFound, + ChunkResponseKey::GetError, + ChunkResponseKey::PutSuccess, + ChunkResponseKey::PutAlreadyExists, + ChunkResponseKey::PutPaymentRequired, + ChunkResponseKey::PutError, + ChunkResponseKey::Quote, + ChunkResponseKey::MerkleQuote, + ChunkResponseKey::QuoteV2, + ChunkResponseKey::MerkleQuoteV2, + ChunkResponseKey::Other, + ]; + let mut seen = std::collections::HashSet::new(); + for k in resp { + assert!(k.index() < ChunkResponseKey::N); + assert!(seen.insert(k.index())); + } + } + + #[test] + fn records_accumulate() { + let before_b = RX_BYTES[ChunkRequestKind::Get.index()].load(Ordering::Relaxed); + let before_c = RX_COUNT[ChunkRequestKind::Get.index()].load(Ordering::Relaxed); + record_rx(ChunkRequestKind::Get, 40); + record_rx(ChunkRequestKind::Get, 2); + assert_eq!( + RX_BYTES[ChunkRequestKind::Get.index()].load(Ordering::Relaxed), + before_b + 42 + ); + assert_eq!( + RX_COUNT[ChunkRequestKind::Get.index()].load(Ordering::Relaxed), + before_c + 2 + ); + let before_t = TX_BYTES[ChunkResponseKey::GetSuccess.index()].load(Ordering::Relaxed); + record_tx(ChunkResponseKey::GetSuccess, 4096); + assert_eq!( + TX_BYTES[ChunkResponseKey::GetSuccess.index()].load(Ordering::Relaxed), + before_t + 4096 + ); + } +} diff --git a/src/upgrade/apply.rs b/src/upgrade/apply.rs index 897bd8eb..b375cb14 100644 --- a/src/upgrade/apply.rs +++ b/src/upgrade/apply.rs @@ -10,6 +10,7 @@ use crate::error::{Error, Result}; use crate::logging::{debug, error, info, warn}; use crate::upgrade::binary_cache::BinaryCache; +use crate::upgrade::traffic::{self, UpgradeFetch}; use crate::upgrade::{signature, UpgradeInfo, UpgradeResult}; use flate2::read::GzDecoder; use semver::Version; @@ -283,9 +284,35 @@ impl AutoApplyUpgrader { } /// Download a file to the specified path. - async fn download(&self, url: &str, dest: &Path) -> Result<()> { + async fn download(&self, url: &str, dest: &Path, kind: UpgradeFetch) -> Result<()> { debug!("Downloading: {}", url); + // V2-834: count the body once fully read; anything short of that is + // an error for this fetch kind. + let bytes = match self.fetch_body(url).await { + Ok(bytes) => bytes, + Err(e) => { + traffic::record_error(kind); + return Err(e); + } + }; + traffic::record_rx(kind, bytes.len()); + + if bytes.len() > MAX_ARCHIVE_SIZE_BYTES { + return Err(Error::Upgrade(format!( + "Downloaded file too large: {} bytes (max {})", + bytes.len(), + MAX_ARCHIVE_SIZE_BYTES + ))); + } + + fs::write(dest, &bytes)?; + debug!("Downloaded {} bytes to {}", bytes.len(), dest.display()); + Ok(()) + } + + /// GET `url` and read the whole body. + async fn fetch_body(&self, url: &str) -> Result { let response = self .client .get(url) @@ -300,22 +327,10 @@ impl AutoApplyUpgrader { ))); } - let bytes = response + response .bytes() .await - .map_err(|e| Error::Network(format!("Failed to read response: {e}")))?; - - if bytes.len() > MAX_ARCHIVE_SIZE_BYTES { - return Err(Error::Upgrade(format!( - "Downloaded file too large: {} bytes (max {})", - bytes.len(), - MAX_ARCHIVE_SIZE_BYTES - ))); - } - - fs::write(dest, &bytes)?; - debug!("Downloaded {} bytes to {}", bytes.len(), dest.display()); - Ok(()) + .map_err(|e| Error::Network(format!("Failed to read response: {e}"))) } /// Resolve the upgrade binary, checking the cache first and falling back @@ -405,11 +420,13 @@ impl AutoApplyUpgrader { // Step 1: Download archive info!("Downloading ant-node binary..."); - self.download(&info.download_url, &archive_path).await?; + self.download(&info.download_url, &archive_path, UpgradeFetch::Archive) + .await?; // Step 2: Download signature info!("Downloading signature..."); - self.download(&info.signature_url, &sig_path).await?; + self.download(&info.signature_url, &sig_path, UpgradeFetch::Signature) + .await?; // Step 3: Verify signature on archive BEFORE extraction info!("Verifying ML-DSA signature on archive..."); diff --git a/src/upgrade/mod.rs b/src/upgrade/mod.rs index 5502ac43..a762ad6d 100644 --- a/src/upgrade/mod.rs +++ b/src/upgrade/mod.rs @@ -14,6 +14,7 @@ mod monitor; mod release_cache; mod rollout; mod signature; +pub(crate) mod traffic; pub use apply::{AutoApplyUpgrader, RESTART_EXIT_CODE}; pub use binary_cache::BinaryCache; @@ -212,9 +213,29 @@ impl Upgrader { /// # Errors /// /// Returns an error if the download fails. - async fn download(&self, url: &str, dest: &Path) -> Result<()> { + async fn download(&self, url: &str, dest: &Path, kind: traffic::UpgradeFetch) -> Result<()> { debug!("Downloading: {}", url); + // V2-834: count the body once fully read; anything short of that is + // an error for this fetch kind. + let bytes = match self.fetch_body(url).await { + Ok(bytes) => bytes, + Err(e) => { + traffic::record_error(kind); + return Err(e); + } + }; + traffic::record_rx(kind, bytes.len()); + + Self::enforce_max_binary_size(bytes.len())?; + + fs::write(dest, &bytes)?; + debug!("Downloaded {} bytes to {}", bytes.len(), dest.display()); + Ok(()) + } + + /// GET `url` and read the whole body. + async fn fetch_body(&self, url: &str) -> Result { let response = self .client .get(url) @@ -229,16 +250,10 @@ impl Upgrader { ))); } - let bytes = response + response .bytes() .await - .map_err(|e| Error::Network(format!("Failed to read response: {e}")))?; - - Self::enforce_max_binary_size(bytes.len())?; - - fs::write(dest, &bytes)?; - debug!("Downloaded {} bytes to {}", bytes.len(), dest.display()); - Ok(()) + .map_err(|e| Error::Network(format!("Failed to read response: {e}"))) } /// Ensure the downloaded binary is within a sane size limit. @@ -313,14 +328,28 @@ impl Upgrader { let new_binary = temp_dir.path().join("new_binary"); let sig_path = temp_dir.path().join("signature"); - if let Err(e) = self.download(&info.download_url, &new_binary).await { + if let Err(e) = self + .download( + &info.download_url, + &new_binary, + traffic::UpgradeFetch::Binary, + ) + .await + { warn!("Download failed: {e}"); return Ok(UpgradeResult::RolledBack { reason: format!("Download failed: {e}"), }); } - if let Err(e) = self.download(&info.signature_url, &sig_path).await { + if let Err(e) = self + .download( + &info.signature_url, + &sig_path, + traffic::UpgradeFetch::Signature, + ) + .await + { warn!("Signature download failed: {e}"); return Ok(UpgradeResult::RolledBack { reason: format!("Signature download failed: {e}"), diff --git a/src/upgrade/monitor.rs b/src/upgrade/monitor.rs index e366a398..fcf1f0e3 100644 --- a/src/upgrade/monitor.rs +++ b/src/upgrade/monitor.rs @@ -12,6 +12,7 @@ use crate::error::{Error, Result}; use crate::logging::{debug, info, warn}; use crate::upgrade::release_cache::ReleaseCache; use crate::upgrade::rollout::StagedRollout; +use crate::upgrade::traffic; use crate::upgrade::UpgradeInfo; use semver::Version; use serde::Deserialize; @@ -357,18 +358,29 @@ impl UpgradeMonitor { .header("Accept", "application/vnd.github+json") .send() .await - .map_err(|e| Error::Network(format!("GitHub API request failed: {e}")))?; + .map_err(|e| { + traffic::record_error(traffic::UpgradeFetch::Manifest); + Error::Network(format!("GitHub API request failed: {e}")) + })?; if !response.status().is_success() { + traffic::record_error(traffic::UpgradeFetch::Manifest); return Err(Error::Network(format!( "GitHub API returned status: {}", response.status() ))); } - response - .json() - .await + // V2-834: materialise the body so its size is observable, then parse. + let body = match response.bytes().await { + Ok(body) => body, + Err(e) => { + traffic::record_error(traffic::UpgradeFetch::Manifest); + return Err(Error::Network(format!("Failed to read releases: {e}"))); + } + }; + traffic::record_rx(traffic::UpgradeFetch::Manifest, body.len()); + serde_json::from_slice(&body) .map_err(|e| Error::Network(format!("Failed to parse releases: {e}"))) } diff --git a/src/upgrade/traffic.rs b/src/upgrade/traffic.rs new file mode 100644 index 00000000..b8d8329b --- /dev/null +++ b/src/upgrade/traffic.rs @@ -0,0 +1,76 @@ +//! Cumulative upgrade-download traffic accounting (V2-834 Part D.2). +//! +//! Binary/archive downloads are tens of megabytes and bursty; manifest polls +//! are small but periodic. All are plain HTTPS GETs whose only meaningful +//! byte figure is the response body, counted here once the body has been +//! read in full. Process-global relaxed atomics, same style as the +//! replication and chunk-RPC tables. + +use std::sync::atomic::{AtomicU64, Ordering}; + +/// What an upgrade HTTP fetch was for. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UpgradeFetch { + /// Release archive (auto-apply path). + Archive, + /// Raw binary (legacy path). + Binary, + /// Detached signature file. + Signature, + /// GitHub releases / manifest poll. + Manifest, +} + +impl UpgradeFetch { + const N: usize = 4; + + const fn index(self) -> usize { + match self { + Self::Archive => 0, + Self::Binary => 1, + Self::Signature => 2, + Self::Manifest => 3, + } + } +} + +static RX_BYTES: [AtomicU64; UpgradeFetch::N] = [const { AtomicU64::new(0) }; UpgradeFetch::N]; +static RX_COUNT: [AtomicU64; UpgradeFetch::N] = [const { AtomicU64::new(0) }; UpgradeFetch::N]; +/// Fetches that failed before a body was fully read (network error, non-2xx, +/// body read error). No bytes are attributed to these. +static ERROR_COUNT: [AtomicU64; UpgradeFetch::N] = [const { AtomicU64::new(0) }; UpgradeFetch::N]; + +/// Record a fully-read response body. +pub fn record_rx(kind: UpgradeFetch, bytes: usize) { + let i = kind.index(); + RX_BYTES[i].fetch_add(bytes as u64, Ordering::Relaxed); + RX_COUNT[i].fetch_add(1, Ordering::Relaxed); +} + +/// Record a fetch that did not yield a full body. +pub fn record_error(kind: UpgradeFetch) { + ERROR_COUNT[kind.index()].fetch_add(1, Ordering::Relaxed); +} + +/// Emit the cumulative upgrade fetch figures as one INFO line, target +/// `ant_node::upgrade::traffic`. +pub fn log_upgrade_traffic_summary() { + use UpgradeFetch as U; + + let rb = |k: U| RX_BYTES[k.index()].load(Ordering::Relaxed); + let rc = |k: U| RX_COUNT[k.index()].load(Ordering::Relaxed); + let ec = |k: U| ERROR_COUNT[k.index()].load(Ordering::Relaxed); + + crate::logging::info!( + target: "ant_node::upgrade::traffic", + archive_rx_bytes = rb(U::Archive), archive_rx_count = rc(U::Archive), + archive_error_count = ec(U::Archive), + binary_rx_bytes = rb(U::Binary), binary_rx_count = rc(U::Binary), + binary_error_count = ec(U::Binary), + signature_rx_bytes = rb(U::Signature), signature_rx_count = rc(U::Signature), + signature_error_count = ec(U::Signature), + manifest_rx_bytes = rb(U::Manifest), manifest_rx_count = rc(U::Manifest), + manifest_error_count = ec(U::Manifest), + "upgrade traffic summary (cumulative)" + ); +} From 34fee3e32cc37ed3260cb052940e1542665c8097 Mon Sep 17 00:00:00 2001 From: Chris O'Neil Date: Tue, 22 Sep 2026 22:46:47 +0100 Subject: [PATCH 2/2] chore: advance the saorsa/ant-protocol pins to the V2-834 merges Completes the lockstep for this repo. saorsa-transport#169 merged as 1995901c, saorsa-core#165 as 3dec586c (which also advanced its own transport pin), and ant-protocol#36 as 3f41f74c (which advanced its core pin). Point all three patch entries at those commits so the node actually builds against the counters this PR's code reports alongside. The three must move together: `[patch.crates-io]` cannot rewrite a git dependency, and ant-protocol pins saorsa-core by git rev while saorsa-core pins saorsa-transport by git rev. Bumping one alone resolves two copies of that crate and fails to compile. Verified the lockfile holds exactly one saorsa-transport, one saorsa-core and one ant-protocol after the update. clippy (panic/unwrap/expect denied, --all-features) and fmt clean; cargo check --all-targets clean. Test suites run in CI. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 8 ++++---- Cargo.toml | 6 +++--- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3ee7e5a8..19fc19b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -883,7 +883,7 @@ dependencies = [ [[package]] name = "ant-protocol" version = "2.4.0" -source = "git+https://github.com/WithAutonomi/ant-protocol?rev=1764950d7af880aa13af679ac8c2d0fcbcde0776#1764950d7af880aa13af679ac8c2d0fcbcde0776" +source = "git+https://github.com/WithAutonomi/ant-protocol?rev=3f41f74c4a96c499d3e698c7c1210afe0e1b8742#3f41f74c4a96c499d3e698c7c1210afe0e1b8742" dependencies = [ "blake3", "bytes", @@ -5324,7 +5324,7 @@ dependencies = [ [[package]] name = "saorsa-core" version = "0.27.4" -source = "git+https://github.com/WithAutonomi/saorsa-core?rev=02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c#02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c" +source = "git+https://github.com/WithAutonomi/saorsa-core?rev=3dec586c41eb3a9febe9fd4ed01db07d73e017c5#3dec586c41eb3a9febe9fd4ed01db07d73e017c5" dependencies = [ "anyhow", "async-trait", @@ -5392,7 +5392,7 @@ dependencies = [ [[package]] name = "saorsa-transport" version = "0.36.4" -source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=3ed66a9c0880583ceab1e1550dce756e2716d111#3ed66a9c0880583ceab1e1550dce756e2716d111" +source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=1995901c69d34af43c9d22144a253a92670240d9#1995901c69d34af43c9d22144a253a92670240d9" dependencies = [ "anyhow", "async-trait", @@ -7121,7 +7121,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f3610c9f..e3e88890 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -231,9 +231,9 @@ webrtc-direct = [ [patch.crates-io] saorsa-pqc = { git = "https://github.com/saorsa-labs/saorsa-pqc", rev = "29a2b272c4ee8b72c17102da00033dbfe1ddcd37" } evmlib = { git = "https://github.com/WithAutonomi/evmlib", rev = "ccd65f18bf3750af8b8c3cfe07548bc77e3c368d" } -ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "1764950d7af880aa13af679ac8c2d0fcbcde0776" } -saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c" } -saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "3ed66a9c0880583ceab1e1550dce756e2716d111" } +ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "3f41f74c4a96c499d3e698c7c1210afe0e1b8742" } +saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "3dec586c41eb3a9febe9fd4ed01db07d73e017c5" } +saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "1995901c69d34af43c9d22144a253a92670240d9" } [profile.release] lto = true