Struct Node 

pub struct Node { /* private fields */ }

Implementations

impl Node

pub fn new() -> Self

Creates a new root-level node with default configuration and in-memory storage.

This is the simplest way to get started with BEAM. The node will use MemoryStorage and have no network adapters connected.

pub fn id(&self) -> String

Returns the unique identifier of this node (the path joined by /).

The root node has an empty uid. Child nodes have uids like "parent_key/child_key".

pub fn peer_id(&self) -> String

Returns the peer ID of this node’s actor context.

The peer ID is a random string generated at node creation time. It identifies this node instance in the P2P mesh.

pub fn new_with_config( config: Config, storage_adapters: Vec<Box<dyn Actor>>, network_adapters: Vec<Box<dyn Actor>>, ) -> Self

Creates a new root-level node with custom configuration, storage, and network adapters.

Arguments

• config - Node configuration (see Config)
• storage_adapters - Storage actors (e.g. MemoryStorage, RedbStorage)
• network_adapters - Network actors (e.g. OutgoingWebsocketManager, WsServer)

pub fn metrics(&self) -> Arc<Metrics>

Returns a clone of the shared Arc<Metrics> handle.

The returned Arc points to the same atomic counters as the Router’s internal field. External observers (tests, telemetry exporters) read counters via crate::metrics::Metrics::snapshot or record events via crate::metrics::Metrics::record_dropped_send.

Cloning the Arc is cheap (refcount bump); the atomic counters are shared across all clones.

pub async fn flush_storage( &self, timeout: Option<Duration>, ) -> Result<(), String>

Flushes pending writes to persistent storage and waits for acknowledgement.

Sends a [Flush] message to all storage adapters via the router, then waits for the first adapter to acknowledge. If no acknowledgement arrives within the timeout, returns an error.

Arguments

• timeout - Maximum time to wait. Defaults to 30 seconds if None.

Errors

• "router not initialized" — node has no router (shouldn’t happen in normal use)
• "failed to send flush to router" — router channel is closed
• "flush ack channel closed" — oneshot sender was dropped
• "flush timed out" — no acknowledgement within timeout

pub fn on(&mut self) -> Receiver<Value>

Subscribes to this node’s value updates.

Returns a broadcast::Receiver that will receive Value updates whenever the node’s value changes. The current value (if any) is requested from storage via a Get message — it arrives asynchronously.

pub async fn once(&mut self, wait: Option<Duration>) -> Option<Value>

Reads the node’s value once, or None if not found within the timeout.

This is a convenience wrapper around Node::on with a timeout. The default timeout is 66ms (matching Gun.js’s opt.wait).

Arguments

• wait - Optional timeout. Defaults to 66ms.

pub fn connect_peer(&self, url: &str)

Connects to a remote peer via WebSocket with automatic reconnection.

Retries with exponential backoff starting at 1 second, maxing at 60 seconds. The spawned WsConn actor auto-handshakes via [Message::Hi] on pre_start, and the [Router] registers the peer on Hi receipt.

Arguments

• url - WebSocket URL (e.g. "wss://relay.example.com/ws")

Panics

Panics if the URL is invalid (should not happen with well-formed URLs).

pub fn get(&mut self, key: &str) -> Node

Returns a child node corresponding to the given key, creating it if necessary.

This is the primary graph traversal method. Calling node.get("key") returns a child node. If the child already exists, the existing instance is returned; otherwise a new child is created lazily.

Arguments

• key - The child key. Must not be empty (empty returns self).

pub fn map(&self) -> Receiver<(String, Value)>

Subscribes to all children of this node.

Returns a broadcast::Receiver that emits (child_key, value) tuples for each child. The current children (if any) are requested from storage via a Get message.

pub async fn put_quorum( &mut self, value: Value, policy: AckPolicy, ) -> Result<ReplicationStatus, String>

Writes a value to this node and waits for storage acknowledgement.

The value is immediately sent to local on() subscribers (so any in-process listeners see it before the ack returns), then a [Put] message is sent to the router for storage and network relay. The timestamp is the current Unix epoch in milliseconds.

Returns Ok(()) once a storage adapter has committed the put durably and acked back. Returns Err(String) if the adapter reports a commit failure, the router was not initialized, the router channel rejected the message, or the ack did not arrive within the timeout window (default 30 seconds).

Why Async?

Without the ack, put returns synchronously after queuing the message to the storage actor — but the storage actor may not have processed the message yet. A subsequent get or once would read stale state. The async ack closes that race: this future resolves only after the storage actor has committed the put. See module docs for the full race-condition history.

Ack Pattern

Mirrors Node::flush_storage:

1. Register a oneshot keyed by the put’s id in pending_puts
2. Send Message::Put to the router
3. Storage adapter commits, then sends Put { in_response_to: Some(id), updated_nodes: { "_ack": { "_ack"|"_err": ... } } } back to this node’s addr directly (NOT through the router)
4. Node::handle_put drains pending_puts and resolves the oneshot

Arguments

• value - The value to set (see Value for supported types)

Writes a value and waits for it to be replicated to N peers per the given AckPolicy.

Resolves with a ReplicationStatus when the policy threshold is satisfied, or Err(String) on timeout, router failure, or unrecoverable storage error.

