diff --git a/Cargo.lock b/Cargo.lock index 998f8cf1..2cb0db82 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -884,8 +884,7 @@ dependencies = [ [[package]] name = "ant-protocol" version = "4.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c269f3a7c2e117cf444b97a8326d3583a8b016b3ce688c39ec5310bdebc367c5" +source = "git+https://github.com/WithAutonomi/ant-protocol?rev=718f60682bb3704b68fbba0bea34f2a0416fa876#718f60682bb3704b68fbba0bea34f2a0416fa876" dependencies = [ "blake3", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 6e658ece..15851629 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -214,6 +214,13 @@ webrtc-direct = [ "saorsa-transport/webrtc-direct", ] +[patch.crates-io] +# The published 4.0.0 carries pointers (ADR-0016) but not the final state this +# change relies on (ADR-0018). A rev, not a branch, so the pin is immutable. +# Drop it once a release includes the companion change. Versions are bumped at +# release. +ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "718f60682bb3704b68fbba0bea34f2a0416fa876" } + [profile.release] lto = true codegen-units = 1 diff --git a/docs/adr/ADR-0018-pointer-transfer-by-final-redirection.md b/docs/adr/ADR-0018-pointer-transfer-by-final-redirection.md new file mode 100644 index 00000000..efa664fd --- /dev/null +++ b/docs/adr/ADR-0018-pointer-transfer-by-final-redirection.md @@ -0,0 +1,319 @@ +# ADR-0018: Pointer ownership transfer by final redirection + +- **Status:** Proposed +- **Date:** 2026-09-29 +- **Decision owners:** Anselme (@grumbach) +- **Reviewers:** TBD +- **Supersedes:** none. Amends ADR-0016's merge rule at the final counter. +- **Superseded by:** none +- **Related:** ADR-0016 (pointers), ADR-0005 (repair quorum); V2-1354; WithAutonomi/ant-protocol#40, WithAutonomi/ant-node#239, WithAutonomi/ant-client#210 + +## Context + +ADR-0016 fixes a pointer's owner key at creation and offers handover only as +indirection: the owner points the pointer at a pointer the recipient owns. The +former owner keeps its key, so that is a revocable forwarding and not a sale. +The one thing that could make it stick is the counter running out, and under +ADR-0016 it does not: + +```text +1. larger counter +2. smaller target bytes +``` + +At `counter == u64::MAX` no counter is larger, but rule 2 still applies. A +former owner who handed the address over at `u64::MAX` can sign another state +at `u64::MAX` whose target sorts first — about two tries of grinding — and it +displaces the handover on every node, deterministically. ADR-0016 says so +("does not even freeze the pointer") and tells owners to migrate *before* the +final counter, which is exactly the revocable forwarding again. + +People want to hand pointers over — a name, an application's root, a published +handle — while every reader keeps using the same address. The address derives +from the owner key, so it cannot follow a new key. What the address *resolves +to* can. + +## Decision Drivers + +- The address readers use must not change. +- No new record type, field, message or storage format; nothing a node has to + interpret beyond the counter it already compares. +- Once a node holds a transfer, no arrival may move it off, the former + owner's included. +- Forks the owner can still make must be detectable by any reader, and must + not be able to take a transfer back from the nodes that hold it. +- Below the final counter nothing changes: ADR-0016's convergence argument + still holds there. + +## Considered Options + +1. **ADR-0016 as is.** Handover by revocable forwarding. Not a transfer. +2. **Certificates and epochs** (ADR-0016's withdrawn alternative): a transfer + certificate chain, an admission index per epoch and a 5-of-7 branch quorum. + Real re-keying, at the cost of a second record type, lineage walks, Sybil + exposure in ownership and a stuck-not-reversed failure mode under churn. +3. **Final state resolved by target, as today.** Grindable in about two tries. +4. **Final state resolved by payment time or order.** A pre-buy defeats it: + pay early, withhold, publish later. Clock trust besides. +5. **A final state is replaced by nothing; the first one a node takes is the + one it keeps.** Clients settle the owner's only remaining fork by the close + group's majority, and nodes look before taking one. + +## Decision + +We choose option 5. + +### Merge + +One rule ahead of ADR-0016's two: + +```text +0. a final state (counter == u64::MAX) is replaced by nothing +1. larger counter +2. smaller target bytes +``` + +Below the final counter the order is ADR-0016's, total and deterministic. At +it, two *different* final states are unordered, so each node keeps whichever it +took first. `replaces` stays a strict partial order — never both ways round, +transitive — and every pair of distinct states is ordered except two final +ones. + +This lives in `ant_protocol::pointer::PointerState::replaces`, so the node's +store, its admission gate, fresh offers, repair and hints all take it without a +line of their own. + +### What this changes in ADR-0016 + +ADR-0016 is left as written. Where the two disagree, this ADR holds, at the +final counter only: + +- ADR-0016's merge rule gains rule 0 above. +- "An owner who jumps straight to `u64::MAX` ... does not even freeze the + pointer" no longer holds: a state at `u64::MAX` is the last one the pointer + holds on every node that takes it. +- "Ownership cannot change ... a revocable forwarding state, not a sale" no + longer holds for a forwarding signed at the final counter: no node that holds + it gives it up. The owner key still cannot change, and a forwarding below the + final counter is still revocable. + +### Transfer + +A transfer is the final state whose target is another pointer: + +```text +counter = u64::MAX +target = (Pointer, recipient) recipient = the new owner's pointer address +``` + +`Pointer::transfer_to` signs it; `transferred_to()` recognises it. A final state +with any other target freezes the pointer — nobody receives it. A pointer target +below the final counter is still ADR-0016's revocable forwarding. + +Readers already follow pointer targets (`pointer_resolve`), so a transferred +address resolves through the recipient's pointer, which only the recipient can +move. The recipient's pointer should serve that one address: a pointer's +address derives from its owner key, so everything handed to the same recipient +pointer resolves to the same place. A recipient uses a fresh key per received +pointer. + +### The fork that remains + +Only the owner can sign a final state, so only the owner can fork one. It +keeps the earlier record and its key, so it can sign a second final state at +once or much later, and any node the first has not reached will take it. Each +node keeps its first. Nothing local can settle that, and this design does not +try. What it guarantees instead: + +- **No fork after the fact.** Once a node holds a final state, no arrival — + client PUT, fresh offer or repair — moves it off it. A second final state can + only land on a node that holds none. +- **Nodes look before a final state.** Before taking a final state it does not + hold, from a client or from a fresh offer, a node asks its close group which + state each holds. A peer claiming a *different* final state is asked for the + record, and if it is the state claimed and verifies — only the owner could + have signed it — the node + refuses, answering `Stale` with the state the group proved. A claim alone + refuses nothing, so one dishonest peer cannot block a transfer. Each peer's + question and fetch run as one pipeline, all at once, so a peer that claims a + rival and then stalls its fetch cannot hide another peer's proof until the + budget runs out. The look runs after payment is verified, asks only peers + that have sent a pointer message, and is bounded at four seconds; silence + proves nothing and the write goes ahead. This is what closes the gap the + merge rule leaves: a node that joined the group after the transfer would + otherwise take a second final state on the merge rule alone. +- **A look costs one round, once.** A payment proof, once verified, is cached, + so replaying one paid final state costs its sender nothing after the first + time. A node therefore remembers each final state a look proved, up to two + per address and 16,384 addresses, oldest forgotten first, and never lets a + later proof replace an earlier one; it refuses a replayed loser from that + memory, ahead of the signature check and without asking anyone. A look that + found nothing answers the same state again for two seconds without asking, + enough for replays queued behind it, and not at all once the write it + cleared has failed. + Looks for one address wait their turn, so replays queued behind a look find + its answer, and a burst costs one round; at most 64 run at once, and one + that cannot start within two seconds answers the PUT with an error and + drops the fresh offer, neither taking nor refusing the state for good; the + client retries such an error with the proof it already paid for. Looks are + spread over the address's last byte, since the addresses one node holds + share their leading bits. + Two seconds of waiting and four of looking add at most six seconds to a + PUT after its payment is verified; a PUT whose payment check is itself slow + can outlast the client's ten-second store timeout, and the client then + retries it as a missing answer with the proof it already paid for. Flooding + past all this needs a new paid final state per round. +- **A node restores its own final state without looking.** A node that lost + the file of a final state it held is admitted that exact state again and + nothing else (ADR-0016's lost-record rule), so taking it back is a restore, + not a new final state. It asks nobody: a peer on the other side of a fork + would otherwise keep it from restoring its own copy for good. +- **Readers see the majority, and see forks.** A client read that meets a final + state is settled only once one final state is held by a majority of the close + group, and returns that one. If two different final states are seen and + neither has a majority, the read fails as forked rather than guess. One final + state below a majority with no rival is returned once corroborated, as any + state is: it is a transfer still spreading. +- **Recipients check before they rely on it.** `pointer_finality` asks the + whole group and answers `Open`, `Settling` (one final state, short of a + majority), `Final` (one final state, a majority holds it, no rival seen) or + `Forked` (rivals seen; the majority's state if there is one). A recipient + treats the transfer as done only on `Final`; anything else is the former + owner's to explain. + +### Replication + +- **Repair** is the merge rule over quorum-backed states, so a node holding a + final state adopts nothing else, and a node holding none adopts the final + state a quorum holds. Between two final states a group wide enough to back + both at quorum — never a seven-node group at four — the larger side is + adopted, not whichever answered first. It must be strictly larger even + counting every peer that did not answer for its rival, so a tie, seen or + possibly hidden behind silent peers, adopts neither until the group + settles. +- **Hints** for a different final state are dropped by a node holding one, + since the hint cannot replace it. No refetch loop. +- **Possession.** A member holding a *different final* state is not penalised + when checked for the one this node offered: it holds what the merge rule told + it to, and the fork is the owner's. It is logged at warn. A member holding + nothing is penalised as before. + +### Client + +- `pointer_transfer` refuses before paying if the pointer is already final, if + the recipient is the pointer itself, if the recipient pointer does not exist, + or if the recipient's chain leads back to this address. It then signs, + pays, stores and reports `pointer_finality`. +- `pointer_update` on a final pointer refuses before paying, as it always did + when the counter could not advance. +- `pointer_controller` follows transfers — final pointer-targeted states only — + to the pointer whose owner now decides what the address resolves to. + +## Consequences + +### Positive + +- Pointers can be handed over for good, and the address readers use never + changes. +- No new record, field, message or storage format. The wire is untouched; the + only protocol change is one comparison. +- Once the handover lands on a node, the former owner has no move there: no + larger counter exists and no equal one replaces. +- A handover can be chained: the recipient can transfer its own pointer on. +- ADR-0016's "revocable forwarding" and "migrate before the final counter" + caveats are gone: migration *is* the final update. + +### Negative / Trade-offs + +- **A race is a permanent fork.** An owner who signs two final states and races + them leaves each node on its first, forever. With a majority on one side, + reads return that side and flag the fork to anyone who checks finality. With + no majority, reads of that pointer fail as forked, and nothing — not the + owner, not repair — can fix it. Only the owner can do this, and only to its + own pointer, as only the owner can lose its key. +- **A transfer is irreversible.** One sent to the wrong recipient pointer is + gone. The client checks the recipient exists and does not loop back; it + cannot check intent. +- **The look before a final state is not a lock.** It is fail-open on silence + and bounded in time, and it only runs when a node has learned which peers + understand pointers, which takes a sync round after joining. A node asked in + that first round, or one whose group cannot answer in four seconds, takes a + second final state on the merge rule alone, and so does a node running with + no replication, a local devnet or one whose replication engine failed to + start, since there is nothing to ask the group with. Reads still return the + majority side; the residual risk is a minority fork that `pointer_finality` + reports. +- Under a flood of distinct paid final states a node answers some honest final + PUTs with an error rather than look for them late. The client retries them + with the proof it already paid for; a write that still falls short fails, + and the owner tries again. +- A final PUT costs its node one state query per capable close-group peer, and + a fetch per claimed rival, before it commits. +- **Mixed fleets.** A node on ADR-0016's rule still lets a smaller-target final + state displace the first. Until a close group's majority runs this rule a + former owner can win back the nodes that do not. Reads follow the majority, so + the handover holds wherever most of the group has upgraded; recipients should + wait for the upgrade before relying on a transfer. Nothing on the wire tells + the two rules apart: the pointer format version is unchanged, so no node can + refuse to replicate with a peer on the old rule. +- Transferred reads take one extra hop, and a chain of transfers one per hop, + bounded by the client's resolve depth. + +### Neutral / Operational + +- A node refusing a final state because its group proved another logs it at + info with both state ids; a possession check that finds the other side of a + fork logs at warn. Either is the owner equivocating. +- Pricing, payments, audits and storage commitments are unchanged: a final + state is a state, paid, stored, committed and audited like any other. + +## Validation + +- Protocol: a final state is replaced by no counter and no other final state, + whatever its target; below the final counter the order is unchanged; + `replaces` is a strict partial order over a set including final states, total + except between two of them; a fold keeps the first final state it meets; + `transfer_to` signs at the final counter to the recipient, refuses to sign past + a final record, and only a final pointer target counts as a transfer. +- Node, store: a stored transfer is not displaced by a final state whose target + sorts first, nor by any lower counter. +- Node, properties: with one final state among the records every delivery order + converges on it, exhaustively over all permutations; with two, the first + delivered is kept. +- Node, request handler: a final state the group proves already superseded is + refused and named, and nothing is written; one nobody contradicts is taken; + the group is asked only about a final state the node lacks; two final states + raced to two nodes leave each on its first; a node restores its own lost + final state without asking, and still refuses the rival; a busy look neither + takes nor refuses a final state for good; a proven loser is refused again + before its signature is checked, and without asking. +- Node, the look: a peer that claims a rival and stalls its fetch does not hide + another peer's proof; a claim the served record does not back is no proof; + proven final states refuse others, not themselves, and the oldest is + forgotten first past the cap; a peer that serves a different rival than it + claimed proves nothing; a second proof at an address never replaces + the first; replays queued behind a look reuse its answer, and a proof + answers every replay of the loser; a clear answer is dropped when the write + it cleared fails; looks for addresses sharing their leading bits run side by + side. +- Node, repair: a node holding a final state adopts nothing else; of two final + states with quorum, the larger side is adopted in either answer order, and + a tie adopts neither in either order, nor does a side that silent peers + could still tie. +- Node, live network: a transfer written to one node reaches the group and no + node takes a second final state; a node that missed the transfer refuses a + different one because a peer serves the one it holds, and still takes the + group's own; the possession check does not penalise the other side of a fork + and still penalises a member holding nothing. +- Client: reads over a group holding a transfer, a majority fork, a no-majority + fork and a transfer still spreading; the finality check for each; transfer + preflight refusals; end to end against a local testnet with real settlement, + a transfer is final, readers are redirected to the recipient's pointer, the + recipient moves it, and the former owner's attempt to take it back is refused + by every node. + +## 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 4d8dd0f9..b58e9a32 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -37,5 +37,6 @@ See [`TOOLING.md`](./TOOLING.md) for `adrs`, `adr-kit`, and AI harness setup. - [ADR-0015: Direct browser clients over WebRTC Direct](./ADR-0015-direct-browser-clients-over-webrtc-direct.md) - [ADR-0016: Pointers — paid mutable references with an immutable owner](./ADR-0016-pointers-immutable-owner.md) - [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-0022: Remove the LMDB Chunk Store](./ADR-0022-remove-the-lmdb-chunk-store.md) diff --git a/src/node.rs b/src/node.rs index 1598ada1..ac1f3588 100644 --- a/src/node.rs +++ b/src/node.rs @@ -270,11 +270,10 @@ impl NodeBuilder { }; // ADR-0016: pointers replicate through the same engine. The PUT handler - // hands each newly stored paid state to it on this channel. + // hands each newly stored paid state to it, and asks it before taking + // a final state (ADR-0018). if let Some(service) = protocol.pointer_service() { - let (writes, fresh_writes) = tokio::sync::mpsc::unbounded_channel(); - service.attach_fresh_writes(writes); - engine.with_pointers(service.store().clone(), fresh_writes); + engine.with_pointer_service(service); } // ADR-0004: wire the engine's commitment state as the quote generator's diff --git a/src/pointer/mod.rs b/src/pointer/mod.rs index c4787f39..b859231d 100644 --- a/src/pointer/mod.rs +++ b/src/pointer/mod.rs @@ -3,10 +3,18 @@ //! Implements `docs/adr/ADR-0016-pointers-immutable-owner.md`. //! //! A pointer is a mutable, owner-signed reference stored at an address derived -//! from the owner's public key. Ownership is fixed at creation: there is no -//! transfer, no lineage, no certificates and no key rotation. That choice is -//! what lets the design be this small — the owner key is inlined in the -//! record, so validating a pointer needs nothing but the pointer. +//! from the owner's public key. The owner key is fixed at creation: there is no +//! lineage, no certificates and no key rotation. That choice is what lets the +//! design be this small — the owner key is inlined in the record, so +//! validating a pointer needs nothing but the pointer. +//! +//! What the address resolves to can still be handed over for good +//! (`docs/adr/ADR-0018-pointer-transfer-by-final-redirection.md`): a state at +//! the final counter is replaced by nothing, so once a node holds one pointing +//! at the new owner's pointer, the former owner cannot take it back there. The +//! store gets that from the merge rule; the service adds one look at the close +//! group before taking a final state, so a node that had not heard of the +//! first refuses a second one when a peer proves the first in time. //! //! # What lives here //! @@ -53,8 +61,8 @@ pub mod store; pub use ant_protocol::pointer::{ pointer_address, state_id_for_body, ParsedPointer, Pointer, PointerError, PointerState, - PointerTarget, PointerTargetKind, DATA_TYPE_POINTER, POINTER_BODY_LEN, POINTER_FORMAT_VERSION, - POINTER_WIRE_LEN, TARGET_WIRE_LEN, + PointerTarget, PointerTargetKind, DATA_TYPE_POINTER, FINAL_COUNTER, POINTER_BODY_LEN, + POINTER_FORMAT_VERSION, POINTER_WIRE_LEN, TARGET_WIRE_LEN, }; -pub use service::PointerService; +pub use service::{FinalStateWitness, FinalityCheck, PointerService}; pub use store::{PointerStore, PutOutcome}; diff --git a/src/pointer/service.rs b/src/pointer/service.rs index 9c5b0749..7d4c866f 100644 --- a/src/pointer/service.rs +++ b/src/pointer/service.rs @@ -16,11 +16,22 @@ //! are both 32 bytes from the same range; the domain separator makes a //! collision infeasible, not impossible. A node that holds one kind at an //! address refuses the other rather than silently choosing. +//! 4. **A look before a final state.** A state at the final counter is +//! replaced by nothing, so a node that takes one can never be corrected. +//! Before taking one it neither holds nor lost, the node asks its close +//! group whether a *different* final state is already held there, and +//! refuses if a peer proves one with the signed record (ADR-0018). That is +//! what keeps a former owner from handing an address over a second time to +//! a node that had not heard of the first, whenever a peer can prove the +//! first in time. A proven loser is remembered and refused again before its +//! signature is checked. //! //! # Order of work //! //! ```text -//! parse → compare with what is held → verify signature → check payment → commit +//! parse → compare with what is held → (final state only) a proven loser? +//! → verify signature → check payment +//! → (final state only) ask the close group → commit //! ``` //! //! The comparison precedes the signature check, so re-submitting a state the @@ -35,6 +46,7 @@ use ant_protocol::chunk::{ XorName, }; use bytes::Bytes; +use futures::future::BoxFuture; use parking_lot::RwLock; use saorsa_core::P2PNode; use tokio::sync::mpsc; @@ -46,7 +58,46 @@ use crate::pointer::store::{Inspected, PointerStore, PutOutcome}; use crate::replication::admission; use crate::replication::pointer::PointerFreshWrite; use crate::storage::{ChunkStore, SELF_CLOSENESS_GATE_WIDTH}; -use ant_protocol::pointer::POINTER_WIRE_LEN; +use ant_protocol::pointer::{PointerState, POINTER_WIRE_LEN}; + +/// Where a node looks, before it takes a final state, for a different final +/// state its close group already holds (ADR-0018). +/// +/// A trait rather than the replication engine itself, because the engine is +/// built after this service and needs a running P2P node, and because what the +/// service decides from the answer is worth testing without one. +pub trait FinalStateWitness: Send + Sync { + /// Ask the close group whether a final state other than `state` is + /// already held at `state.address`. + fn check_final<'a>(&'a self, state: &'a PointerState) -> BoxFuture<'a, FinalityCheck>; + + /// A final state at `state.address`, other than `state`, that an earlier + /// check already proved, without asking anyone. + /// + /// Cheap enough to consult ahead of the signature check, so a final state + /// that lost once is refused again for nothing, however often the same + /// paid record is replayed. + fn proven_conflict(&self, state: &PointerState) -> Option; + + /// Forget that a look found `state` clear: the write it cleared did not + /// land, so a retry must look again rather than reuse the answer. + fn forget_clear(&self, state: &PointerState); +} + +/// What a close group said about a final state a node is about to take. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FinalityCheck { + /// No peer proved a different final state. Also the answer when the group + /// could not be asked in time: silence is not evidence, so it never + /// refuses a write. + Clear, + /// A different final state a peer holds, proven by the signed record, + /// which only the owner could have made. + Conflict(PointerState), + /// Too many checks were running to start this one in time. The state is + /// neither taken nor refused for good, and the sender may try again. + Busy, +} /// Handles pointer requests against a [`PointerStore`]. #[derive(Clone)] @@ -70,6 +121,10 @@ pub struct PointerService { /// handler before it has a running P2P node. `None` in unit tests that /// never attach one, exactly as the chunk path does. p2p_node: Arc>>>, + /// Asked before a final state is taken (ADR-0018). Attached with the + /// replication engine; `None` where nothing replicates, and then a final + /// state is taken on the merge rule alone. + final_witness: Arc>>>, } impl std::fmt::Debug for PointerService { @@ -92,6 +147,7 @@ impl PointerService { payments: None, fresh_writes: Arc::new(RwLock::new(None)), p2p_node: Arc::new(RwLock::new(None)), + final_witness: Arc::new(RwLock::new(None)), } } @@ -125,6 +181,11 @@ impl PointerService { *self.fresh_writes.write() = Some(writes); } + /// Ask `witness` before taking a final state this node does not hold. + pub fn attach_final_state_witness(&self, witness: Arc) { + *self.final_witness.write() = Some(witness); + } + /// The store this service fronts. #[must_use] pub const fn store(&self) -> &PointerStore { @@ -162,10 +223,11 @@ impl PointerService { } }; - let address = parsed.state().address; - let state_id = parsed.state().state_id; + let state = *parsed.state(); + let address = state.address; + let state_id = state.state_id; - if let Some(refusal) = self.admit(parsed.state()).await { + if let Some(refusal) = self.admit(&state).await { return refusal; } @@ -191,6 +253,13 @@ impl PointerService { } } + // Last, because it costs the close group a round trip, and only for a + // paid final state: if a peer proves a different final state is + // already held, this one lost the race to it everywhere that matters. + if let Some(refusal) = self.final_refusal(&state).await { + return refusal; + } + // Charge the bytes this write will take, and hold the charge until it // lands. Checking capacity and then writing is the race the file store // exists to close: concurrent writers all pass one cached measurement @@ -202,6 +271,7 @@ impl PointerService { Ok(reservation) => Some(reservation), Err(e) => { debug!("Rejecting pointer PUT for {}: {e}", hex::encode(address)); + self.forget_clear(&state); return PointerPutResponse::Error(ProtocolError::StorageFailed(e.to_string())); } }, @@ -211,7 +281,7 @@ impl PointerService { // release it the moment this future is dropped, while the blocking // transaction it started runs on and publishes the file. let record_bytes = request.record.to_vec(); - match self.store.commit(record, reservation).await { + let response = match self.store.commit(record, reservation).await { Ok(PutOutcome::Changed) => { // Offer it to the rest of the close group, proof included, so // every member holds it whichever of them the client reached. @@ -243,6 +313,24 @@ impl PointerService { warn!("Pointer commit failed for {}: {e}", hex::encode(address)); PointerPutResponse::Error(ProtocolError::StorageFailed(e.to_string())) } + }; + // A look that cleared a final state is reused briefly for replays of + // it; a write that did not land must not leave that answer behind. + if !matches!( + response, + PointerPutResponse::Success { .. } | PointerPutResponse::Unchanged { .. } + ) { + self.forget_clear(&state); + } + response + } + + /// Forget a clear look for `state`, if a witness took one. + fn forget_clear(&self, state: &PointerState) { + if state.is_terminal() { + if let Some(witness) = self.final_witness.read().as_ref() { + witness.forget_clear(state); + } } } @@ -251,10 +339,7 @@ impl PointerService { /// `Some(response)` means refuse. Ordered cheapest first, and all of it /// ahead of the signature check, so a forged record for an address this /// node does not serve buys no cryptography. - async fn admit( - &self, - state: &ant_protocol::pointer::PointerState, - ) -> Option { + async fn admit(&self, state: &PointerState) -> Option { let address = state.address; // Any state the merge rule prefers to what this node knows, whatever @@ -269,9 +354,15 @@ impl PointerService { hex::encode(address), state.counter ); + // Named as the state this node knows, even one whose file it + // lost, so the sender can see what it must catch up to. return Some(PointerPutResponse::Stale { address, - state_id: self.store.state_id(&address).unwrap_or_default(), + state_id: self + .store + .remembered(&address) + .map(|known| known.state_id) + .unwrap_or_default(), }); } @@ -316,9 +407,81 @@ impl PointerService { ))); } } + + // A final state an earlier check proved lost to another is refused + // again here, before it buys a signature check or a round trip. + if self.needs_final_check(state) { + let proven = self + .final_witness + .read() + .as_ref() + .and_then(|witness| witness.proven_conflict(state)); + if let Some(conflict) = proven { + return Some(PointerPutResponse::Stale { + address, + state_id: conflict.state_id, + }); + } + } None } + /// Whether `state` needs the close group's word before this node takes + /// it: it is final, and this node neither holds nor remembers a final + /// state at its address. + /// + /// Nothing else is asked about. A lower counter can be replaced, so taking + /// one blind costs nothing a later state cannot fix. A node that holds a + /// final state has had its answer from the merge rule before this runs: + /// the same state is unchanged and any other is stale. And a node that + /// lost the file of a final state it held is admitted only that state + /// again ([`PointerStore::admits`]), which restores what it already took + /// rather than taking a new one. Asking there would let any peer on the + /// other side of a fork keep it from restoring its own copy. + fn needs_final_check(&self, state: &PointerState) -> bool { + state.is_terminal() + && !self + .store + .remembered(&state.address) + .is_some_and(|known| known.is_terminal()) + } + + /// The answer to a paid final state the close group proves already lost, + /// or that could not be checked in time; `None` to take it. + async fn final_refusal(&self, state: &PointerState) -> Option { + if !self.needs_final_check(state) { + return None; + } + let witness = self.final_witness.read().as_ref().map(Arc::clone)?; + match witness.check_final(state).await { + FinalityCheck::Clear => None, + // Named as the state the group proved, as for any stale arrival. + FinalityCheck::Conflict(conflict) => { + info!( + "Refusing final pointer state {} at {}: the close group already holds \ + final state {}, so the owner has finalized it before", + hex::encode(state.state_id), + hex::encode(state.address), + hex::encode(conflict.state_id) + ); + Some(PointerPutResponse::Stale { + address: state.address, + state_id: conflict.state_id, + }) + } + FinalityCheck::Busy => { + debug!( + "Deferring final pointer state {} at {}: too many finality checks running", + hex::encode(state.state_id), + hex::encode(state.address) + ); + Some(PointerPutResponse::Error(ProtocolError::Internal( + "too many finality checks running; try again".to_string(), + ))) + } + } + } + /// Handle a pointer GET. pub async fn handle_get(&self, request: PointerGetRequest) -> PointerGetResponse { match self.store.get(&request.address).await { @@ -399,8 +562,9 @@ fn cross_kind_refusal(address: XorName, chunk_present: Result) -> Option

(MlDsaPublicKey, MlDsaSecretKey) { ml_dsa_65().generate_keypair_from_seed(&[seed; 32]) @@ -777,6 +941,372 @@ mod tests { assert_eq!(held.counter(), u64::MAX); } + /// A close group whose answer to the finality question is fixed, and + /// which counts how often it was asked. + struct StubWitness { + answer: FinalityCheck, + proven: Option, + asked: AtomicUsize, + forgotten: AtomicUsize, + } + + impl StubWitness { + fn answering(answer: FinalityCheck) -> Arc { + Arc::new(Self { + answer, + proven: None, + asked: AtomicUsize::new(0), + forgotten: AtomicUsize::new(0), + }) + } + + /// One that already proved `conflict` in an earlier check. + fn having_proven(conflict: PointerState) -> Arc { + Arc::new(Self { + answer: FinalityCheck::Clear, + proven: Some(conflict), + asked: AtomicUsize::new(0), + forgotten: AtomicUsize::new(0), + }) + } + + fn asked(&self) -> usize { + self.asked.load(Ordering::SeqCst) + } + } + + impl FinalStateWitness for StubWitness { + fn check_final<'a>(&'a self, _state: &'a PointerState) -> BoxFuture<'a, FinalityCheck> { + self.asked.fetch_add(1, Ordering::SeqCst); + let answer = self.answer; + Box::pin(async move { answer }) + } + + fn proven_conflict(&self, state: &PointerState) -> Option { + self.proven + .filter(|proven| proven.state_id != state.state_id) + } + + fn forget_clear(&self, _state: &PointerState) { + self.forgotten.fetch_add(1, Ordering::SeqCst); + } + } + + fn final_state(seed: u8, target_byte: u8) -> Pointer { + let (pk, sk) = keypair(seed); + let target = PointerTarget::new(PointerTargetKind::Pointer, [target_byte; 32]); + Pointer::sign(&sk, &pk, FINAL_COUNTER, target).expect("sign") + } + + #[tokio::test] + async fn a_final_state_the_group_proves_is_already_superseded_is_refused() { + // The node has the pointer's ordinary state and has never seen a final + // one, so the merge rule alone would take this. The group proves the + // owner already finalized it elsewhere, so it is refused, named as the + // state that got there first, and nothing is written. + let (service, _dir) = service().await; + let current = signed(1, 3, 1); + service.handle_put(put(¤t)).await; + + let established = final_state(1, 0xAA); + let late = final_state(1, 0x01); + let witness = StubWitness::answering(FinalityCheck::Conflict(established.state())); + service.attach_final_state_witness(witness.clone()); + + match service.handle_put(put(&late)).await { + PointerPutResponse::Stale { address, state_id } => { + assert_eq!(address, late.address()); + assert_eq!( + state_id, + established.state_id(), + "the first final state is named" + ); + } + other => panic!("a second final state must be refused, got {other:?}"), + } + assert_eq!(witness.asked(), 1); + assert_eq!( + service.store().state_id(&late.address()), + Some(current.state_id()), + "nothing was written" + ); + } + + #[tokio::test] + async fn a_final_state_nobody_contradicts_is_taken() { + // Silence is not evidence: a group that proves nothing lets the final + // state through on the merge rule. + let (service, _dir) = service().await; + let witness = StubWitness::answering(FinalityCheck::Clear); + service.attach_final_state_witness(witness.clone()); + + let transfer = final_state(2, 0x42); + assert!(matches!( + service.handle_put(put(&transfer)).await, + PointerPutResponse::Success { .. } + )); + assert_eq!(witness.asked(), 1); + assert_eq!( + service.store().state_id(&transfer.address()), + Some(transfer.state_id()) + ); + } + + #[tokio::test] + async fn the_group_is_asked_only_about_a_final_state_the_node_lacks() { + // A lower counter can always be replaced later, so it is taken without + // a round trip. A node already holding a final state has its answer + // from the merge rule before any round trip: the same state is + // unchanged, and any other is stale. + let (service, _dir) = service().await; + let witness = StubWitness::answering(FinalityCheck::Clear); + service.attach_final_state_witness(witness.clone()); + + for counter in [0u64, 1, u64::MAX - 1] { + assert!(matches!( + service.handle_put(put(&signed(3, counter, 1))).await, + PointerPutResponse::Success { .. } + )); + } + assert_eq!(witness.asked(), 0, "no final state yet, no question"); + + let first = final_state(3, 0x10); + assert!(matches!( + service.handle_put(put(&first)).await, + PointerPutResponse::Success { .. } + )); + assert_eq!(witness.asked(), 1); + + let second = final_state(3, 0x01); + match service.handle_put(put(&second)).await { + PointerPutResponse::Stale { state_id, .. } => { + assert_eq!(state_id, first.state_id()); + } + other => panic!("a second final state must be stale, got {other:?}"), + } + assert!(matches!( + service.handle_put(put(&final_state(3, 0x10))).await, + PointerPutResponse::Unchanged { .. } + )); + assert_eq!(witness.asked(), 1, "a held final state answers for itself"); + } + + #[tokio::test] + async fn a_node_restores_its_own_lost_final_state_without_asking() { + // The node took a final state, then lost the file. The group holds + // the other side of a fork, which a check would prove. Asking would + // keep the node from restoring what it already took, for good; the + // lost state itself is the one thing it is admitted, so nobody is + // asked, and the rival is still refused. + let (service, _dir) = service().await; + let own = final_state(5, 0x10); + let rival = final_state(5, 0x01); + assert!(matches!( + service.handle_put(put(&own)).await, + PointerPutResponse::Success { .. } + )); + std::fs::remove_file(service.store().file_for(&own.address())).expect("remove"); + assert!(service + .store() + .get(&own.address()) + .await + .expect("get") + .is_none()); + + let witness = StubWitness::answering(FinalityCheck::Conflict(rival.state())); + service.attach_final_state_witness(witness.clone()); + assert!(matches!( + service.handle_put(put(&own)).await, + PointerPutResponse::Success { .. } + )); + assert_eq!(witness.asked(), 0, "restoring its own state asks nobody"); + assert_eq!( + service.store().state_id(&own.address()), + Some(own.state_id()) + ); + std::fs::remove_file(service.store().file_for(&own.address())).expect("remove"); + assert!(service + .store() + .get(&own.address()) + .await + .expect("get") + .is_none()); + match service.handle_put(put(&rival)).await { + PointerPutResponse::Stale { state_id, .. } => { + assert_eq!( + state_id, + own.state_id(), + "the rival is told which state won" + ); + } + other => panic!("the rival must be stale, got {other:?}"), + } + } + + /// A group that answers clear while, meanwhile, a rival final state + /// lands on this node: the race a clear look can lose. + struct RacingWitness { + store: PointerStore, + rival: Pointer, + forgotten: AtomicUsize, + } + + impl FinalStateWitness for RacingWitness { + fn check_final<'a>(&'a self, _state: &'a PointerState) -> BoxFuture<'a, FinalityCheck> { + Box::pin(async move { + self.store + .put_bytes(&self.rival.to_bytes()) + .await + .expect("the rival lands"); + FinalityCheck::Clear + }) + } + + fn proven_conflict(&self, _state: &PointerState) -> Option { + None + } + + fn forget_clear(&self, _state: &PointerState) { + self.forgotten.fetch_add(1, Ordering::SeqCst); + } + } + + #[tokio::test] + async fn a_cleared_final_state_whose_write_fails_is_looked_for_again() { + // The look found nothing, but a rival final state landed while it + // looked, so the write it cleared comes back stale. The clear answer + // must not outlive that write: a retry has to look again. + let (landing, _other) = service().await; + let (service, _dir) = service().await; + let witness = Arc::new(RacingWitness { + store: service.store().clone(), + rival: final_state(8, 0x01), + forgotten: AtomicUsize::new(0), + }); + service.attach_final_state_witness(witness.clone()); + assert!(matches!( + service.handle_put(put(&final_state(8, 0x42))).await, + PointerPutResponse::Stale { .. } + )); + assert_eq!(witness.forgotten.load(Ordering::SeqCst), 1); + + // A write that lands keeps its answer. + let witness = StubWitness::answering(FinalityCheck::Clear); + landing.attach_final_state_witness(witness.clone()); + assert!(matches!( + landing.handle_put(put(&final_state(9, 0x42))).await, + PointerPutResponse::Success { .. } + )); + assert_eq!(witness.forgotten.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn a_busy_check_neither_takes_nor_refuses_a_final_state_for_good() { + let (service, _dir) = service().await; + let current = signed(6, 3, 1); + service.handle_put(put(¤t)).await; + let witness = StubWitness::answering(FinalityCheck::Busy); + service.attach_final_state_witness(witness.clone()); + + let transfer = final_state(6, 0x42); + assert!(matches!( + service.handle_put(put(&transfer)).await, + PointerPutResponse::Error(ProtocolError::Internal(_)) + )); + assert_eq!(witness.asked(), 1); + assert_eq!( + service.store().state_id(&transfer.address()), + Some(current.state_id()), + "nothing was written" + ); + assert!( + service.store().admits(&transfer.state()), + "and the state can still be taken later" + ); + } + + #[tokio::test] + async fn a_proven_conflict_is_refused_again_before_any_signature_check() { + // Replaying a paid final state that already lost costs a node + // nothing: the refusal comes from what an earlier check proved, ahead + // of the signature check and without asking the group. A record + // whose signature does not even verify shows the order. + let (unwitnessed, _other) = service().await; + let (service, _dir) = service().await; + let established = final_state(7, 0xAA); + let late = final_state(7, 0x01); + let witness = StubWitness::having_proven(established.state()); + service.attach_final_state_witness(witness.clone()); + + let mut forged = late.to_bytes(); + if let Some(last) = forged.last_mut() { + *last ^= 0xFF; + } + match service + .handle_put(PointerPutRequest::new(Bytes::from(forged.clone()))) + .await + { + PointerPutResponse::Stale { state_id, .. } => { + assert_eq!(state_id, established.state_id()); + } + other => panic!("a proven loser must be refused as stale, got {other:?}"), + } + assert_eq!(witness.asked(), 0, "and nobody is asked again"); + assert!( + matches!( + unwitnessed + .handle_put(PointerPutRequest::new(Bytes::from(forged.clone()))) + .await, + PointerPutResponse::Error(_) + ), + "the record really does fail its signature check" + ); + + // The proven state itself is not refused by its own proof. + assert!(matches!( + service.handle_put(put(&established)).await, + PointerPutResponse::Success { .. } + )); + } + + #[tokio::test] + async fn two_final_states_raced_to_two_nodes_leave_each_on_its_first() { + // The one fork ADR-0018 allows: the owner signs two final states and + // sends one to each node before either has heard of the other. Each + // keeps what it took first and refuses the other, whichever sorts + // first. Nothing on a node can settle it; a reader decides by how many + // of the group hold each side. + let (first_node, _a) = service().await; + let (second_node, _b) = service().await; + let one = final_state(4, 0x09); + let other = final_state(4, 0x01); + + assert!(matches!( + first_node.handle_put(put(&one)).await, + PointerPutResponse::Success { .. } + )); + assert!(matches!( + second_node.handle_put(put(&other)).await, + PointerPutResponse::Success { .. } + )); + assert!(matches!( + first_node.handle_put(put(&other)).await, + PointerPutResponse::Stale { .. } + )); + assert!(matches!( + second_node.handle_put(put(&one)).await, + PointerPutResponse::Stale { .. } + )); + assert_eq!( + first_node.store().state_id(&one.address()), + Some(one.state_id()) + ); + assert_eq!( + second_node.store().state_id(&other.address()), + Some(other.state_id()) + ); + } + #[test] fn a_pointer_is_refused_where_a_chunk_already_sits() { // The two kinds share one 32-byte address space. A collision is diff --git a/src/pointer/store.rs b/src/pointer/store.rs index edc645a1..e10ecf7d 100644 --- a/src/pointer/store.rs +++ b/src/pointer/store.rs @@ -608,6 +608,16 @@ impl PointerStore { .map(|entry| entry.state) } + /// The state this node holds at `address`, or held until it lost the + /// file. + /// + /// Unlike [`Self::state`], a lost record still answers: it is the one + /// state, besides a newer one, that [`Self::admits`] lets restore it. + #[must_use] + pub fn remembered(&self, address: &XorName) -> Option { + self.snapshot(address).map(|entry| entry.state) + } + /// Whether a paid PUT of `state` may be taken. /// /// Any state the merge rule prefers to what is held, whatever its counter. diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 029166c3..62937413 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -68,6 +68,7 @@ use crate::payment::{ MIN_PAYMENT_PROOF_SIZE_BYTES, }; use crate::pointer::store::PointerStore; +use crate::pointer::{FinalStateWitness, PointerService}; use crate::replication::audit::AuditTickResult; use crate::replication::audit_coordinator::AuditChallengeCoordinator; use crate::replication::audit_metrics::{ @@ -2234,6 +2235,10 @@ impl ReplicationEngine { /// Replicate pointers too (ADR-0016): the records in `store`, and the /// fresh writes the pointer PUT handler sends on `fresh_writes`. /// + /// The PUT handler is not wired to ask this engine anything before it + /// takes a final state (ADR-0018); [`Self::with_pointer_service`] wires + /// both. + /// /// Call before [`Self::start`]. pub fn with_pointers( &mut self, @@ -2254,6 +2259,22 @@ impl ReplicationEngine { self.pointer_fresh_rx = Some(fresh_writes); } + /// Replicate the pointers `service` stores (ADR-0016), and wire the + /// service to this engine both ways: it hands each newly stored paid + /// state here to be offered on, and asks here before it takes a final + /// state (ADR-0018). + /// + /// Call before [`Self::start`]. + pub fn with_pointer_service(&mut self, service: &PointerService) { + let (writes, fresh_writes) = mpsc::unbounded_channel(); + service.attach_fresh_writes(writes); + self.with_pointers(service.store().clone(), fresh_writes); + if let Some(replication) = &self.pointers { + let witness: Arc = replication.clone(); + service.attach_final_state_witness(witness); + } + } + /// The pointer replication, when enabled. Tests use it to drive rounds. #[cfg(any(test, feature = "test-utils"))] #[must_use] diff --git a/src/replication/pointer.rs b/src/replication/pointer.rs index c8aa3c83..7afa5565 100644 --- a/src/replication/pointer.rs +++ b/src/replication/pointer.rs @@ -24,7 +24,13 @@ //! they hold that state or a newer one by returning a valid record. //! - **Possession.** Some minutes after offering a fresh state, the node asks //! each close-group member for the record. A member that is still responsible -//! and cannot produce that state or a newer one is penalised. +//! and cannot produce that state or a newer one is penalised — unless it holds +//! a *different final* state, which is a fork its owner made and not a failure +//! to store. +//! - **Finality.** Before taking a final state it does not hold, from a client +//! or a fresh offer, a node asks the close group whether a different final +//! state is already held, and refuses if a peer proves one with the signed +//! record (ADR-0018). See [`PointerReplication::check_final`]. //! //! Requests go only to peers that have sent a pointer message themselves (see //! [`PointerReplication::is_capable`]). A peer built before pointers cannot @@ -33,11 +39,13 @@ //! neighbour-sync round pushes hints, empty or not, so capability is learned //! within a cycle. -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; +use std::future::Future; use std::sync::Arc; use std::time::{Duration, Instant}; use ant_protocol::pointer::{Pointer, PointerState, POINTER_WIRE_LEN}; +use futures::future::BoxFuture; use futures::stream::{self, StreamExt}; use parking_lot::Mutex; use rand::Rng; @@ -45,6 +53,7 @@ use saorsa_core::identity::PeerId; use saorsa_core::{P2PNode, TrustEvent}; use tokio::sync::{mpsc, OwnedSemaphorePermit, RwLock, Semaphore}; use tokio::task::{spawn_blocking, JoinHandle}; +use tokio::time::Instant as TokioInstant; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; @@ -54,7 +63,8 @@ use crate::payment::{ PaymentVerifier, VerificationContext, MAX_PAYMENT_PROOF_SIZE_BYTES, MIN_PAYMENT_PROOF_SIZE_BYTES, }; -use crate::pointer::store::{Inspected, PointerStore}; +use crate::pointer::store::{Inspected, PointerStore, PutOutcome}; +use crate::pointer::{FinalStateWitness, FinalityCheck}; use crate::replication::admission; use crate::replication::commitment_state::ResponderCommitmentState; use crate::replication::config::{ @@ -145,9 +155,90 @@ const UNDECIDED_BACKOFF: Duration = Duration::from_secs(30); /// Most prune candidates examined per pass. const MAX_PRUNE_CANDIDATES_PER_PASS: usize = 256; +/// How long a node spends asking its close group for a conflicting final +/// state before taking one (ADR-0018). +/// +/// It runs inside a client's PUT, after payment has been verified, so it adds +/// to whatever that verification took; it is kept short so a PUT whose +/// payment is quick stays inside the client's ten-second store timeout. A +/// group that has not answered by then is taken to hold nothing that +/// conflicts: silence is never a vote, here as anywhere else. +pub const FINAL_STATE_CHECK_BUDGET: Duration = Duration::from_secs(4); + +/// How long a finality check waits for its turn before it gives up and +/// answers [`FinalityCheck::Busy`]. +/// +/// With [`FINAL_STATE_CHECK_BUDGET`] after it, a check adds at most six +/// seconds to a PUT after its payment is verified. +pub const FINAL_STATE_CHECK_WAIT: Duration = Duration::from_secs(2); + +/// How many finality checks run at once. +/// +/// Each runs on its own share of the address space. Two checks for one +/// address never run together, so a burst of replays of one final state +/// costs the group one round of questions. +pub const FINAL_STATE_CHECK_STRIPES: usize = 64; + +/// How many addresses a node remembers proven final states for, oldest +/// forgotten first. +/// +/// A proof is a final state the close group was shown to hold, so it never +/// goes stale; the cap only bounds memory, at most two states of about 200 +/// bytes each per address. +const MAX_PROVEN_FINALS: usize = 16_384; + +/// Proven final states remembered per address. Two different ones are enough +/// to refuse every final state there: each conflicts with the other, and any +/// third with both. +const MAX_PROOFS_PER_ADDRESS: usize = 2; + +/// How long a look that found no conflict answers again for the same final +/// state without asking anyone. +/// +/// As long as a replay queued behind it can have waited for its turn, and no +/// longer, since a rival could reach the group after it. The answer is also +/// forgotten as soon as the write it cleared fails. +pub const FINAL_STATE_CLEAR_REUSE: Duration = FINAL_STATE_CHECK_WAIT; + +/// Most clear looks remembered for reuse. Each is at most +/// [`FINAL_STATE_CLEAR_REUSE`] old, so this only bounds a burst. +const MAX_CLEAR_LOOKS: usize = 4096; + /// One peer's answer about one address: the state it holds there, if any. type StateAnswer = ((PeerId, XorName), Option); +/// Whether `record` is exactly the state a quorum backed. +/// +/// The whole state, not only its identifier: the backed state is built from +/// peers' unauthenticated summaries, and a summary can pair a real state's +/// identifier with some other counter or target. +fn backs(record: &Pointer, wanted: &PointerState) -> bool { + record.state() == *wanted +} + +/// One peer's state response, read against the addresses it was asked about. +/// +/// An explicit "nothing held" is an answer. A summary about some other +/// address is no answer at all, and is left out, so that it does not count as +/// a peer that answered. +fn read_state_answers( + peer: PeerId, + addresses: &[XorName], + states: Vec>, +) -> Vec { + addresses + .iter() + .zip(states) + .filter_map(|(address, summary)| match summary { + None => Some(((peer, *address), None)), + Some(summary) if summary.address == *address => { + Some(((peer, *address), Some(PointerState::from(summary)))) + } + Some(_) => None, + }) + .collect() +} + /// A pointer state this node accepted from a paying client, to be offered to /// the rest of its close group. pub struct PointerFreshWrite { @@ -201,37 +292,63 @@ pub(crate) fn evaluate( if group_size == 0 || quorum_needed == 0 { return Verdict::Undecided; } + // Votes count together only for the same whole state. A summary is a + // peer's unauthenticated claim: one naming a real state's identifier with + // some other counter must not absorb, or reshape, the honest votes for it. let mut by_state: Vec<(PointerState, Vec)> = Vec::new(); for (peer, answer) in answers { let Some(state) = answer else { continue }; - match by_state - .iter_mut() - .find(|(known, _)| known.state_id == state.state_id) - { + match by_state.iter_mut().find(|(known, _)| known == state) { Some((_, holders)) => holders.push(*peer), None => by_state.push((*state, vec![*peer])), } } let beats_held = |state: &PointerState| held.is_none_or(|held| state.replaces(held)); + // The merge winner, and between two final states — which the merge rule + // leaves unordered — the one more of the group holds. Only a group wider + // than twice the quorum can back two at once, and then arrival order must + // not be what decides. + let prefer = |candidate: &(PointerState, Vec), + current: &(PointerState, Vec)| { + candidate.0.replaces(¤t.0) + || (!current.0.replaces(&candidate.0) && candidate.1.len() > current.1.len()) + }; let best = by_state .iter() .filter(|(state, holders)| holders.len() >= quorum_needed && beats_held(state)) .fold( None::<&(PointerState, Vec)>, |best, candidate| match best { - Some(current) if !candidate.0.replaces(¤t.0) => Some(current), + Some(current) if !prefer(candidate, current) => Some(current), _ => Some(candidate), }, ); + let unanswered = group_size.saturating_sub(answers.len()); if let Some((state, holders)) = best { + // A final state is adopted only as the strictly larger side, counting + // every peer that did not answer for its rival: a final state is + // never replaced, so two repairing nodes that each saw part of a tie + // would otherwise adopt opposite sides for good. That covers a rival + // seen among the answers and one only silent peers could hold. + if state.is_terminal() { + let rival_could_match = by_state.iter().any(|(other, others)| { + other.state_id != state.state_id + && beats_held(other) + && !other.replaces(state) + && !state.replaces(other) + && others.len().saturating_add(unanswered) >= holders.len() + }); + if rival_could_match || unanswered >= holders.len() { + return Verdict::Undecided; + } + } return Verdict::Adopt { state: *state, holders: holders.clone(), }; } - let unanswered = group_size.saturating_sub(answers.len()); let largest = by_state .iter() .filter(|(state, _)| beats_held(state)) @@ -291,6 +408,10 @@ enum Possession { Holds, /// The peer could not produce `fresh` or anything newer. Missing, + /// The peer served a different final state, and `fresh` is final too: the + /// owner signed both, and each node kept the one it took first. Carries + /// the state the peer holds, so the fork can be named. + Forked(XorName), /// Nothing to judge the peer on: it served a record that does not verify, /// which the fetch has already charged it for, or this node could not /// check what it served. @@ -304,6 +425,8 @@ fn judge_possession(fetched: Fetched, fresh: &PointerState) -> Possession { let state = record.state(); if state.state_id == fresh.state_id || state.replaces(fresh) { Possession::Holds + } else if state.is_terminal() && fresh.is_terminal() { + Possession::Forked(state.state_id) } else { Possession::Missing } @@ -361,6 +484,8 @@ pub struct PointerReplication { sync_answers: Arc, /// When each peer was last answered with hints. answered_syncs: Mutex, + /// The bounds on looks before a final state (ADR-0018). + finality: FinalityLooks, shutdown: CancellationToken, tracker: TaskTracker, } @@ -398,6 +523,158 @@ impl AnsweredSyncs { } } +/// Final states a finality check proved the close group holds, remembered so +/// that a paid final state which lost to one is refused again without asking +/// anyone. +#[derive(Default)] +struct ProvenFinals { + /// Up to [`MAX_PROOFS_PER_ADDRESS`] different proven states per address. + by_address: HashMap>, + /// Addresses in the order they were first proven, oldest first. + order: VecDeque, +} + +impl ProvenFinals { + /// A proven final state at `state.address` other than `state`. + fn conflict(&self, state: &PointerState) -> Option { + self.by_address + .get(&state.address)? + .iter() + .find(|proven| proven.state_id != state.state_id) + .copied() + } + + /// Remember `proven` beside what is already proven at its address, never + /// in place of it: a proof replaced would let the state it disproved in. + /// Forgets the oldest addresses past the cap. + fn remember(&mut self, proven: PointerState) { + if let Some(states) = self.by_address.get_mut(&proven.address) { + let known = states.iter().any(|state| state.state_id == proven.state_id); + if !known && states.len() < MAX_PROOFS_PER_ADDRESS { + states.push(proven); + } + } else { + self.by_address.insert(proven.address, vec![proven]); + self.order.push_back(proven.address); + } + while self.by_address.len() > MAX_PROVEN_FINALS { + let Some(oldest) = self.order.pop_front() else { + break; + }; + self.by_address.remove(&oldest); + } + } +} + +/// The bounds on looks before a final state (ADR-0018), apart from the +/// network so they can be tested without one. +/// +/// A look's answer is kept either way: a proof for good, a clear look briefly. +/// Looks for one address take turns, so replays queued behind a look find its +/// answer rather than asking again; at most [`FINAL_STATE_CHECK_STRIPES`] +/// run at once. +struct FinalityLooks { + proven: Mutex, + /// When each final state was last looked for and found clear, on the + /// runtime's clock, which the look's own time bounds use as well. + clear: Mutex>, + /// One turn per share of the address space. + turns: Vec>, +} + +impl FinalityLooks { + fn new() -> Self { + Self { + proven: Mutex::new(ProvenFinals::default()), + clear: Mutex::new(HashMap::new()), + turns: (0..FINAL_STATE_CHECK_STRIPES) + .map(|_| tokio::sync::Mutex::new(())) + .collect(), + } + } + + /// A proven final state that conflicts with `state`, asking nobody. + fn proven_conflict(&self, state: &PointerState) -> Option { + self.proven.lock().conflict(state) + } + + /// Whether a look found `state` clear recently enough to answer again. + fn recently_clear(&self, state: &PointerState, now: TokioInstant) -> bool { + self.clear + .lock() + .get(&(state.address, state.state_id)) + .is_some_and(|at| now.saturating_duration_since(*at) < FINAL_STATE_CLEAR_REUSE) + } + + /// Forget a clear look for `state`: the write it cleared did not land, so + /// a retry must look again. + fn forget_clear(&self, state: &PointerState) { + self.clear.lock().remove(&(state.address, state.state_id)); + } + + fn note_clear(&self, state: &PointerState, now: TokioInstant) { + let mut clear = self.clear.lock(); + if clear.len() >= MAX_CLEAR_LOOKS { + clear.retain(|_, at| now.saturating_duration_since(*at) < FINAL_STATE_CLEAR_REUSE); + } + if clear.len() < MAX_CLEAR_LOOKS { + clear.insert((state.address, state.state_id), now); + } + } + + /// Answer whether `state` may be taken, running `look` only when neither + /// a proof nor a recent clear look answers it already. + async fn check(&self, state: &PointerState, look: Look) -> FinalityCheck + where + Look: FnOnce() -> Fut + Send, + Fut: Future> + Send, + { + if !state.is_terminal() { + return FinalityCheck::Clear; + } + if let Some(conflict) = self.proven_conflict(state) { + return FinalityCheck::Conflict(conflict); + } + // By the last byte: the addresses one node is responsible for share + // their leading bits, so the first byte would put them all on one turn. + let stripe = usize::from(state.address.last().copied().unwrap_or_default()) + % FINAL_STATE_CHECK_STRIPES; + let Some(turn) = self.turns.get(stripe) else { + return FinalityCheck::Busy; + }; + let Ok(_turn) = tokio::time::timeout(FINAL_STATE_CHECK_WAIT, turn.lock()).await else { + return FinalityCheck::Busy; + }; + // A look that held the turn may have answered this meanwhile. + if let Some(conflict) = self.proven_conflict(state) { + return FinalityCheck::Conflict(conflict); + } + if self.recently_clear(state, TokioInstant::now()) { + return FinalityCheck::Clear; + } + match tokio::time::timeout(FINAL_STATE_CHECK_BUDGET, look()).await { + Ok(Some(record)) => { + let conflict = record.state(); + self.proven.lock().remember(conflict); + FinalityCheck::Conflict(conflict) + } + Ok(None) => { + self.note_clear(state, TokioInstant::now()); + FinalityCheck::Clear + } + // Silence proves nothing, and is not kept for reuse either: the + // next look may be answered. + Err(_) => { + debug!( + "Close group of pointer {} did not answer the finality check in time", + hex::encode(state.address) + ); + FinalityCheck::Clear + } + } + } +} + impl PointerReplication { /// Pointer replication over `store`, sharing the engine's resources. #[allow(clippy::too_many_arguments)] @@ -430,6 +707,7 @@ impl PointerReplication { out_of_range: Mutex::new(HashMap::new()), sync_answers: Arc::new(Semaphore::new(MAX_CONCURRENT_SYNC_ANSWERS)), answered_syncs: Mutex::new(AnsweredSyncs::default()), + finality: FinalityLooks::new(), shutdown, tracker, } @@ -950,17 +1228,7 @@ impl PointerReplication { let Some(ReplicationMessageBody::PointerStateResponse(response)) = body else { return Vec::new(); }; - addresses - .iter() - .zip(response.states) - .map(|(address, summary)| { - // An answer about a different address is no answer. - let state = summary - .filter(|summary| summary.address == *address) - .map(PointerState::from); - ((peer, *address), state) - }) - .collect() + read_state_answers(peer, &addresses, response.states) }) .buffer_unordered(VERIFICATION_CONCURRENCY) .collect() @@ -980,9 +1248,10 @@ impl PointerReplication { let Some(record) = self.fetch_record(holder, &address).await else { continue; }; - // The quorum backed this state. A holder that has moved on since - // serves a different one, which that quorum says nothing about. - if record.state_id() != wanted.state_id { + // The quorum backed this state, exactly. A holder that has moved + // on since serves a different one, which that quorum says nothing + // about. + if !backs(&record, &wanted) { continue; } match self.store_verified(record, None).await { @@ -1017,8 +1286,10 @@ impl PointerReplication { return Ok(false); } let reservation = self.chunks.reserve(POINTER_WIRE_LEN as u64)?; - self.store.commit(record, Some(reservation)).await?; - Ok(true) + // Held now, whether this wrote it or it was already here; a commit + // that found a state it loses to is not. + let outcome = self.store.commit(record, Some(reservation)).await?; + Ok(!matches!(outcome, PutOutcome::Stale)) } // ----------------------------------------------------------------------- @@ -1297,6 +1568,16 @@ impl PointerReplication { { return; } + // A final state already proven to have lost is dropped before it + // buys a signature check. + let needs_final_check = state.is_terminal() + && !self + .store + .remembered(&state.address) + .is_some_and(|known| known.is_terminal()); + if needs_final_check && self.finality.proven_conflict(&state).is_some() { + return; + } let record = match self.store.verify(parsed).await { Ok(record) => record, Err(e) => { @@ -1321,10 +1602,21 @@ impl PointerReplication { ); return; } - match self + // A final state this node neither holds nor remembers is looked for + // in the group first, exactly as a client PUT of one is: the peer + // offering it may simply be the side of a race this node has not + // heard the other side of. A node that lost its own final state is + // admitted only that state, so restoring it asks nobody. + if needs_final_check && !self.offered_final_is_clear(&source, &state).await { + return; + } + let stored = self .store_verified(record, Some(self.config.paid_list_close_group_size)) - .await - { + .await; + if !matches!(stored, Ok(true)) && needs_final_check { + self.finality.forget_clear(&state); + } + match stored { Ok(true) => debug!( "Stored fresh pointer {} from {source}", hex::encode(state.address) @@ -1337,6 +1629,116 @@ impl PointerReplication { } } + // ----------------------------------------------------------------------- + // Finality + // ----------------------------------------------------------------------- + + /// Whether a freshly offered final state may be taken: no peer proved a + /// different one, and the look was not too busy to run. + async fn offered_final_is_clear(&self, source: &PeerId, state: &PointerState) -> bool { + match self.check_final(state).await { + FinalityCheck::Clear => true, + FinalityCheck::Conflict(conflict) => { + info!( + "Refusing fresh final pointer state {} at {} from {source}: the close \ + group already holds final state {}", + hex::encode(state.state_id), + hex::encode(state.address), + hex::encode(conflict.state_id) + ); + false + } + FinalityCheck::Busy => { + debug!( + "Dropping fresh final pointer state {} at {} from {source}: too many \ + finality checks running", + hex::encode(state.state_id), + hex::encode(state.address) + ); + false + } + } + } + + /// Ask the close group whether a final state at `state.address`, other + /// than `state`, is already held, and have a peer that says so prove it + /// by serving the signed record. + /// + /// Asked before this node takes a final state it does not hold. A final + /// state is replaced by nothing, so a node that took a second one would + /// hold it for good; asking first is what keeps a former owner from + /// finalizing an address again on a node that had not yet heard it was + /// final, such as one that joined the group since. + /// + /// A peer's word is not enough to refuse: a state summary is a claim + /// anyone can make, and one dishonest peer could otherwise block every + /// handover. A signed record is not a claim. Only the owner can sign a + /// final state, so one that verifies is the owner's own proof that it + /// finalized the pointer before. + /// + /// A proof is remembered, beside any other for the address, so the same + /// loser is refused again without a question; a clear look is reused for + /// the same state for [`FINAL_STATE_CLEAR_REUSE`]. Checks for one address + /// wait their turn, so a burst of replays queued behind a look costs that + /// one round, and at most [`FINAL_STATE_CHECK_STRIPES`] run at once; one + /// that cannot start within [`FINAL_STATE_CHECK_WAIT`] answers `Busy`. + /// Once started it is bounded by [`FINAL_STATE_CHECK_BUDGET`], and only + /// peers that have sent a pointer message are asked. A group that cannot + /// be asked in time finds nothing, and the write goes ahead on the merge + /// rule alone: a race is a fork the client detects, not one this can + /// prevent. + pub async fn check_final(&self, state: &PointerState) -> FinalityCheck { + self.finality + .check(state, || self.find_conflicting_final(state)) + .await + } + + /// The body of [`Self::check_final`], without its bounds. + async fn find_conflicting_final(&self, state: &PointerState) -> Option { + let self_id = *self.p2p.peer_id(); + let address = state.address; + let peers: Vec = self + .p2p + .dht_manager() + .find_closest_nodes_local(&address, self.config.close_group_size) + .await + .into_iter() + .map(|node| node.peer_id) + .filter(|peer| *peer != self_id && self.is_capable(peer)) + .collect(); + first_proven_final( + peers, + state, + |peer| async move { self.ask_state(&peer, &address).await }, + |peer| async move { self.fetch_record(&peer, &address).await }, + ) + .await + } + + /// Ask one peer which state it holds at `address`. + async fn ask_state(&self, peer: &PeerId, address: &XorName) -> Option { + let body = self + .request( + peer, + ReplicationMessageBody::PointerStateRequest(PointerStateRequest { + addresses: vec![*address], + }), + FINAL_STATE_CHECK_BUDGET, + ) + .await?; + let ReplicationMessageBody::PointerStateResponse(response) = body else { + return None; + }; + // An answer about a different address is no answer. + response + .states + .into_iter() + .next() + .flatten() + .filter(|summary| summary.address == *address) + .map(PointerState::from) + } + // ----------------------------------------------------------------------- // Possession // ----------------------------------------------------------------------- @@ -1373,8 +1775,25 @@ impl PointerReplication { continue; } let fetched = self.fetch_outcome(peer, &fresh.address).await; - if judge_possession(fetched, &fresh) != Possession::Missing { - continue; + match judge_possession(fetched, &fresh) { + Possession::Missing => {} + // Two different final states: the owner signed both and each + // node kept the one it took first, as this one did. The peer + // is holding what the merge rule told it to hold, so it is not + // penalised for the owner's fork. Said loudly, because a read + // of this pointer now depends on which side most of the group + // is on. + Possession::Forked(held) => { + warn!( + "Pointer {} is forked: peer {peer} holds final state {}, this node \ + offered final state {}", + hex::encode(fresh.address), + hex::encode(held), + hex::encode(fresh.state_id) + ); + continue; + } + Possession::Holds | Possession::NotJudged => continue, } // Asking can wait on this peer's other requests and on the peers // before it, and it may have left the group meanwhile. It is @@ -1537,12 +1956,71 @@ impl PointerReplication { } } +impl FinalStateWitness for PointerReplication { + fn check_final<'a>(&'a self, state: &'a PointerState) -> BoxFuture<'a, FinalityCheck> { + Box::pin(Self::check_final(self, state)) + } + + fn proven_conflict(&self, state: &PointerState) -> Option { + self.finality.proven_conflict(state) + } + + fn forget_clear(&self, state: &PointerState) { + self.finality.forget_clear(state); + } +} + +/// The first final state other than `state` that one of `peers` claims with +/// `ask` and then proves with `fetch`. +/// +/// Each peer's question and fetch run as one pipeline, all of them at once, +/// and the first proof wins. A peer that claims a rival and then stalls its +/// fetch holds up nobody else's proof: were the fetches made one at a time +/// after the claims, that one peer would hold the check until its budget ran +/// out, and a finality check that runs out finds nothing. +async fn first_proven_final( + peers: Vec, + state: &PointerState, + ask: Ask, + fetch: Fetch, +) -> Option +where + Ask: Fn(PeerId) -> AskFut + Sync, + AskFut: Future> + Send, + Fetch: Fn(PeerId) -> FetchFut + Sync, + FetchFut: Future> + Send, +{ + let rival = |held: &PointerState| held.is_terminal() && held.state_id != state.state_id; + let (ask, fetch, rival) = (&ask, &fetch, &rival); + let width = peers.len().max(1); + let mut proofs = stream::iter(peers) + .map(|peer| async move { + let held = ask(peer).await?; + if !rival(&held) { + return None; + } + let record = fetch(peer).await?; + // The record must be the state the peer claimed: a claim the + // served record does not back is no proof, even if what it serves + // is some other rival. + (backs(&record, &held) && rival(&record.state())).then_some(record) + }) + .buffer_unordered(width); + while let Some(proof) = proofs.next().await { + if proof.is_some() { + return proof; + } + } + None +} + #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod tests { use super::*; - use ant_protocol::pointer::{PointerTarget, PointerTargetKind}; + use ant_protocol::pointer::{PointerTarget, PointerTargetKind, FINAL_COUNTER}; use saorsa_pqc::api::sig::ml_dsa_65; + use std::sync::atomic::{AtomicUsize, Ordering}; fn state(counter: u64, target: u8, id: u8) -> PointerState { PointerState { @@ -1818,8 +2296,394 @@ mod tests { } } + #[test] + fn a_node_holding_a_final_state_adopts_nothing_else() { + // Not even another final state the whole group holds: the merge rule + // replaces a final state with nothing, and repair is the merge rule + // applied to what the group says. + let held = state(u64::MAX, 9, 90); + let other = state(u64::MAX, 1, 91); + let got = evaluate( + Some(&held), + 7, + &answers(&[ + (1, Some(other)), + (2, Some(other)), + (3, Some(other)), + (4, Some(other)), + (5, Some(other)), + ]), + 4, + ); + assert_eq!(got, Verdict::Refused); + } + + #[test] + fn of_two_final_states_with_quorum_the_one_more_hold_is_adopted() { + // Two final states are unordered, so only a group wide enough to back + // both can get here. Arrival order must not pick; the larger side does. + let fewer = state(u64::MAX, 1, 70); + let more = state(u64::MAX, 9, 71); + for order in [[fewer, more], [more, fewer]] { + let mut list = Vec::new(); + let mut next = 1u8; + for candidate in order { + let holders = if candidate.state_id == more.state_id { + 5 + } else { + 4 + }; + for _ in 0..holders { + list.push((next, Some(candidate))); + next += 1; + } + } + // Everyone answered, so the four cannot become five. + match evaluate(None, 9, &answers(&list), 4) { + Verdict::Adopt { state, holders } => { + assert_eq!(state.state_id, more.state_id); + assert_eq!(holders.len(), 5); + } + other => panic!("expected the larger side, got {other:?}"), + } + } + } + + #[test] + fn a_silent_peer_that_could_tie_two_final_states_blocks_adoption() { + // Eight peers, quorum four: four hold one final state, three another, + // one is silent. The silent one could hold the second, making a tie, + // so neither side is adopted, whichever answered first. + let four = state(u64::MAX, 1, 90); + let three = state(u64::MAX, 9, 91); + for order in [[four, three], [three, four]] { + let mut list = Vec::new(); + let mut next = 1u8; + for candidate in order { + let count = if candidate.state_id == four.state_id { + 4 + } else { + 3 + }; + for _ in 0..count { + list.push((next, Some(candidate))); + next += 1; + } + } + assert_eq!(evaluate(None, 8, &answers(&list), 4), Verdict::Undecided); + } + + // A lone final state with quorum, and as many silent peers as hold + // it, could meet an unseen rival: undecided too. One fewer silent + // peer and it is adopted. + let alone = state(u64::MAX, 5, 92); + let held: Vec<(u8, Option)> = (1..=4).map(|p| (p, Some(alone))).collect(); + assert_eq!(evaluate(None, 8, &answers(&held), 4), Verdict::Undecided); + assert!(matches!( + evaluate(None, 7, &answers(&held), 4), + Verdict::Adopt { .. } + )); + } + + #[test] + fn two_final_states_backed_equally_are_adopted_by_neither_order() { + // A group of eight with a quorum of four can back two final states + // four to four. Whichever answered first must not decide. + let one = state(u64::MAX, 1, 80); + let other = state(u64::MAX, 9, 81); + for order in [[one, other], [other, one]] { + let mut list = Vec::new(); + let mut next = 1u8; + for candidate in order { + for _ in 0..4 { + list.push((next, Some(candidate))); + next += 1; + } + } + assert_eq!(evaluate(None, 8, &answers(&list), 4), Verdict::Undecided); + } + } + + #[test] + fn a_forged_summary_does_not_absorb_the_votes_for_a_final_state() { + // One peer names final state A's identifier with a lower counter. It + // must stand alone, not carry A's three honest votes as a non-final + // state of four and so slip past the guard for final states. + let a = state(u64::MAX, 1, 93); + let forged = PointerState { + counter: u64::MAX - 1, + ..a + }; + let b = state(u64::MAX, 9, 94); + let mut list = vec![(1u8, Some(forged))]; + list.extend((2..=4).map(|p| (p, Some(a)))); + list.extend((5..=7).map(|p| (p, Some(b)))); + assert!(!matches!( + evaluate(None, 8, &answers(&list), 4), + Verdict::Adopt { .. } + )); + } + + #[tokio::test] + async fn a_claim_that_names_a_rival_with_another_target_proves_nothing() { + // The claim reuses a real rival's identifier with some other target; + // the peer then serves the real rival. The record is not what was + // claimed, so it is no proof. + let taking = final_record(18, 0x01); + let rival = final_record(18, 0xAA); + let mut claim = rival.state(); + claim.target = PointerTarget::new(PointerTargetKind::Chunk, [0xCC; 32]); + let found = first_proven_final( + vec![peer(1)], + &taking.state(), + |_| async move { Some(claim) }, + |_| { + let record = rival.clone(); + async move { Some(record) } + }, + ) + .await; + assert!(found.is_none()); + } + + #[test] + fn a_record_backs_only_the_whole_state_a_quorum_named() { + let record = final_record(17, 0x01); + assert!(backs(&record, &record.state())); + let forged = PointerState { + counter: 7, + ..record.state() + }; + assert!( + !backs(&record, &forged), + "a summary pairing the record's identifier with another counter is not backed" + ); + } + + #[test] + fn a_summary_for_another_address_is_no_answer_at_all() { + let (a, b, c) = ([1u8; 32], [2u8; 32], [3u8; 32]); + let mut elsewhere = state(3, 1, 95); + elsewhere.address = [9u8; 32]; + let mut here = state(3, 1, 96); + here.address = c; + let read = read_state_answers( + peer(1), + &[a, b, c], + vec![None, Some(elsewhere.into()), Some(here.into())], + ); + assert_eq!(read, vec![((peer(1), a), None), ((peer(1), c), Some(here))]); + } + #[test] fn a_group_nobody_can_be_asked_in_is_undecided() { assert_eq!(evaluate(None, 0, &[], 0), Verdict::Undecided); } + + fn final_record(seed: u8, target: u8) -> Pointer { + let (pk, sk) = ml_dsa_65().generate_keypair_from_seed(&[seed; 32]); + let target = PointerTarget::new(PointerTargetKind::Pointer, [target; 32]); + Pointer::sign(&sk, &pk, FINAL_COUNTER, target).expect("sign") + } + + #[tokio::test(start_paused = true)] + async fn a_claimant_that_stalls_its_fetch_does_not_hide_a_rival_another_peer_proves() { + // One peer claims a rival final state first and then never serves + // it; an honest peer claims the same rival a moment later and serves + // it at once. The honest proof must land inside the budget. + let taking = final_record(9, 0x01); + let rival = final_record(9, 0xAA); + let (staller, honest) = (peer(1), peer(2)); + let found = tokio::time::timeout( + FINAL_STATE_CHECK_BUDGET, + first_proven_final( + vec![staller, honest], + &taking.state(), + |asked| { + let claim = rival.state(); + async move { + if asked == honest { + tokio::time::sleep(Duration::from_millis(100)).await; + } + Some(claim) + } + }, + |asked| { + let record = rival.clone(); + async move { + if asked == staller { + std::future::pending::<()>().await; + } + Some(record) + } + }, + ), + ) + .await; + assert_eq!( + found.ok().flatten().map(|record| record.state_id()), + Some(rival.state_id()), + "the honest peer's proof was hidden behind the stalled fetch" + ); + } + + #[tokio::test] + async fn only_a_different_final_state_that_is_served_is_a_proof() { + let taking = final_record(10, 0x01); + let rival = final_record(10, 0xAA); + let (same, unbacked, open) = (peer(1), peer(2), peer(3)); + let found = first_proven_final( + vec![same, unbacked, open], + &taking.state(), + |asked| { + let claim = if asked == same { + taking.state() + } else if asked == unbacked { + rival.state() + } else { + state(3, 1, 1) + }; + async move { Some(claim) } + }, + // The peer claiming a rival serves the state being taken instead. + |_| { + let record = taking.clone(); + async move { Some(record) } + }, + ) + .await; + assert!( + found.is_none(), + "a claim the record does not back is no proof" + ); + } + + #[tokio::test] + async fn a_peer_that_serves_a_different_rival_than_it_claimed_proves_nothing() { + let taking = final_record(16, 0x01); + let claimed = final_record(16, 0xAA); + let served = final_record(16, 0xBB); + let found = first_proven_final( + vec![peer(1)], + &taking.state(), + |_| { + let claim = claimed.state(); + async move { Some(claim) } + }, + |_| { + let record = served.clone(); + async move { Some(record) } + }, + ) + .await; + assert!(found.is_none()); + } + + #[test] + fn a_proven_final_state_refuses_others_and_forgets_the_oldest_past_the_cap() { + let mut proven = ProvenFinals::default(); + let held = final_record(11, 0xAA).state(); + let other = final_record(11, 0x01).state(); + proven.remember(held); + assert_eq!(proven.conflict(&other), Some(held)); + assert_eq!( + proven.conflict(&held), + None, + "a proof is no conflict with itself" + ); + + for i in 0..MAX_PROVEN_FINALS { + let mut filler = held; + filler.address = [0; 32]; + if let Some(slot) = filler.address.get_mut(..8) { + slot.copy_from_slice(&(i as u64 + 1).to_be_bytes()); + } + proven.remember(filler); + } + assert_eq!(proven.by_address.len(), MAX_PROVEN_FINALS); + assert_eq!(proven.order.len(), MAX_PROVEN_FINALS); + assert_eq!( + proven.conflict(&other), + None, + "the oldest proof is forgotten first" + ); + } + + #[test] + fn a_second_proof_never_replaces_the_first() { + // Both sides of a fork proven at one address: each is refused by the + // other, so a later look that proved the other side cannot let back + // in the state the first look disproved. + let mut proven = ProvenFinals::default(); + let a = final_record(12, 0xAA).state(); + let b = final_record(12, 0xBB).state(); + let c = final_record(12, 0xCC).state(); + proven.remember(a); + proven.remember(b); + proven.remember(c); + assert_eq!(proven.conflict(&b), Some(a)); + assert_eq!(proven.conflict(&a), Some(b)); + assert!(proven.conflict(&c).is_some(), "a third is refused by both"); + assert_eq!( + proven.by_address.get(&a.address).map(Vec::len), + Some(MAX_PROOFS_PER_ADDRESS) + ); + } + + #[tokio::test(start_paused = true)] + async fn looks_for_addresses_sharing_leading_bits_run_side_by_side() { + // The addresses a node is responsible for share their leading bits. + // Two such looks, each longer than a check waits for its turn, must + // both run rather than one wait and come back Busy. + let looks = FinalityLooks::new(); + let mut first = final_record(14, 0x01).state(); + let mut second = final_record(15, 0x01).state(); + first.address = [0xAB; 32]; + second.address = [0xAB; 32]; + if let Some(last) = second.address.last_mut() { + *last = 0xAC; + } + let slow = || async { + tokio::time::sleep(FINAL_STATE_CHECK_WAIT + Duration::from_secs(1)).await; + None + }; + let (a, b) = tokio::join!(looks.check(&first, slow), looks.check(&second, slow)); + assert_eq!(a, FinalityCheck::Clear); + assert_eq!(b, FinalityCheck::Clear, "the second look was held up"); + } + + #[tokio::test(start_paused = true)] + async fn replays_queued_behind_a_look_reuse_its_answer() { + let looks = FinalityLooks::new(); + let taking = final_record(13, 0x01); + let asked = AtomicUsize::new(0); + let look = || { + asked.fetch_add(1, Ordering::SeqCst); + async { + tokio::time::sleep(Duration::from_millis(500)).await; + None + } + }; + let state = taking.state(); + let checks = (0..8).map(|_| looks.check(&state, look)); + let answers = futures::future::join_all(checks).await; + assert!(answers.iter().all(|answer| *answer == FinalityCheck::Clear)); + assert_eq!(asked.load(Ordering::SeqCst), 1, "eight replays, one look"); + + // Past the reuse window the group is asked again. + tokio::time::sleep(FINAL_STATE_CLEAR_REUSE).await; + assert_eq!(looks.check(&state, look).await, FinalityCheck::Clear); + assert_eq!(asked.load(Ordering::SeqCst), 2); + + // A proof answers every replay of the loser without a look. + let loser = final_record(13, 0x02).state(); + let rival = final_record(13, 0xAA); + let proved = looks.check(&loser, || async { Some(rival.clone()) }).await; + assert_eq!(proved, FinalityCheck::Conflict(rival.state())); + assert_eq!( + looks.check(&loser, look).await, + FinalityCheck::Conflict(rival.state()) + ); + assert_eq!(asked.load(Ordering::SeqCst), 2); + } } diff --git a/tests/e2e/pointer_replication.rs b/tests/e2e/pointer_replication.rs index c7fafd22..8548a7c0 100644 --- a/tests/e2e/pointer_replication.rs +++ b/tests/e2e/pointer_replication.rs @@ -25,7 +25,9 @@ use ant_node::replication::protocol::{ }; use ant_node::replication::slice::nonced_block_root; use ant_node::ReplicationConfig; -use ant_protocol::pointer::{Pointer, PointerState, PointerTarget, PointerTargetKind}; +use ant_protocol::pointer::{ + Pointer, PointerState, PointerTarget, PointerTargetKind, FINAL_COUNTER, +}; use bytes::Bytes; use saorsa_core::identity::PeerId; use saorsa_pqc::api::sig::{ml_dsa_65, MlDsaPublicKey, MlDsaSecretKey}; @@ -60,6 +62,18 @@ fn signed(pk: &MlDsaPublicKey, sk: &MlDsaSecretKey, counter: u64, target: u8) -> .expect("sign") } +/// The final state that hands `pk`'s pointer over to the pointer at +/// `recipient` (ADR-0018). +fn transfer(pk: &MlDsaPublicKey, sk: &MlDsaSecretKey, recipient: u8) -> Pointer { + Pointer::sign( + sk, + pk, + FINAL_COUNTER, + PointerTarget::new(PointerTargetKind::Pointer, [recipient; 32]), + ) + .expect("sign") +} + fn store(node: &TestNode) -> PointerStore { node.ant_protocol .as_ref() @@ -461,6 +475,177 @@ async fn the_possession_check_penalises_only_a_member_that_dropped_the_record() harness.teardown().await.expect("teardown"); } +/// A transfer written to one node reaches the whole group, and after that no +/// node takes a second final state — not even one whose target sorts first, +/// which the previous merge rule let displace the first everywhere (ADR-0018). +#[tokio::test] +#[serial] +async fn a_transfer_reaches_the_group_and_no_node_takes_a_second_one() { + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let (pk, sk) = owner(); + let created = signed(&pk, &sk, 0, 1); + let handed_over = transfer(&pk, &sk, 0x77); + let take_back = transfer(&pk, &sk, 0x01); + assert!(take_back.target().to_bytes() < handed_over.target().to_bytes()); + for record in [&created, &handed_over, &take_back] { + mark_paid(&harness, record); + } + + put(harness.test_node(1).expect("node"), &created).await; + for i in 0..harness.node_count() { + assert!( + wait_for(&harness, i, &created).await, + "node {i} lacks the create" + ); + } + match put(harness.test_node(2).expect("node"), &handed_over).await { + PointerPutResponse::Success { state_id, .. } => { + assert_eq!(state_id, handed_over.state_id()); + } + other => panic!("the transfer was refused: {other:?}"), + } + for i in 0..harness.node_count() { + assert!( + wait_for(&harness, i, &handed_over).await, + "node {i} never received the transfer" + ); + } + + // The former owner, paid up, tries to take it back on every node. + for i in 0..harness.node_count() { + match put(harness.test_node(i).expect("node"), &take_back).await { + PointerPutResponse::Stale { state_id, .. } => assert_eq!( + state_id, + handed_over.state_id(), + "node {i} must name the transfer it holds" + ), + other => panic!("node {i} took a second final state: {other:?}"), + } + assert!(holds(harness.test_node(i).expect("node"), &handed_over)); + } + + harness.teardown().await.expect("teardown"); +} + +/// A node that never heard of a transfer — one that joined after it, or lost +/// its copy — would take any final state on the merge rule alone. Before it +/// does, it asks its group, and a peer that serves the transfer it holds is +/// proof enough to refuse the second one. +#[tokio::test] +#[serial] +async fn a_node_that_missed_the_transfer_refuses_another_its_group_proves() { + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let (pk, sk) = owner(); + let created = signed(&pk, &sk, 0, 1); + let handed_over = transfer(&pk, &sk, 0x77); + let second = transfer(&pk, &sk, 0x01); + mark_paid(&harness, &second); + mark_paid(&harness, &handed_over); + + // Everyone holds the create, and learns everyone else understands + // pointers while there is nothing newer to hint. + let everyone: Vec = (0..harness.node_count()).collect(); + for i in &everyone { + store(harness.test_node(*i).expect("node")) + .put_bytes(&created.to_bytes()) + .await + .expect("put"); + } + exchange_hints(&harness, &everyone, &everyone).await; + + // The transfer lands everywhere but one node, written straight into the + // stores so nothing replicates or hints it there. + let unaware = 4; + for i in others(&harness, &[unaware]) { + store(harness.test_node(i).expect("node")) + .put_bytes(&handed_over.to_bytes()) + .await + .expect("put"); + } + let node = harness.test_node(unaware).expect("node"); + assert!( + holds(node, &created), + "the unaware node has only the create" + ); + + match put(node, &second).await { + PointerPutResponse::Stale { state_id, .. } => assert_eq!( + state_id, + handed_over.state_id(), + "the refusal names the transfer the group proved" + ), + other => panic!("the unaware node took a second final state: {other:?}"), + } + assert!(holds(node, &created), "nothing was written"); + + // The transfer the group holds is not refused: it is no conflict. + match put(node, &handed_over).await { + PointerPutResponse::Success { state_id, .. } => { + assert_eq!(state_id, handed_over.state_id()); + } + other => panic!("the group's own transfer was refused: {other:?}"), + } + + harness.teardown().await.expect("teardown"); +} + +/// Two different final states are a fork only the owner can make, by racing +/// them. A member holding the other side took what reached it first, as the +/// merge rule says; the possession check does not penalise it for the owner's +/// fork, while it still penalises a member that holds nothing. +#[tokio::test] +#[serial] +async fn the_possession_check_does_not_penalise_the_other_side_of_a_fork() { + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let (pk, sk) = owner(); + let one_side = transfer(&pk, &sk, 0x77); + let other_side = transfer(&pk, &sk, 0x01); + let checker = 3; + let forked = 1; + let dropper = 2; + for i in others(&harness, &[forked, dropper]) { + store(harness.test_node(i).expect("node")) + .put_bytes(&one_side.to_bytes()) + .await + .expect("put"); + } + store(harness.test_node(forked).expect("node")) + .put_bytes(&other_side.to_bytes()) + .await + .expect("put"); + + let everyone: Vec = (0..harness.node_count()).collect(); + exchange_hints(&harness, &everyone, &[checker]).await; + + let checker_node = harness.test_node(checker).expect("node"); + let checker_p2p = checker_node.p2p_node.as_ref().expect("p2p"); + let forked_peer = peer(harness.test_node(forked).expect("node")); + let dropper_peer = peer(harness.test_node(dropper).expect("node")); + let forked_before = checker_p2p.peer_trust(&forked_peer); + let dropper_before = checker_p2p.peer_trust(&dropper_peer); + + replication(checker_node) + .check_possession(one_side.state(), &[forked_peer, dropper_peer]) + .await; + + assert!( + checker_p2p.peer_trust(&forked_peer) >= forked_before, + "the member holding the other final state was penalised" + ); + assert!( + checker_p2p.peer_trust(&dropper_peer) < dropper_before, + "the member holding nothing was not penalised" + ); + + harness.teardown().await.expect("teardown"); +} + /// A network whose close group is two nodes, so that in five nodes some node /// is always outside the retention width (two plus the margin of two) for any /// address, with no pruning hysteresis. diff --git a/tests/e2e/testnet.rs b/tests/e2e/testnet.rs index f42156b7..ca33a901 100644 --- a/tests/e2e/testnet.rs +++ b/tests/e2e/testnet.rs @@ -1388,11 +1388,11 @@ impl TestNetwork { .await { Ok(mut engine) => { - // Pointers replicate through the same engine (ADR-0016). + // Pointers replicate through the same engine (ADR-0016), + // which the service also asks before a final state + // (ADR-0018). if let Some(service) = protocol.pointer_service() { - let (writes, fresh_writes) = tokio::sync::mpsc::unbounded_channel(); - service.attach_fresh_writes(writes); - engine.with_pointers(service.store().clone(), fresh_writes); + engine.with_pointer_service(service); } let dht_events = p2p.dht_manager().subscribe_events(); engine.start(dht_events); diff --git a/tests/pointer_convergence.rs b/tests/pointer_convergence.rs index 4e1864aa..6f011c53 100644 --- a/tests/pointer_convergence.rs +++ b/tests/pointer_convergence.rs @@ -6,11 +6,18 @@ //! These are the property tests behind that claim, plus the two anti-abuse //! properties the merge rule exists to provide: one payment funds one state, //! and no re-signature of a stored state can displace it. +//! +//! ADR-0018 carves out one exception: a final state (counter `u64::MAX`) is +//! replaced by nothing, so two *different* final states are unordered and a +//! node keeps the first one it took. The claim holds for every set with at +//! most one final state in it; with two, the first final state delivered wins, +//! and nothing delivered after it moves a node off it. #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] use std::collections::BTreeSet; +use ant_node::pointer::store::{PointerStore, PutOutcome}; use ant_protocol::pointer::{Pointer, PointerTarget, PointerTargetKind}; use proptest::prelude::*; use saorsa_pqc::api::sig::{ @@ -438,8 +445,6 @@ fn the_wire_format_is_what_the_adr_says() { /// layer does, which is where the cost would actually have been paid. #[tokio::test] async fn sixty_four_signatures_buy_exactly_one_write() { - use ant_node::pointer::store::{PointerStore, PutOutcome}; - let dir = tempfile::tempdir().expect("tempdir"); let store = PointerStore::new(dir.path()).await.expect("open store"); @@ -559,11 +564,10 @@ fn an_unknown_version_is_refused_rather_than_accepted_at_its_own_price() { } } -/// At `u64::MAX` no *counter* can out-rank the winner, but a smaller *target* -/// still can. That asymmetry is why migration has to happen before the terminal -/// update rather than on it. +/// At `u64::MAX` nothing out-ranks the held state: no counter is larger, and +/// an equal counter no longer resolves by target (ADR-0018). #[test] -fn a_terminal_counter_cannot_be_out_counted_only_out_targeted() { +fn a_final_state_is_out_ranked_by_nothing() { let terminal = signed(13, u64::MAX, 5, PointerTargetKind::Chunk); assert!(terminal.is_terminal()); assert!(terminal.next_counter().is_err(), "no successor exists"); @@ -575,91 +579,137 @@ fn a_terminal_counter_cannot_be_out_counted_only_out_targeted() { assert!(terminal.replaces(&earlier)); } - // The order does not degenerate there: equal-counter conflicts at the - // maximum still resolve deterministically, so replicas cannot split. + // Nor does another final state, whichever way its target sorts: the first + // one a node took is the one it keeps. let low = signed(13, u64::MAX, 1, PointerTargetKind::Chunk); let high = signed(13, u64::MAX, 2, PointerTargetKind::Chunk); - assert!(low.replaces(&high)); + assert!(!low.replaces(&high)); assert!(!high.replaces(&low)); - assert_eq!(winner(&[&high, &low]).state_id(), low.state_id()); + assert_eq!(winner(&[&high, &low]).state_id(), high.state_id()); assert_eq!(winner(&[&low, &high]).state_id(), low.state_id()); } -/// Why migration must happen *before* the terminal update. -/// -/// At `u64::MAX` the counter can no longer advance, but the pointer is not -/// frozen: the merge order still resolves equal counters by target bytes, and -/// *smaller* target bytes win. So a migration written at the terminal counter -/// can still be displaced — by the owner, or by anyone replaying an older -/// signed record of theirs with a smaller target. The only safe migration is -/// one made while a successor counter still exists, because a strictly larger -/// counter is the one move nothing can answer. +/// A transfer is final on a node's disk: once stored, the former owner cannot +/// take the address back — not with a later counter, which does not exist, and +/// not with a final state whose target sorts first, which the previous rule +/// would have let displace it. #[tokio::test] -async fn migration_must_happen_before_the_terminal_update() { - use ant_node::pointer::store::{PointerStore, PutOutcome}; - use ant_protocol::pointer::PointerTarget; - use saorsa_pqc::api::sig::ml_dsa_65; - +async fn a_stored_transfer_cannot_be_taken_back() { let dir = tempfile::tempdir().expect("tempdir"); let store = PointerStore::new(dir.path()).await.expect("store"); let (pk, sk) = ml_dsa_65().generate_keypair_from_seed(&[14u8; 32]); - let sign_at = |counter: u64, target: PointerTarget| { - Pointer::sign(&sk, &pk, counter, target).expect("sign") - }; - - // One update short of the end: a successor counter still exists. - let penultimate = sign_at( - u64::MAX - 1, + let current = Pointer::create( + &sk, + &pk, PointerTarget::new(PointerTargetKind::Chunk, [0x10u8; 32]), - ); - assert!(!penultimate.is_terminal()); + ) + .expect("create"); assert_eq!( - store.put_bytes(&penultimate.to_bytes()).await.expect("put"), + store.put_bytes(¤t.to_bytes()).await.expect("put"), PutOutcome::Changed ); - // The safe migration: spend the last counter. A strictly larger counter - // beats every target, so nothing at u64::MAX - 1 can answer it. - let migration = sign_at( - penultimate.next_counter().expect("successor exists"), - PointerTarget::new(PointerTargetKind::Pointer, [0x80u8; 32]), - ); + // The handover: one final state, pointing at the recipient's pointer. + let recipient = [0x80u8; 32]; + let transfer = current.transfer_to(&sk, recipient).expect("transfer"); + assert_eq!(transfer.transferred_to(), Some(recipient)); assert_eq!( - store.put_bytes(&migration.to_bytes()).await.expect("put"), + store.put_bytes(&transfer.to_bytes()).await.expect("put"), PutOutcome::Changed ); - assert!(migration.is_terminal()); - assert!(migration.next_counter().is_err()); - // Now the danger. The counter is spent, so the only remaining moves are to - // strictly smaller target bytes — and they still win. - let smaller_target = sign_at( + // A former owner grinding a target that sorts first. Under the previous + // rule this displaced the transfer; now it is stale. + let take_back = Pointer::sign( + &sk, + &pk, u64::MAX, - PointerTarget::new(PointerTargetKind::Chunk, [0x01u8; 32]), - ); - assert!( - smaller_target.to_bytes() != migration.to_bytes(), - "a genuinely different state" - ); + PointerTarget::new(PointerTargetKind::Chunk, [0x00u8; 32]), + ) + .expect("sign"); + assert!(take_back.target().to_bytes() < transfer.target().to_bytes()); assert_eq!( - store - .put_bytes(&smaller_target.to_bytes()) - .await - .expect("put"), - PutOutcome::Changed, - "a terminal pointer is NOT frozen: a smaller target still displaces it" + store.put_bytes(&take_back.to_bytes()).await.expect("put"), + PutOutcome::Stale, + "a second final state must not displace the first" ); - // Larger target bytes cannot claw it back: the move is one-way. + // Nor any lower counter, and the held state is the transfer throughout. + for counter in [0u64, 1, u64::MAX - 1] { + let older = Pointer::sign( + &sk, + &pk, + counter, + PointerTarget::new(PointerTargetKind::Chunk, [0x00u8; 32]), + ) + .expect("sign"); + assert_eq!( + store.put_bytes(&older.to_bytes()).await.expect("put"), + PutOutcome::Stale + ); + } assert_eq!( - store.put_bytes(&migration.to_bytes()).await.expect("put"), - PutOutcome::Stale, - "and the displaced migration can never be restored" + store.state_id(&transfer.address()), + Some(transfer.state_id()) ); +} + +proptest! { + #![proptest_config(ProptestConfig::with_cases(16))] + + /// With one final state among the records, every delivery order still + /// ends on it: rule 0 only leaves two *final* states unordered. + #[test] + fn one_final_state_wins_in_every_delivery_order( + counters in prop::collection::vec(0u64..4, 1..5), + targets in prop::collection::vec(0u8..4, 1..5), + final_target in 0u8..4, + ) { + let mut records: Vec = counters + .iter() + .zip(targets.iter()) + .map(|(counter, target)| signed(1, *counter, *target, PointerTargetKind::Chunk)) + .collect(); + let finalized = signed(1, u64::MAX, final_target, PointerTargetKind::Pointer); + records.push(finalized.clone()); + + let mut order: Vec<&Pointer> = records.iter().collect(); + let mut permutations = 0usize; + permute(&mut order, 0, &mut |candidate| { + prop_assert_eq!(winner(candidate).state_id(), finalized.state_id()); + Ok(()) + }, &mut permutations)?; + prop_assert_eq!(permutations, factorial(records.len())); + } - // Which is the whole point: a migration made at the terminal counter is - // not final, so it has to be made earlier, where the counter still answers. - assert!(smaller_target.replaces(&migration)); - assert!(!migration.replaces(&smaller_target)); + /// With two different final states, the one delivered first is what a + /// node keeps, whatever else arrives before, between or after them. + #[test] + fn the_first_final_state_delivered_is_kept( + counters in prop::collection::vec(0u64..4, 1..4), + targets in prop::collection::vec(0u8..4, 1..4), + first_target in 0u8..8, + second_target in 8u8..16, + ) { + let mut records: Vec = counters + .iter() + .zip(targets.iter()) + .map(|(counter, target)| signed(1, *counter, *target, PointerTargetKind::Chunk)) + .collect(); + records.push(signed(1, u64::MAX, first_target, PointerTargetKind::Pointer)); + records.push(signed(1, u64::MAX, second_target, PointerTargetKind::Pointer)); + + let mut order: Vec<&Pointer> = records.iter().collect(); + let mut permutations = 0usize; + permute(&mut order, 0, &mut |candidate| { + let first_final = candidate + .iter() + .find(|record| record.is_terminal()) + .expect("two finals were delivered"); + prop_assert_eq!(winner(candidate).state_id(), first_final.state_id()); + Ok(()) + }, &mut permutations)?; + prop_assert_eq!(permutations, factorial(records.len())); + } }