diff --git a/src/core/context/context_test.rs b/src/core/context/context_test.rs index 40a01a2..b5a72b5 100644 --- a/src/core/context/context_test.rs +++ b/src/core/context/context_test.rs @@ -155,8 +155,11 @@ mod tests { /// a context created with a timeout fails a longer-running operation with a deadline error #[tokio::test] async fn test_run_respects_deadline() { - let ctx = - IrrevocableContext::with_timeout(&span_fixture(), "timeout_ctx", Duration::from_millis(10)); + let ctx = IrrevocableContext::with_timeout( + &span_fixture(), + "timeout_ctx", + Duration::from_millis(10), + ); let result = ctx .run(async { @@ -166,14 +169,20 @@ mod tests { .await; assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("deadline exceeded")); + assert!(result + .unwrap_err() + .to_string() + .contains("deadline exceeded")); } /// is_deadline_exceeded reflects the deadline crossing #[tokio::test] async fn test_is_deadline_exceeded() { - let ctx = - IrrevocableContext::with_timeout(&span_fixture(), "timeout_ctx", Duration::from_millis(5)); + let ctx = IrrevocableContext::with_timeout( + &span_fixture(), + "timeout_ctx", + Duration::from_millis(5), + ); assert!(!ctx.is_deadline_exceeded()); assert!(ctx.deadline().is_some()); sleep(Duration::from_millis(20)).await; @@ -191,17 +200,23 @@ mod tests { cancel(); let child_clone = child.clone(); - wait_until(move || child_clone.is_cancelled(), Duration::from_millis(100)) - .await - .expect("child should be cancelled after invoking cancel closure"); + wait_until( + move || child_clone.is_cancelled(), + Duration::from_millis(100), + ) + .await + .expect("child should be cancelled after invoking cancel closure"); } /// run() on an already-expired context returns the deadline error even when the /// operation itself is immediately ready (guards the short-circuit / select! race) #[tokio::test] async fn test_run_short_circuits_when_deadline_already_passed() { - let ctx = - IrrevocableContext::with_timeout(&span_fixture(), "expired_ctx", Duration::from_millis(5)); + let ctx = IrrevocableContext::with_timeout( + &span_fixture(), + "expired_ctx", + Duration::from_millis(5), + ); // let the deadline lapse before we ever call run() sleep(Duration::from_millis(20)).await; assert!(ctx.is_deadline_exceeded()); @@ -210,7 +225,10 @@ mod tests { let result = ctx.run(async { Ok::(42) }).await; assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("deadline exceeded")); + assert!(result + .unwrap_err() + .to_string() + .contains("deadline exceeded")); } /// cancellation that fires *while* run() is in-flight wins the race inside the @@ -238,6 +256,9 @@ mod tests { ); assert!(result.is_err()); - assert!(result.unwrap_err().to_string().contains("context cancelled")); + assert!(result + .unwrap_err() + .to_string() + .contains("context cancelled")); } } diff --git a/src/core/mod.rs b/src/core/mod.rs index 323c974..861606d 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -10,7 +10,18 @@ pub use crate::core::lookup::array_lookup_table::LOOKUP_TABLE_LEVELS; pub use crate::core::lookup::LookupTable; pub use crate::core::lookup::LookupTableLevel; pub use crate::core::model::address::Address; +pub use crate::core::model::direction::Direction; pub use crate::core::model::identifier::Identifier; +pub use crate::core::model::identity::Identity; pub use crate::core::model::memvec::MembershipVector; +pub use model::join::BuddyReq; +pub use model::join::CheckNeighborReq; +pub use model::join::LinkReq; +pub use model::join::LinkRes; +pub use model::join::MaxLevelReq; +pub use model::join::MaxLevelRes; +pub use model::join::NeighborReq; +pub use model::join::NeighborRes; pub use model::search::IdSearchReq; pub use model::search::IdSearchRes; +pub use model::search::Nonce; diff --git a/src/core/model/join.rs b/src/core/model/join.rs new file mode 100644 index 0000000..53330f3 --- /dev/null +++ b/src/core/model/join.rs @@ -0,0 +1,158 @@ +//! request/response payloads for the join (insert) protocol and its background +//! backpointer-repair mechanism (see `docs/protocol/concurrent-insert.md`). +//! +//! every `side`/`direction` field is receiver-owned and never re-interpreted +//! hop-to-hop: `Direction::Right` always means the receiving node's own right +//! slot (nodes with larger identifiers), `Direction::Left` always the receiving +//! node's own left slot — a global concept, not relative to the current hop or +//! the originator. +//! +//! forwardable requests ([`LinkReq`], [`BuddyReq`], [`CheckNeighborReq`]) carry +//! the originator's/claimant's full [`Identity`] so the node terminating the +//! chain can install the entry in its own table and reply directly to the +//! originator. the `nonce` is set once by the originator and threaded unchanged +//! through every hop. + +use crate::core::lookup::LookupTableLevel; +use crate::core::model::direction::Direction; +use crate::core::model::identity::Identity; +use crate::core::model::search::Nonce; +use crate::core::Identifier; + +/// asks the introducer for the highest level at which it has any populated +/// lookup-table entry, seeding the joining node's stage-1 search level. +#[derive(Debug, Copy, Clone)] +pub struct MaxLevelReq { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// the identifier of the joining node that initiated the request. + pub origin: Identifier, +} + +/// the introducer's reply to [`MaxLevelReq`]. +#[derive(Debug, Copy, Clone)] +pub struct MaxLevelRes { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// the highest level at which the responder has any populated lookup-table entry. + pub max_level: LookupTableLevel, +} + +/// asks a node for its current neighbor entry at a given level and direction. +#[derive(Debug, Copy, Clone)] +pub struct NeighborReq { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// the identifier of the node that initiated the request. + pub origin: Identifier, + /// the lookup-table level being queried. + pub level: LookupTableLevel, + /// receiver-owned direction: `Direction::Right` always means the receiving node's own + /// right slot (nodes with larger identifiers), `Direction::Left` its own left slot — + /// never re-interpreted hop-to-hop, not relative to the current hop or the originator. + pub direction: Direction, +} + +/// the reply to [`NeighborReq`], carrying the queried neighbor entry, if any. +#[derive(Debug, Copy, Clone)] +pub struct NeighborRes { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// the lookup-table level that was queried. + pub level: LookupTableLevel, + /// receiver-owned direction: `Direction::Right` always means the responding node's own + /// right slot (nodes with larger identifiers), `Direction::Left` its own left slot — + /// never re-interpreted hop-to-hop, not relative to the current hop or the originator. + pub direction: Direction, + /// the responder's neighbor entry at the queried level and direction, if populated. + pub neighbor: Option, +} + +/// asks the receiver to adopt the candidate as its neighbor on the given side and +/// level. the request is forwardable: a receiver with an existing neighbor sitting +/// between itself and the candidate passes the request along instead of linking, so +/// it may travel several hops before some node installs the candidate. that node +/// replies directly to the candidate with a [`LinkRes`], which is why the +/// candidate's full [`Identity`] travels in the request. +#[derive(Debug, Copy, Clone)] +pub struct LinkReq { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// the full identity of the joining node proposing itself as the receiver's + /// neighbor at this side and level; the receiver either installs it in its own + /// table or forwards the request when an existing neighbor sits between them. + pub candidate: Identity, + /// receiver-owned side: `Direction::Right` always means the receiving node's own + /// right slot (nodes with larger identifiers), `Direction::Left` its own left slot — + /// never re-interpreted hop-to-hop, not relative to the current hop or the originator. + pub side: Direction, + /// the lookup-table level at which the link is requested. + pub level: LookupTableLevel, +} + +/// reports how a link request ended: `linked` names the node that adopted the +/// candidate as its neighbor, or is `None` when the forwarding chain ran out +/// without finding anyone to link. besides answering [`LinkReq`] and [`BuddyReq`], +/// this message also arrives unsolicited during backpointer repair, when a node +/// pushes a correction telling the recipient to update its own pointer — so a +/// receiver must not assume it matches a pending request. +#[derive(Debug, Copy, Clone)] +pub struct LinkRes { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// receiver-owned side: `Direction::Right` always means the receiving node's own + /// right slot (nodes with larger identifiers), `Direction::Left` its own left slot — + /// never re-interpreted hop-to-hop, not relative to the current hop or the originator. + pub side: Direction, + /// the lookup-table level the link decision applies to. + pub level: LookupTableLevel, + /// the identity of the node that installed the candidate in its own table — + /// from the candidate's point of view, its new neighbor at this side and level; + /// `None` means no candidate exists on this side at this level. + pub linked: Option, +} + +/// searches for the candidate's "buddy": the nearest node on the given side whose +/// membership vector shares at least `level` prefix bits with the candidate — that +/// is, the candidate's neighbor-to-be at `level`. the name follows the `buddyOp` +/// message of the skip graphs paper. unlike [`NeighborReq`] (a read-only query +/// answered in place), this is a search: each receiver that fails the prefix check +/// forwards the request outward along its links at the level below, and the first +/// node that passes it installs the candidate in its own table and replies directly +/// with a [`LinkRes`]. there is no dedicated response type: `linked: None` in the +/// [`LinkRes`] means the chain ran out — no buddy exists on this side at this level. +#[derive(Debug, Copy, Clone)] +pub struct BuddyReq { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// the full identity of the joining node seeking a buddy at this level. + pub candidate: Identity, + /// receiver-owned side: `Direction::Right` always means the receiving node's own + /// right slot (nodes with larger identifiers), `Direction::Left` its own left slot — + /// never re-interpreted hop-to-hop, not relative to the current hop or the originator. + pub side: Direction, + /// the lookup-table level at which a neighbor is sought. + pub level: LookupTableLevel, +} + +/// a periodic repair probe asking the receiver "does your `side` slot at `level` +/// point back at the claimant?" — sent by a node sweeping its own table to verify +/// that each of its neighbors holds the reciprocal pointer. the probe is +/// forwardable: a receiver whose slot holds a node sitting strictly between itself +/// and the claimant passes the probe to that node instead, so it may travel several +/// hops. the node where it stops either finds the pointer already correct (no-op) +/// or relinks the slot to the claimant and pushes a [`LinkRes`] correction back to +/// it — which is why the claimant's full [`Identity`] travels in the probe. +#[derive(Debug, Copy, Clone)] +pub struct CheckNeighborReq { + /// correlation nonce, set once by the originator and threaded unchanged through every hop. + pub nonce: Nonce, + /// the full identity of the node claiming to be the receiver's neighbor. + pub claimant: Identity, + /// receiver-owned side: `Direction::Right` always means the receiving node's own + /// right slot (nodes with larger identifiers), `Direction::Left` its own left slot — + /// never re-interpreted hop-to-hop, not relative to the current hop or the originator. + pub side: Direction, + /// the lookup-table level being checked. + pub level: LookupTableLevel, +} diff --git a/src/core/model/mod.rs b/src/core/model/mod.rs index da234e9..2de6281 100644 --- a/src/core/model/mod.rs +++ b/src/core/model/mod.rs @@ -5,5 +5,6 @@ pub mod address; pub(crate) mod direction; pub mod identifier; pub mod identity; +pub(crate) mod join; pub mod memvec; pub(crate) mod search; diff --git a/src/network/mod.rs b/src/network/mod.rs index a81281b..e51e069 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -1,7 +1,10 @@ pub mod mock; mod processor; -use crate::core::{IdSearchReq, IdSearchRes, Identifier}; +use crate::core::{ + BuddyReq, CheckNeighborReq, IdSearchReq, IdSearchRes, Identifier, LinkReq, LinkRes, + MaxLevelReq, MaxLevelRes, NeighborReq, NeighborRes, +}; #[allow(unused)] pub use processor::MessageProcessor; @@ -9,9 +12,28 @@ pub use processor::MessageProcessor; /// Event is an application-layer semantic contrast to the lower-level transport-layer Message struct. #[derive(Debug, Clone)] pub enum Event { - TestMessage(String), // A payload for testing purposes, it is a simple string event, and is not used in production. - SearchByIdRequest(IdSearchReq), // A payload representing an identifier search request. - SearchByIdResponse(IdSearchRes), // A payload representing an identifier search response. + /// a payload for testing purposes: a simple string event, not used in production. + TestMessage(String), + /// a payload representing an identifier search request. + SearchByIdRequest(IdSearchReq), + /// a payload representing an identifier search response. + SearchByIdResponse(IdSearchRes), + /// a request for the receiver's highest populated lookup-table level. + GetMaxLevelOp(MaxLevelReq), + /// the response carrying the responder's highest populated lookup-table level. + RetMaxLevelOp(MaxLevelRes), + /// a request for the receiver's neighbor entry at a given level and direction. + GetNeighborOp(NeighborReq), + /// the response carrying the queried neighbor entry, if any. + RetNeighborOp(NeighborRes), + /// a forwardable request to link the candidate at the receiver's given side and level. + GetLinkOp(LinkReq), + /// a link confirmation reply; also reused as a repair push correction. + SetLinkOp(LinkRes), + /// a forwardable stage-2 request for a prefix-matching neighbor at a given level. + BuddyOp(BuddyReq), + /// a forwardable periodic repair probe checking backpointer consistency. + CheckNeighborOp(CheckNeighborReq), } /// Core event processing logic that implementations must provide.