Wire-level flow

1. Build a [Put] and register a oneshot in pending_puts
2. Send Message::RegisterQuorum { put_id, requester, policy } to Router (this creates a tracked [crate::router::QuorumEntry])
3. Send Message::Put(put) to Router for relay to peers
4. Peers eventually reply with Put { @: put_id, .. }
5. Router’s handle_put ack branch counts each peer ack in the QuorumEntry
6. When acked_by >= policy.quorum, Router sends a sentinel Put { @: put_id, updated_nodes: { "__quorum_met__": ack_count } } back to this Node
7. This Node’s handle_put drain decodes the sentinel via Node::decode_quorum_payload and resolves the oneshot with the ReplicationStatus

Examples

use beam::{Node, Value, AckPolicy};

let node = Node::new();
let policy = AckPolicy::for_peer_count(3); // majority of 3 peers
let status = node.put_quorum(Value::Text("hello".into()), policy).await?;
assert!(status.quorum_met);
assert!(status.acked_by >= 2);

Arguments

• value - The value to set (see Value for supported types)
• policy - The AckPolicy describing how many peer acks are required and how long to wait

Errors

Returns Err(String) if:

• The router is not initialized
• The Router fails to receive the RegisterQuorum or Put message
• The policy timeout elapses before quorum is met
• The ack channel closes (Router dropped us)

pub async fn put(&mut self, value: Value) -> Result<(), String>

pub async fn batch_put( &mut self, ops: Vec<(Vec<String>, Value)>, ) -> Result<(), String>

Writes multiple values in a single storage transaction and waits for ack.

Each operation is a (path, value) pair where path is a vector of keys from the caller’s Node down to the leaf. The caller should invoke this on the root Node.

Returns Ok(()) once the storage adapter has committed the batch durably and acked back. Returns Err(String) on commit failure, router error, or ack timeout (default 30s).

Atomicity

Unlike multiple sequential put calls, all operations in a batch either succeed together or fail together. Storage adapters that support transactions (e.g. crate::adapters::RedbStorage) wrap the batch in a single transaction.

Ack Pattern

The whole batch shares a single ack keyed by BatchPut.id. The originating node registers one oneshot, the storage adapter sends one ack after the entire transaction commits or aborts.

Arguments

• ops - Vector of (path, value) pairs

pub fn stop(&mut self)

Stops the node and all its child actors and adapters.

This calls ActorContext::stop from the node’s actor context, which aborts all child tasks and sends stop signals to all child actors.

pub async fn shutdown(&mut self, timeout: Duration) -> Result<(), String>

Gracefully shuts down the node, ensuring data integrity.

This is the preferred shutdown path. The sequence is:

1. Flush storage — calls Node::flush_storage to ensure all pending writes in the actor mailboxes are processed and committed by the storage adapters. The router processes messages in order, so any puts ahead of the flush are committed before the flush ack returns.

2. Signal shutdown — broadcasts true on the shutdown watch channel. Long-running child tasks (accept loops, signal processors) that select! on shutdown_rx break their loops and stop accepting new connections or work.

3. Drain — waits briefly for in-flight messages to complete and network connections to close. The drain duration is bounded by the remaining time budget after the flush.

4. Force stop — calls Node::stop to abort any remaining tasks and send stop signals to all child actors. This is the same as a hard shutdown, but by this point all critical work should already be done.

Arguments

• timeout — maximum total time for the graceful shutdown sequence. If the flush and drain do not complete within this duration, the method proceeds to force stop and returns an error.

Returns

• Ok(()) — graceful shutdown completed within the timeout.
• Err(String) — timed out; force stop was used. The error message describes which phase timed out.

Example

use web_time::Duration;
use beam::Node;

let mut node = Node::new();
// ... use node ...
if let Err(e) = node.shutdown(Duration::from_secs(30)).await {
    eprintln!("graceful shutdown timed out: {}, force-stopped", e);
}

impl Node

pub async fn put_with_options( &mut self, value: Value, _options: PutOptions, ) -> Result<(), String>

Sets a value with options (currently cert is no-op; reserved for future enforcement).

See Node::put for the basic version. The options parameter allows passing a [PutOptions] with a certificate for delegated writes, though certificate enforcement is not yet implemented.

impl Node

pub fn user(&mut self) -> UserBuilder<'_>

Begin user creation or authentication on this node

Trait Implementations

impl Actor for Node

fn handle<'life0, 'life1, 'async_trait>( &'life0 mut self, msg: Message, _context: &'life1 ActorContext, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Handle an incoming message.

fn pre_start<'life0, 'life1, 'async_trait>( &'life0 mut self, _context: &'life1 ActorContext, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Called once before the actor starts processing messages.

fn stopping<'life0, 'life1, 'async_trait>( &'life0 mut self, _context: &'life1 ActorContext, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Called once after the actor’s message loop exits.

fn subscribe_to_everything(&self) -> bool

Whether this actor wants to receive all messages (not just addressed to it). Used by the Multicast adapter.

fn try_clone_storage(&self) -> Option<Box<dyn Actor>>

Attempts to produce a clone of this actor for storage read/write splitting.

impl Clone for Node

fn clone(&self) -> Node

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Default for Node

fn default() -> Self

Returns the “default value” for a type.