diff --git a/Cargo.lock b/Cargo.lock index 248d9970..56385dab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -895,8 +895,7 @@ dependencies = [ [[package]] name = "ant-protocol" version = "5.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d95baa1de35f7183806372d59baa34b93c2de8f28c76c664f3225008ed17317" +source = "git+https://github.com/WithAutonomi/ant-protocol?rev=0c6ae27b6261227a32e6fe0f976163c42bde960e#0c6ae27b6261227a32e6fe0f976163c42bde960e" dependencies = [ "blake3", "bytes", @@ -5278,8 +5277,7 @@ dependencies = [ [[package]] name = "saorsa-core" version = "0.29.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49de4f92b6808afd54a206d7dcb6cf754f796ee876394d6838b79b39dd1a72dd" +source = "git+https://github.com/WithAutonomi/saorsa-core?rev=c3c4c81e63db7e3048f488f7cf0fec4cff3131d4#c3c4c81e63db7e3048f488f7cf0fec4cff3131d4" dependencies = [ "anyhow", "async-trait", @@ -7055,7 +7053,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index aa249a27..355364f3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -214,6 +214,13 @@ webrtc-direct = [ "saorsa-transport/webrtc-direct", ] +[patch.crates-io] +# ADR-0020 (V2-1358): saorsa-core's closer-peers encoding and +# ant-protocol's get-or-closer request, each rebased onto its main. Drop +# both once releases include them. +saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "c3c4c81e63db7e3048f488f7cf0fec4cff3131d4" } +ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "0c6ae27b6261227a32e6fe0f976163c42bde960e" } + [profile.release] lto = true codegen-units = 1 diff --git a/docs/adr/ADR-0020-kademlia-chunk-get-or-closer.md b/docs/adr/ADR-0020-kademlia-chunk-get-or-closer.md new file mode 100644 index 00000000..37e72bc1 --- /dev/null +++ b/docs/adr/ADR-0020-kademlia-chunk-get-or-closer.md @@ -0,0 +1,331 @@ +# ADR-0020: Kademlia-style chunk GET: return the chunk or closer peers + +- **Status:** Proposed +- **Date:** 2026-10-07 +- **Decision owners:** +- **Reviewers:** +- **Supersedes:** none +- **Superseded by:** none +- **Related:** Linear V2-1358; ant-client investigation + `docs/investigations/2026-10-06-kademlia-chunk-get/`; + [ADR-0015](ADR-0015-direct-browser-clients-over-webrtc-direct.md) (browser + capabilities); + [ADR-0023](ADR-0023-spread-hot-reads-and-refuse-full-browser-sessions.md) + (spread first reads); saorsa-core, ant-protocol and ant-client branches + `mickvandijke/v2-1358-kad-chunk-get` + +## Context + +A native chunk read runs two separate steps today: + +1. saorsa-core's iterative closest-peers lookup runs until it converges on the + target's close group. +2. GETs are then sent one at a time, in XOR order. + +While the lookup runs, early GETs go to peers that were already connected. +A `ChunkGetResponse::NotFound` carries no peers, so a GET never helps +discovery. + +Measured on the production network, reading 150 chunks of one 612 MB file: + +- **Native reads spend most of their time finding a holder.** Discovery takes + 5.1 s at p50 and transfer 0.8 s, so 82% of read time is discovery. +- **The lookup runs past the point where it could stop.** It needs about 6 s to + converge, but the round that first reaches the close group is usually round + 2, about 0.3 s in. +- **Sending a GET beside every FIND_NODE removes almost all of that wait.** The + read stops at the first verified chunk. In the measurements: + + | | p50 | p90 | + | --- | ---: | ---: | + | Today | 5.95 s | 13.0 s | + | GET beside every FIND_NODE | 0.97 s | 1.9 s | + + The whole file then reads in 49–55 s instead of 199–205 s. +- **That speed costs bytes.** The round that reaches the close group queries + up to alpha holders at once, so the read downloads 2.5–3.3 copies of a chunk + on average. + +Kademlia's FIND_VALUE folds the value into the lookup: a node that holds the +key returns it; one that doesn't returns closer peers. This network cannot +just switch to that, for three reasons: + +- **Old peers.** Nodes on 0.20 and 0.21 drop an unknown `ChunkMessageBody` + variant. Natively they send no reply, so the client waits out its timeout. + Over WebRTC Direct, an undecodable request closes the browser's data channel. +- **No native capability exchange.** A connected node's user agent + (`node/0.21.0 migration/files`) is the only signal of what it supports. +- **Peer validation lives in saorsa-core.** FIND_NODE answers are checked + there: owner-signed address records, publish sequences, LAN filtering and + dial plans. A chunk-protocol message that carries peers must not bypass + those checks. + +## Decision Drivers + +- Cut native read latency to about one round trip per lookup step, and stop + the lookup as soon as a verified chunk arrives. +- Stay compatible in both directions: + - Old clients work unchanged with new nodes. + - New clients work with old nodes at full speed, not after timeouts. + - Rollout can happen in any order. +- Keep peer encoding and validation in saorsa-core, so a peer learned from a + chunk response is trusted exactly as far as one learned from FIND_NODE. +- Leave the chunk-absence rule unchanged: a completed lookup and a majority of + `NotFound`s. +- Give the client one place to evolve read policy without further wire + changes. + +## Considered Options + +1. **Status quo.** Native discovery stays at about 5 s per chunk. +2. **Lookup progress feeds the early GETs (client only).** Measured native p50 + 1.8 s, p90 4.9 s, one copy. No wire change, but more than 0.8 s slower at + p50 than options 3 and 5. +3. **GET beside each FIND_NODE (client only).** The same latency as option 5, + working with today's nodes. It sends two requests per peer, and the GET + carries nothing useful when the peer doesn't hold the chunk. It remains the + per-peer fallback under option 5. +4. **"Do you hold it?" request, then one fetch.** Measured native p50 1.16 s + with 1.07 copies. Keeps bytes down at the cost of an extra round trip + before the transfer. Deferred: it can be added later as another appended + variant. +5. **Chunk-or-closer-peers request in ant-protocol, with the closer peers in + saorsa-core's FIND_NODE answer format (chosen).** One request per lookup + step to upgraded nodes, with per-peer fallback to option 3 for older ones. +6. **FIND_VALUE inside saorsa-core's DHT.** Rejected: + - DHT messages are capped at 64 KiB, too small for a 4 MiB chunk. + - The DHT layer would have to know about chunk storage. + +## Decision + +We will add a chunk-or-closer-peers request to the chunk protocol. Native +clients will run their own iterative lookup with it, and fall back per peer to +FIND_NODE plus GET. + +### Wire (ant-protocol, additive) + +`ChunkMessageBody` gains two variants, appended after every existing one, so +existing discriminants keep their wire values: + +```rust +GetOrCloserRequest(ChunkGetOrCloserRequest { address: XorName }) +GetOrCloserResponse(ChunkGetOrCloserResponse) + +enum ChunkGetOrCloserResponse { + /// The responder holds the chunk. + Found { address: XorName, content: Bytes }, + /// The responder does not hold the chunk; `peers` are its closest known + /// peers to `address`, encoded by saorsa-core. + Closer { address: XorName, peers: Bytes }, + Error(ProtocolError), +} +``` + +`peers` is opaque to ant-protocol. saorsa-core produces and reads it with two +public calls: + +- **`DhtNetworkManager::encode_closer_peers(key, requester)`** builds exactly + what a FIND_NODE answer to that requester would hold: + - the K closest routing-table peers, without the requester + - QUIC addresses with publish sequences + - owner-signed address records, within the FIND_NODE size limit +- **`DhtNetworkManager::decode_closer_peers(responder, key, payload, + transport_source)`** validates the payload as a FIND_NODE answer from that + responder. It goes through the same signed-record verification, owner-view + protection, LAN filtering and dial-plan checks. + +The peer format is therefore versioned in one place, saorsa-core, as it is for +FIND_NODE today. + +### Node (ant-node) + +- **If the node holds the chunk,** it answers `Found`, read from storage like a + GET. +- **Otherwise it answers `Closer`,** using `encode_closer_peers` with the + authenticated sender as the requester. +- **It advertises support** with the user-agent token `get-or-closer/1` + (`ant_protocol::GET_OR_CLOSER_AGENT_TOKEN`), after `node/`. The + `node/` prefix that saorsa-core gates DHT membership on is kept. +- **It does not advertise support to browsers.** Browsers cannot decode the + native peer format. The WebRTC Direct listener keeps refusing the request, + and HELLO does not list it. + +### Native client (ant-client) + +A chunk read runs a client-driven iterative lookup on saorsa-core's public +lookup engine (`IterativeLookup`). The parameters match today's native lookup: +alpha 3, K, a 5 s grace period per round, and a 120 s deadline. + +Every peer the lookup queries answers it at once, but only peers that +probably hold the chunk are asked for it. A peer is a **likely holder** when +fewer than a close group of the peers the read knows of are nearer the chunk. +Until the read knows two close groups of peers, as in its first round, it +cannot tell, and every peer counts as one. + +For each peer the lookup queries, the client: + +1. **Connects to it.** It uses `DhtNetworkManager::connect_lookup_peer`, which + the lookup's own request would need anyway. It then reads the peer's user + agent. +2. **If the peer is the likely holder chosen to go next,** asks it for the + chunk and for closer peers in one step: + - If it advertises `get-or-closer/1`, sends `GetOrCloserRequest`: + - `Found`: verify the address and the content hash, then stop the read. + - `Closer`: `decode_closer_peers` supplies the lookup candidates. It + rejects a payload over the 64 KiB FIND_NODE limit before parsing it, + although the chunk message carrying it may be up to 5 MiB. + - Otherwise, sends FIND_NODE (`DhtNetworkManager::find_node_on_peer`) and + a GET at the same time. FIND_NODE supplies the candidates, and a + `Success` from the GET stops the read. +3. **If it is a likely holder that must wait its turn,** sends FIND_NODE now, + so the lookup never waits for the holder gate, and a GET when its turn + comes. A waiting peer that the lookup has since learnt is not among the + nearest is skipped. +4. **If it is far from the chunk,** sends FIND_NODE only, as today's lookup + does. +5. **If the connection fails,** marks the peer failed in the lookup, as a + failed FIND_NODE is marked today. + +The read driver owns every request, not the lookup round that triggered it. +The round's grace window therefore never cancels a chunk transfer that is +still arriving. Conflicting reports about a peer are resolved with +`client_routing::compute_winner`, as in both existing lookups. + +- **Likely holders are asked one at a time.** The next is asked once the + current one answers without the chunk, or after 500 ms if it is still + sending. At most two are asked at once, as today's read never runs more + than two GETs: on a link too slow to receive a chunk in 500 ms, further + copies would only slow each other. +- **The first likely holder is chosen uniformly**, among those the lookup + round has handed over, as the ordinary read chooses its first GET + ([ADR-0023](ADR-0023-spread-hot-reads-and-refuse-full-browser-sessions.md)). + Asking the nearest holder first would put every client's read of a popular + chunk on one node of its close group. +- **A peer whose content fails verification only drops out of the read.** + Likely holders are judged from what the lookup has learnt, and in its first + round any peer is one. Ending the read on one bad answer would therefore let + a dishonest node on a lookup path fail reads. +- **When the lookup ends without a chunk,** or fails, the client runs today's + read path (`retrieve_progressive`). Its absence rule, retry round and rule + that bad content ends the read are unchanged. Its first round starts from + the peers the lookup ended on instead of looking them up again, so reading + a chunk that is not stored still costs one lookup. +- **Browser clients keep today's read path.** A browser variant was built and + measured on a 60-node testnet (2026-10-12). Nodes took the request inside + `chunk_protocol` and listed it in HELLO, and answered a missing chunk with + their browser `find_node` reply. Against today's browser read it was 17% + slower from fra1 and 5% faster from syd1, neither significant. A warm + browser there already holds sessions with the chunk's close group, so + today's read sends its first GET to a holder, and a lookup cannot start + sooner than that. + +## Consequences + +### Positive + +- **Native read latency drops to about one round trip per lookup step.** + Measured with the per-peer fallback that this decision keeps for old nodes: + p50 5.95 s to about 1 s, p90 13 s to about 2 s, and the whole file 3.7 times + faster. +- **Today's nodes, through the per-peer fallback.** A client built from + this design downloaded the 612 MB production file in 56 s, against 242 s + for the current client, with identical bytes. Production nodes do not + announce the token yet, so every lookup step took the FIND_NODE-plus-GET + route, and all 150 chunks were found without falling back to the old + read path. +- **An upgraded likely holder answers its lookup step and the read in one + request,** instead of FIND_NODE plus GET. The chunk travels in that same + reply. +- **Node load stays as today's.** Peers that only route get FIND_NODE, which + the DHT layer answers, and only likely holders get a chunk request. Nodes + answer get-or-closer within the same 64-request limit as GETs and PUTs, so + under heavy uploads it waits as a GET does, but routing never waits there. +- **Compatible in both directions.** Old clients never send the new variant. + New clients send it only to nodes that advertise it, and treat every other + peer as before. +- **Peer trust is unchanged.** Peers learned from `Closer` pass the same + saorsa-core validation as FIND_NODE answers. +- **The absence rule, the retry policy and the browser path are untouched.** + +### Negative / Trade-offs + +- **Duplicate downloads.** Asking every queried peer for the chunk, as first + measured, downloaded 2.5–3.3 copies per chunk natively. The holder gate + bounds this to at most two copies in flight, the second only after 500 ms, + but a hedged copy can still arrive unused. A "have it" or inline-flag + variant can bound this further without changing the request defined here. +- **The native client runs its own lookup**, beside the one in saorsa-core. + Lookup parameters and peer policy must be kept in step. +- **Capability is a user-agent token**, and is known only once a peer is + connected. Every queried peer has to be connected before the client can + choose what to send it. The lookup's own request would have needed that + connection anyway. +- **A second chunk response shape** to maintain, with its own traffic + accounting. +- **A likely holder can send content even when it is far from the chunk,** + because the first round cannot tell. Each answer is bounded by the 5 MiB + chunk message limit, as a GET answer is today, and is verified before use. A + peer whose answer fails verification costs the client that transfer and + nothing more. +- **A waiting likely holder costs two requests,** FIND_NODE now and a GET + later, where a get-or-closer would have been one. + +### Neutral / Operational + +- **Version bumps.** ant-protocol takes a minor version bump for the appended + variants. saorsa-core gains public API with no wire change. +- **Node user agents gain a token.** Any tooling that parses them must split + on whitespace. +- **Rollout order is free.** Nodes are a no-op until clients use the request, + and clients fall back per peer. + +## Validation + +- **ant-protocol** + - Round-trip tests for the new messages. + - Every existing discriminant keeps its wire value. +- **saorsa-core** + - `encode_closer_peers` / `decode_closer_peers` return the same trusted + candidates as a FIND_NODE answer for the same key. + - `find_node_on_peer` and `connect_lookup_peer` behave against a live peer. + - `decode_closer_peers` rejects a payload over the FIND_NODE size limit + before parsing it, and reads one of exactly that size. +- **ant-node** + - The handler answers `Found` for a held chunk and `Closer`, with decodable + peers, otherwise. `Closer` leaves out the requester, which the same + node's answer to another peer names. + - A get-or-closer carries the same read telemetry as a GET. + - Devnet nodes announce the token, as production nodes do. + - The user agent carries the token and keeps the `node/` prefix. +- **ant-client** + - The read driver is tested on a scripted network. It must: + - stop at the first verified chunk + - fall back per peer for legacy peers + - handle a mixed network + - drop a peer whose content fails verification, and keep reading + - keep a transfer alive past its lookup round's grace window + - fall back to today's path when the lookup finds nothing, handing it the + peers the lookup ended on + - choose the first likely holder among a whole lookup round, not the + first step to arrive + - ask at most two likely holders at once + - answer the lookup at once for a likely holder that waits its turn, and + skip it if the lookup has since passed it + - ask peers far from the chunk only for closer peers + - In-process test networks: + - A network whose nodes all announce the token is read with + get-or-closer requests. One whose nodes do not is read with FIND_NODE + and GET, and is sent no get-or-closer request. + - The full end-to-end suite reads through the new path. + - Live: a new client against today's nodes takes the legacy route for every + peer and stays within the measured latency of option 3. +- **Review triggers** + - Node egress for reads rises enough to matter: add the "have it" variant. + - Browsers spend their reads finding a holder, because they no longer hold + sessions with the close group: revisit the browser variant above. + - saorsa-core changes its native lookup parameters: mirror them in the + client lookup. + +## Notes for AI-assisted work + +AI tools may help draft this ADR, but **must not mark it Accepted without human review**. Accepted ADRs are immutable: create a new superseding ADR rather than editing an Accepted ADR. diff --git a/docs/adr/README.md b/docs/adr/README.md index 43cc2028..c490a544 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -39,5 +39,6 @@ See [`TOOLING.md`](./TOOLING.md) for `adrs`, `adr-kit`, and AI harness setup. - [ADR-0017: Bounded fresh-replication offers and copy-free message sends](./ADR-0017-bounded-fresh-offers-and-copy-free-sends.md) - [ADR-0018: Pointer ownership transfer by final redirection](./ADR-0018-pointer-transfer-by-final-redirection.md) - [ADR-0019: Pointer audits serve the record round 1 bound](./ADR-0019-pointer-audits-serve-the-record-round-one-bound.md) +- [ADR-0020: Kademlia-style chunk GET: return the chunk or closer peers](./ADR-0020-kademlia-chunk-get-or-closer.md) - [ADR-0022: Remove the LMDB Chunk Store](./ADR-0022-remove-the-lmdb-chunk-store.md) - [ADR-0023: Spread hot-record reads across the close group, and refuse full browser sessions explicitly](./ADR-0023-spread-hot-reads-and-refuse-full-browser-sessions.md) diff --git a/src/ant_protocol/mod.rs b/src/ant_protocol/mod.rs index 74921117..c770301e 100644 --- a/src/ant_protocol/mod.rs +++ b/src/ant_protocol/mod.rs @@ -17,11 +17,12 @@ pub use ::ant_protocol::chunk; pub use ::ant_protocol::chunk::{ - settlement_compatibility, ChunkGetRequest, ChunkGetResponse, ChunkMessage, ChunkMessageBody, + advertises_get_or_closer, settlement_compatibility, ChunkGetOrCloserRequest, + ChunkGetOrCloserResponse, ChunkGetRequest, ChunkGetResponse, ChunkMessage, ChunkMessageBody, ChunkPutRequest, ChunkPutResponse, ChunkQuoteRequest, ChunkQuoteRequestV2, ChunkQuoteResponse, MerkleCandidateQuoteRequest, MerkleCandidateQuoteRequestV2, MerkleCandidateQuoteResponse, ProtocolError, SettlementCompatibility, XorName, CHUNK_PROTOCOL_ID, CLOSE_GROUP_MAJORITY, - CLOSE_GROUP_SIZE, CURRENT_SETTLEMENT_VERSION, DATA_TYPE_CHUNK, MAX_CHUNK_SIZE, - MAX_WIRE_MESSAGE_SIZE, MIN_SUPPORTED_SETTLEMENT_VERSION, PROOF_TAG_MERKLE, + CLOSE_GROUP_SIZE, CURRENT_SETTLEMENT_VERSION, DATA_TYPE_CHUNK, GET_OR_CLOSER_AGENT_TOKEN, + MAX_CHUNK_SIZE, MAX_WIRE_MESSAGE_SIZE, MIN_SUPPORTED_SETTLEMENT_VERSION, PROOF_TAG_MERKLE, PROOF_TAG_SINGLE_NODE, PROTOCOL_VERSION, XORNAME_LEN, }; diff --git a/src/devnet.rs b/src/devnet.rs index 18e459bb..4c836614 100644 --- a/src/devnet.rs +++ b/src/devnet.rs @@ -106,11 +106,10 @@ async fn handle_chunk_message( let handled = protocol .try_handle_request_with_context( &data, - Some(ChunkRequestContext::new( - source.to_string(), - received_at.into(), - Duration::ZERO, - )), + Some( + ChunkRequestContext::new(source.to_string(), received_at.into(), Duration::ZERO) + .with_requester(source), + ), ) .await; let telemetry = handled.get_telemetry; @@ -1012,6 +1011,9 @@ impl Devnet { .bootstrap_peers .clone_from(&node.bootstrap_addrs); core_config.diversity_config = Some(IPDiversityConfig::permissive()); + // Devnet nodes serve chunks, so they announce get-or-closer + // (ADR-0020) as a production node does. + core_config.custom_user_agent = Some(crate::node::user_agent(true)); let index = node.index; let p2p_node = P2PNode::new(core_config) diff --git a/src/node.rs b/src/node.rs index ac1f3588..390dd84c 100644 --- a/src/node.rs +++ b/src/node.rs @@ -1,6 +1,6 @@ //! Node implementation - thin wrapper around saorsa-core's `P2PNode`. -use crate::ant_protocol::CHUNK_PROTOCOL_ID; +use crate::ant_protocol::{CHUNK_PROTOCOL_ID, GET_OR_CLOSER_AGENT_TOKEN}; use crate::config::{ default_nodes_dir, default_root_dir, NetworkMode, NodeConfig, NODE_IDENTITY_FILENAME, }; @@ -53,8 +53,18 @@ const PROTOCOL_DRAIN_GRACE: std::time::Duration = std::time::Duration::from_secs /// that lost the prefix would stop being routed to. The version is this build's rather than /// the transport's, because it is the one a peer needs in order to tell which release it is /// talking to. -fn user_agent() -> String { - format!("node/{}", env!("CARGO_PKG_VERSION")) +/// +/// A node that serves chunks also carries the get-or-closer token (ADR-0020). Native peers +/// exchange no capabilities, and a node built before that request drops it without a reply, +/// so this token is how a client knows it may send one. A node with storage switched off +/// never answers chunk requests, so it does not announce it. +pub(crate) fn user_agent(serves_chunks: bool) -> String { + let agent = format!("node/{}", env!("CARGO_PKG_VERSION")); + if serves_chunks { + format!("{agent} {GET_OR_CLOSER_AGENT_TOKEN}") + } else { + agent + } } /// Builder for constructing an Ant node. @@ -334,7 +344,7 @@ impl NodeBuilder { } } - core_config.custom_user_agent = Some(user_agent()); + core_config.custom_user_agent = Some(user_agent(config.storage.enabled)); // Persist close group peers + trust scores across restarts. // Default to root_dir (alongside node_identity.key) when not explicitly set. @@ -968,6 +978,28 @@ impl RunningNode { Ok(()) } + /// Answer one inbound chunk-protocol message from `source` as a running + /// node's router does, and send the reply. For test networks that route + /// messages themselves. + #[cfg(any(test, feature = "test-utils"))] + pub async fn answer_chunk_request( + protocol: &Arc, + p2p: &Arc, + source: &saorsa_core::identity::PeerId, + data: &[u8], + ) { + Self::answer_one_request( + protocol, + p2p, + source, + data, + "chunk", + CHUNK_PROTOCOL_ID, + Instant::now(), + ) + .await; + } + /// Handle one inbound protocol message and send whatever it produced. async fn answer_one_request( protocol: &Arc, @@ -985,11 +1017,10 @@ impl RunningNode { let handled = protocol .try_handle_request_with_context( data, - Some(ChunkRequestContext::new( - source.to_string(), - received_at, - queue_wait, - )), + Some( + ChunkRequestContext::new(source.to_string(), received_at, queue_wait) + .with_requester(*source), + ), ) .await; let telemetry = handled.get_telemetry; @@ -1615,13 +1646,23 @@ mod tests { } /// The prefix is what keeps this node in its peers' routing tables, and the version is - /// this build's, with nothing after it. + /// this build's, with nothing after it on a node that does not serve chunks. #[test] fn the_user_agent_is_this_builds_version_under_the_node_prefix() { assert_eq!( - super::user_agent(), + super::user_agent(false), format!("node/{}", env!("CARGO_PKG_VERSION")) ); - assert!(super::user_agent().starts_with("node/")); + assert!(super::user_agent(true).starts_with("node/")); + } + + #[test] + fn user_agent_advertises_get_or_closer_only_when_serving_chunks() { + assert!(crate::ant_protocol::advertises_get_or_closer( + &super::user_agent(true) + )); + assert!(!crate::ant_protocol::advertises_get_or_closer( + &super::user_agent(false) + )); } } diff --git a/src/storage/handler.rs b/src/storage/handler.rs index 62e2db75..c4d1b075 100644 --- a/src/storage/handler.rs +++ b/src/storage/handler.rs @@ -29,10 +29,11 @@ use crate::ant_protocol::DATA_TYPE_CHUNK; use crate::ant_protocol::{ - settlement_compatibility, ChunkGetRequest, ChunkGetResponse, ChunkMessage, ChunkMessageBody, - ChunkPutRequest, ChunkPutResponse, ChunkQuoteRequest, ChunkQuoteRequestV2, ChunkQuoteResponse, - MerkleCandidateQuoteRequest, MerkleCandidateQuoteRequestV2, MerkleCandidateQuoteResponse, - ProtocolError, SettlementCompatibility, XorName, CHUNK_PROTOCOL_ID, CURRENT_SETTLEMENT_VERSION, + settlement_compatibility, ChunkGetOrCloserRequest, ChunkGetOrCloserResponse, ChunkGetRequest, + ChunkGetResponse, ChunkMessage, ChunkMessageBody, ChunkPutRequest, ChunkPutResponse, + ChunkQuoteRequest, ChunkQuoteRequestV2, ChunkQuoteResponse, MerkleCandidateQuoteRequest, + MerkleCandidateQuoteRequestV2, MerkleCandidateQuoteResponse, ProtocolError, + SettlementCompatibility, XorName, CHUNK_PROTOCOL_ID, CURRENT_SETTLEMENT_VERSION, MAX_CHUNK_SIZE, MIN_SUPPORTED_SETTLEMENT_VERSION, }; use crate::client::compute_address; @@ -48,7 +49,7 @@ use crate::storage::ChunkStore; use ant_protocol::chunk::{PointerGetResponse, PointerPutResponse}; use bytes::Bytes; use parking_lot::RwLock; -use saorsa_core::P2PNode; +use saorsa_core::{P2PNode, PeerId}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -76,6 +77,8 @@ fn duration_ms(duration: Duration) -> u64 { /// encoding, and the response hand-off to the transport. pub struct ChunkRequestContext { source_peer: String, + /// The authenticated sender, when the router knows it. + requester: Option, received_at: Instant, queue_wait_ms: u64, } @@ -85,10 +88,19 @@ impl ChunkRequestContext { pub(crate) fn new(source_peer: String, received_at: Instant, queue_wait: Duration) -> Self { Self { source_peer, + requester: None, received_at, queue_wait_ms: duration_ms(queue_wait), } } + + /// Record the authenticated sender, so a get-or-closer answer can leave + /// it out of the closer peers, as a `FIND_NODE` answer does. + #[must_use] + pub(crate) fn with_requester(mut self, requester: PeerId) -> Self { + self.requester = Some(requester); + self + } } /// A decoded protocol request and, for GETs, its pending one-shot telemetry. @@ -111,6 +123,8 @@ pub struct HandledChunkRequest { /// event with `cancelled` outcomes so the request does not disappear silently. #[cfg_attr(not(feature = "logging"), allow(dead_code))] pub struct GetRequestTelemetry { + /// The read request: `get`, or `get_or_closer` (ADR-0020). + request: &'static str, source_peer: String, request_id: u64, chunk_address: String, @@ -131,14 +145,61 @@ impl GetRequestTelemetry { storage_read_ms: u64, response: &ChunkGetResponse, ) -> Self { - let (storage_outcome, storage_error_category) = match response { + let storage = match response { ChunkGetResponse::Success { .. } => ("success", "none"), ChunkGetResponse::NotFound { .. } => ("not_found", "none"), ChunkGetResponse::Error(ProtocolError::StorageFailed(_)) => ("error", "storage"), ChunkGetResponse::Error(_) => ("error", "protocol"), _ => ("unknown", "unknown"), }; + Self::new( + "get", + context, + request_id, + chunk_address, + storage_read_ms, + storage, + ) + } + + /// A get-or-closer read (ADR-0020). Answering with closer peers means the + /// chunk was not in storage, so its storage outcome is `not_found`. + fn from_get_or_closer_response( + context: ChunkRequestContext, + request_id: u64, + chunk_address: String, + storage_read_ms: u64, + response: &ChunkGetOrCloserResponse, + ) -> Self { + let storage = match response { + ChunkGetOrCloserResponse::Found { .. } => ("success", "none"), + ChunkGetOrCloserResponse::Closer { .. } => ("not_found", "none"), + ChunkGetOrCloserResponse::Error(ProtocolError::StorageFailed(_)) => { + ("error", "storage") + } + ChunkGetOrCloserResponse::Error(_) => ("error", "protocol"), + _ => ("unknown", "unknown"), + }; + Self::new( + "get_or_closer", + context, + request_id, + chunk_address, + storage_read_ms, + storage, + ) + } + + fn new( + request: &'static str, + context: ChunkRequestContext, + request_id: u64, + chunk_address: String, + storage_read_ms: u64, + (storage_outcome, storage_error_category): (&'static str, &'static str), + ) -> Self { Self { + request, source_peer: context.source_peer, request_id, chunk_address, @@ -189,6 +250,7 @@ impl GetRequestTelemetry { let duration_ms = duration_ms(self.received_at.elapsed()); info!( target: "ant_node::storage::rpc_latency", + request = self.request, source_peer = %self.source_peer, request_id = self.request_id, chunk_address = %self.chunk_address, @@ -659,6 +721,13 @@ impl AntProtocol { let key = ChunkResponseKey::of_pointer_put(&response); (ChunkMessageBody::PointerPutResponse(response), key) } + ChunkMessageBody::GetOrCloserRequest(req) => { + let (response, telemetry) = + self.handle_get_or_closer(req, request_id, context).await; + get_telemetry = telemetry; + let key = ChunkResponseKey::of_get_or_closer(&response); + (ChunkMessageBody::GetOrCloserResponse(response), key) + } ChunkMessageBody::PointerGetRequest(req) => { let started = Instant::now(); let address = req.address; @@ -922,6 +991,83 @@ impl AntProtocol { } } + /// Answer a get-or-closer request (ADR-0020): the chunk when this node + /// holds it, otherwise the `FIND_NODE` answer it would give the requester + /// for the address. With a request context, also returns the read's + /// telemetry, as a GET's. + async fn handle_get_or_closer( + &self, + request: ChunkGetOrCloserRequest, + request_id: u64, + context: Option, + ) -> (ChunkGetOrCloserResponse, Option) { + let address = request.address; + let requester = context.as_ref().and_then(|context| context.requester); + let storage_started = Instant::now(); + let stored = self.storage.get(&address).await; + let storage_read_ms = duration_ms(storage_started.elapsed()); + let response = match stored { + Ok(Some(chunk)) => ChunkGetOrCloserResponse::Found { + address, + content: Bytes::from(chunk), + }, + Ok(None) => self.closer_peers(address, requester).await, + Err(e) => { + warn!( + "Failed to read chunk {} for get-or-closer: {e}", + hex::encode(address) + ); + ChunkGetOrCloserResponse::Error(ProtocolError::StorageFailed(e.to_string())) + } + }; + let telemetry = context.map(|context| { + GetRequestTelemetry::from_get_or_closer_response( + context, + request_id, + hex::encode(address), + storage_read_ms, + &response, + ) + }); + (response, telemetry) + } + + /// This node's closest known peers to `address`, encoded by saorsa-core + /// exactly as its `FIND_NODE` answer to `requester` would hold them. + async fn closer_peers( + &self, + address: XorName, + requester: Option, + ) -> ChunkGetOrCloserResponse { + // Bind the handle out of the lock so no guard is held across `.await`. + let attached = self.p2p_node.read().as_ref().map(Arc::clone); + let Some(p2p) = attached else { + return ChunkGetOrCloserResponse::Error(ProtocolError::Internal( + "this node has no routing table to answer from".to_string(), + )); + }; + // Without a known requester there is no one to leave out; the local + // node never appears in its own answer. + let requester = requester.unwrap_or_else(|| *p2p.peer_id()); + match p2p + .dht_manager() + .encode_closer_peers(&address, &requester) + .await + { + Ok(peers) => ChunkGetOrCloserResponse::Closer { + address, + peers: Bytes::from(peers), + }, + Err(e) => { + warn!( + "Failed to encode closer peers for {}: {e}", + hex::encode(address) + ); + ChunkGetOrCloserResponse::Error(ProtocolError::Internal(e.to_string())) + } + } + } + /// Resync the quoting metric to the authoritative count of records the node /// actually holds. /// @@ -1451,6 +1597,62 @@ mod tests { } } + #[tokio::test] + async fn test_get_or_closer_returns_a_held_chunk() { + let (protocol, _temp) = create_test_protocol().await; + let content = b"get-or-closer content"; + let address = compute_address(content); + protocol + .put_local(&address, content) + .await + .expect("store chunk"); + let request = ChunkMessage { + request_id: 30, + body: ChunkMessageBody::GetOrCloserRequest(ChunkGetOrCloserRequest::new(address)), + }; + + let response_bytes = protocol + .try_handle_request(&request.encode().expect("encode")) + .await + .expect("handle get-or-closer") + .expect("expected response"); + let response = ChunkMessage::decode(&response_bytes).expect("decode response"); + + assert_eq!(response.request_id, 30); + let ChunkMessageBody::GetOrCloserResponse(ChunkGetOrCloserResponse::Found { + address: found, + content: found_content, + }) = response.body + else { + panic!("expected GetOrCloserResponse::Found"); + }; + assert_eq!(found, address); + assert_eq!(found_content.as_ref(), content); + } + + #[tokio::test] + async fn test_get_or_closer_without_a_routing_table_is_an_error() { + // A node always has a routing table; a protocol without an attached + // P2P node must refuse rather than claim it knows no closer peers. + let (protocol, _temp) = create_test_protocol().await; + let request = ChunkMessage { + request_id: 31, + body: ChunkMessageBody::GetOrCloserRequest(ChunkGetOrCloserRequest::new([0xCD; 32])), + }; + + let response_bytes = protocol + .try_handle_request(&request.encode().expect("encode")) + .await + .expect("handle get-or-closer") + .expect("expected response"); + let response = ChunkMessage::decode(&response_bytes).expect("decode response"); + + assert!(matches!( + response.body, + ChunkMessageBody::GetOrCloserResponse(ChunkGetOrCloserResponse::Error(_)) + )); + } + #[tokio::test] async fn test_get_telemetry_preserves_join_fields_and_response() { let (protocol, _temp) = create_test_protocol().await; @@ -1475,6 +1677,7 @@ mod tests { .await; let telemetry = handled.get_telemetry.expect("GET telemetry"); + assert_eq!(telemetry.request, "get"); assert_eq!(telemetry.source_peer, "peer-v2-995"); assert_eq!(telemetry.request_id, request_id); assert_eq!(telemetry.chunk_address, hex::encode(address)); @@ -1497,6 +1700,45 @@ mod tests { telemetry.finish_send(Duration::from_millis(3), true); } + /// A get-or-closer is a read, so it carries the same telemetry as a GET. + #[tokio::test] + async fn test_get_or_closer_found_carries_read_telemetry() { + let (protocol, _temp) = create_test_protocol().await; + let content = b"get-or-closer telemetry"; + let address = compute_address(content); + protocol + .put_local(&address, content) + .await + .expect("store chunk"); + let request_id = 996; + let request = ChunkMessage { + request_id, + body: ChunkMessageBody::GetOrCloserRequest(ChunkGetOrCloserRequest::new(address)), + }; + + let handled = protocol + .try_handle_request_with_context( + &request.encode().expect("encode get-or-closer"), + Some(ChunkRequestContext::new( + "peer-v2-996".to_string(), + Instant::now(), + Duration::ZERO, + )), + ) + .await; + + let telemetry = handled.get_telemetry.expect("get-or-closer telemetry"); + assert_eq!(telemetry.request, "get_or_closer"); + assert_eq!(telemetry.request_id, request_id); + assert_eq!(telemetry.chunk_address, hex::encode(address)); + assert_eq!(telemetry.storage_outcome, "success"); + assert_eq!( + handled.traffic_key, + Some(ChunkResponseKey::GetOrCloserFound) + ); + telemetry.finish_send(Duration::from_millis(3), true); + } + #[tokio::test] async fn test_put_address_mismatch() { let (protocol, _temp) = create_test_protocol().await; diff --git a/src/storage/traffic.rs b/src/storage/traffic.rs index 35688e8f..23e1672b 100644 --- a/src/storage/traffic.rs +++ b/src/storage/traffic.rs @@ -12,7 +12,9 @@ use std::sync::atomic::{AtomicU64, Ordering}; -use crate::ant_protocol::{ChunkGetResponse, ChunkMessageBody, ChunkPutResponse}; +use crate::ant_protocol::{ + ChunkGetOrCloserResponse, ChunkGetResponse, ChunkMessageBody, ChunkPutResponse, +}; use ant_protocol::chunk::{PointerGetResponse, PointerPutResponse}; /// Kind of an inbound chunk message, for the rx table. @@ -31,10 +33,12 @@ pub enum ChunkRequestKind { Other, /// Bytes that failed `ChunkMessage::decode`. DecodeError, + /// A get-or-closer request (ADR-0020). + GetOrCloser, } impl ChunkRequestKind { - const N: usize = 10; + const N: usize = 11; const fn index(self) -> usize { match self { @@ -48,6 +52,7 @@ impl ChunkRequestKind { Self::PointerPut => 7, Self::Other => 8, Self::DecodeError => 9, + Self::GetOrCloser => 10, } } } @@ -77,10 +82,15 @@ pub enum ChunkResponseKey { /// A response variant this table does not itemise (e.g. an `Other` /// outcome on a `#[non_exhaustive]` enum). Other, + /// A get-or-closer answer (ADR-0020) carrying the chunk. + GetOrCloserFound, + /// A get-or-closer answer carrying closer peers. + GetOrCloserCloser, + GetOrCloserError, } impl ChunkResponseKey { - const N: usize = 20; + const N: usize = 23; const fn index(self) -> usize { match self { @@ -104,6 +114,9 @@ impl ChunkResponseKey { Self::PointerPutPaymentRequired => 17, Self::PointerPutError => 18, Self::Other => 19, + Self::GetOrCloserFound => 20, + Self::GetOrCloserCloser => 21, + Self::GetOrCloserError => 22, } } } @@ -120,12 +133,23 @@ impl ChunkRequestKind { ChunkMessageBody::MerkleCandidateQuoteRequestV2(_) => Self::MerkleQuoteV2, ChunkMessageBody::PointerGetRequest(_) => Self::PointerGet, ChunkMessageBody::PointerPutRequest(_) => Self::PointerPut, + ChunkMessageBody::GetOrCloserRequest(_) => Self::GetOrCloser, _ => Self::Other, } } } impl ChunkResponseKey { + /// Classify a get-or-closer response by outcome. + pub fn of_get_or_closer(response: &ChunkGetOrCloserResponse) -> Self { + match response { + ChunkGetOrCloserResponse::Found { .. } => Self::GetOrCloserFound, + ChunkGetOrCloserResponse::Closer { .. } => Self::GetOrCloserCloser, + ChunkGetOrCloserResponse::Error(_) => Self::GetOrCloserError, + _ => Self::Other, + } + } + /// Classify a GET response by outcome. pub fn of_get(response: &ChunkGetResponse) -> Self { match response { @@ -211,11 +235,12 @@ pub fn record_send_failed(bytes: usize) { /// Emit the cumulative chunk-RPC traffic as INFO summary lines, target /// `ant_node::storage::traffic`. /// -/// Flat snake-case keys like the replication summary. Three lines sharing the +/// Flat snake-case keys like the replication summary. Four lines sharing the /// same target and message, distinguished by `group`: rx by request kind -/// (`group = 1`), chunk and quote tx by kind × outcome (`group = 2`), and -/// pointer tx by kind × outcome (`group = 3`). Pointers take a line of their -/// own because `group = 2` is already close to `tracing`'s 32-field cap. +/// (`group = 1`), chunk and quote tx by kind × outcome (`group = 2`), pointer +/// tx by kind × outcome (`group = 3`), and get-or-closer rx and tx +/// (`group = 4`). Pointers and get-or-closer take lines of their own because +/// `group = 2` is already close to `tracing`'s 32-field cap. pub fn log_chunk_rpc_traffic_summary() { use ChunkRequestKind as Q; use ChunkResponseKey as R; @@ -286,6 +311,20 @@ pub fn log_chunk_rpc_traffic_summary() { pointer_put_error_tx_count = tc(R::PointerPutError), "chunk rpc traffic summary (cumulative)" ); + + crate::logging::info!( + target: "ant_node::storage::traffic", + group = 4, + get_or_closer_rx_bytes = rb(Q::GetOrCloser), + get_or_closer_rx_count = rc(Q::GetOrCloser), + get_or_closer_found_tx_bytes = tb(R::GetOrCloserFound), + get_or_closer_found_tx_count = tc(R::GetOrCloserFound), + get_or_closer_closer_tx_bytes = tb(R::GetOrCloserCloser), + get_or_closer_closer_tx_count = tc(R::GetOrCloserCloser), + get_or_closer_error_tx_bytes = tb(R::GetOrCloserError), + get_or_closer_error_tx_count = tc(R::GetOrCloserError), + "chunk rpc traffic summary (cumulative)" + ); } #[cfg(test)] @@ -305,6 +344,7 @@ mod tests { ChunkRequestKind::PointerPut, ChunkRequestKind::Other, ChunkRequestKind::DecodeError, + ChunkRequestKind::GetOrCloser, ]; let mut seen = std::collections::HashSet::new(); for k in req { @@ -332,6 +372,9 @@ mod tests { ChunkResponseKey::PointerPutPaymentRequired, ChunkResponseKey::PointerPutError, ChunkResponseKey::Other, + ChunkResponseKey::GetOrCloserFound, + ChunkResponseKey::GetOrCloserCloser, + ChunkResponseKey::GetOrCloserError, ]; let mut seen = std::collections::HashSet::new(); for k in resp { diff --git a/tests/e2e/data_types/chunk.rs b/tests/e2e/data_types/chunk.rs index 650b8e08..7cba5895 100644 --- a/tests/e2e/data_types/chunk.rs +++ b/tests/e2e/data_types/chunk.rs @@ -294,6 +294,135 @@ mod tests { .expect("Failed to teardown harness"); } + /// The peers a get-or-closer answer from `responder` names for `key`, + /// decoded on `decoder` as a `FIND_NODE` answer. + async fn decode_closer_peers( + decoder: &saorsa_core::P2PNode, + responder: &saorsa_core::identity::PeerId, + key: &[u8; 32], + peers: &[u8], + transport_source: Option<&saorsa_core::MultiAddr>, + ) -> Vec { + decoder + .dht_manager() + .decode_closer_peers(responder, key, peers, transport_source) + .await + .expect("closer peers should decode as a FIND_NODE answer") + .into_iter() + .map(|node| node.peer_id) + .collect() + } + + /// ADR-0020: a node answers get-or-closer with the chunk when it holds + /// it, and otherwise with closer peers that the requester checks as a + /// `FIND_NODE` answer. + #[tokio::test(flavor = "multi_thread")] + #[allow(clippy::panic)] + async fn test_get_or_closer_returns_chunk_or_closer_peers() { + let harness = TestHarness::setup_minimal() + .await + .expect("Failed to setup test harness"); + let fixture = ChunkTestFixture::new(); + let requester = harness.test_node(3).expect("Node 3 should exist"); + let p2p = requester + .p2p_node + .as_ref() + .expect("Node 3 should be running"); + let peers = requester.connected_peers().await; + let target = peers.first().expect("Node 3 should have a connected peer"); + + let address = ChunkTestFixture::compute_address(&fixture.small); + harness.prepopulate_payment_cache_for_peer(target, &address); + requester + .store_chunk_on_peer(target, &fixture.small) + .await + .expect("Failed to store chunk on remote node"); + + let (found, _) = requester + .get_or_closer_from_peer(target, &address) + .await + .expect("get-or-closer should be answered"); + let ant_node::ant_protocol::ChunkGetOrCloserResponse::Found { + address: found_address, + content, + } = found + else { + panic!("expected the holder to answer Found, got {found:?}"); + }; + assert_eq!(found_address, address); + assert_eq!(content.as_ref(), fixture.small.as_slice()); + + // No node holds this address, so every node answers with peers. It is + // the requester's own ID, so the requester is the nearest peer the + // responder could name. + let missing = *p2p.peer_id().as_bytes(); + let (closer, transport_source) = requester + .get_or_closer_from_peer(target, &missing) + .await + .expect("get-or-closer should be answered"); + let ant_node::ant_protocol::ChunkGetOrCloserResponse::Closer { + address: closer_address, + peers, + } = closer + else { + panic!("expected a non-holder to answer Closer, got {closer:?}"); + }; + assert_eq!(closer_address, missing); + + // A third node decodes the answers, since a node leaves itself out of + // any answer it decodes. Another node asking the same shows that the + // responder does name the requester to anyone else. + let others = (0..harness.node_count()) + .filter_map(|index| harness.test_node(index)) + .filter(|node| { + node.p2p_node.as_ref().is_some_and(|other| { + other.peer_id() != p2p.peer_id() && other.peer_id() != target + }) + }) + .collect::>(); + // The request goes out without dial addresses, so the bystander must + // already be connected to the responder. + let mut bystander = None; + for node in &others { + if node.connected_peers().await.contains(target) { + bystander = Some(*node); + break; + } + } + let bystander = bystander.expect("another node should be connected to the responder"); + let decoder = others + .iter() + .filter(|node| !std::ptr::eq(**node, bystander)) + .find_map(|node| node.p2p_node.as_ref()) + .expect("a fourth node should exist"); + let named = + decode_closer_peers(decoder, target, &missing, &peers, transport_source.as_ref()).await; + assert!(!named.is_empty(), "the responder should know other peers"); + assert!( + !named.contains(p2p.peer_id()), + "the answer should leave out the requester" + ); + + let (control, control_source) = bystander + .get_or_closer_from_peer(target, &missing) + .await + .expect("get-or-closer should be answered"); + let ant_node::ant_protocol::ChunkGetOrCloserResponse::Closer { peers, .. } = control else { + panic!("expected a non-holder to answer Closer, got {control:?}"); + }; + assert!( + decode_closer_peers(decoder, target, &missing, &peers, control_source.as_ref()) + .await + .contains(p2p.peer_id()), + "another node's answer should name the requester" + ); + + harness + .teardown() + .await + .expect("Failed to teardown harness"); + } + /// Test 8: Reject oversized chunk (> 4MB). /// /// Chunks have a maximum size of 4MB. Attempting to store a larger diff --git a/tests/e2e/testnet.rs b/tests/e2e/testnet.rs index ca33a901..7dddd407 100644 --- a/tests/e2e/testnet.rs +++ b/tests/e2e/testnet.rs @@ -14,8 +14,9 @@ //! - chunk store persistence use ant_node::ant_protocol::{ - ChunkGetRequest, ChunkGetResponse, ChunkMessage, ChunkMessageBody, ChunkPutRequest, - ChunkPutResponse, CHUNK_PROTOCOL_ID, MAX_WIRE_MESSAGE_SIZE, + ChunkGetOrCloserRequest, ChunkGetOrCloserResponse, ChunkGetRequest, ChunkGetResponse, + ChunkMessage, ChunkMessageBody, ChunkPutRequest, ChunkPutResponse, CHUNK_PROTOCOL_ID, + MAX_WIRE_MESSAGE_SIZE, }; use ant_node::client::{send_and_await_chunk_response, DataChunk, XorName}; use ant_node::payment::{ @@ -865,6 +866,46 @@ impl TestNode { .await } + /// Send a get-or-closer request (ADR-0020) to a peer, returning its + /// answer and the transport address that delivered it. + /// + /// # Errors + /// + /// Returns an error if the node is not running, the request cannot be + /// encoded or sent, or no answer arrives in time. + pub async fn get_or_closer_from_peer( + &self, + target_peer_id: &PeerId, + address: &XorName, + ) -> Result<(ChunkGetOrCloserResponse, Option)> { + let p2p = self.p2p_node.as_ref().ok_or(TestnetError::NodeNotRunning)?; + let request_id: u64 = rand::thread_rng().gen(); + let message = ChunkMessage { + request_id, + body: ChunkMessageBody::GetOrCloserRequest(ChunkGetOrCloserRequest::new(*address)), + }; + let message_bytes = message.encode().map_err(|e| { + TestnetError::Serialization(format!("Failed to encode get-or-closer request: {e}")) + })?; + let timeout = Duration::from_secs(DEFAULT_CHUNK_OPERATION_TIMEOUT_SECS); + let response = ant_protocol::send_and_await_chunk_response_with_metadata( + p2p, + target_peer_id, + message_bytes, + request_id, + timeout, + &[], + |body| match body { + ChunkMessageBody::GetOrCloserResponse(response) => Some(Ok(response)), + _ => None, + }, + |e| TestnetError::Retrieval(format!("Failed to send get-or-closer: {e}")), + || TestnetError::Retrieval("Timeout waiting for a get-or-closer answer".to_string()), + ) + .await?; + Ok((response.result?, response.transport_source)) + } + /// Compute content address for chunk data (BLAKE3 hash). #[must_use] pub fn compute_chunk_address(data: &[u8]) -> XorName { @@ -1325,32 +1366,13 @@ impl TestNetwork { ); let protocol = Arc::clone(&protocol_clone); let p2p = Arc::clone(&p2p_clone); + // As a node's own router answers it, so the + // handler knows the authenticated sender. tokio::spawn(async move { - match protocol.try_handle_request(&data).await { - Ok(Some(response)) => { - if let Err(e) = p2p - .send_message( - &source, - CHUNK_PROTOCOL_ID, - response.to_vec(), - &[], - ) - .await - { - warn!( - "Node {node_index} failed to send chunk response to {source}: {e}" - ); - } - } - Ok(None) => { - // Response message — no reply needed - } - Err(e) => { - warn!( - "Node {node_index} chunk protocol handler error: {e}" - ); - } - } + ant_node::RunningNode::answer_chunk_request( + &protocol, &p2p, &source, &data, + ) + .await; }); } }