You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/08/28 00:51:28 UTC
(maka) branch feat/runtime-host-peer-mesh-membership created (now 4cae1a9f4)
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a change to branch feat/runtime-host-peer-mesh-membership
in repository https://gitbox.apache.org/repos/asf/maka.git
at 4cae1a9f4 feat(runtime-host): add peer mesh membership
This branch includes the following new commits:
new 4cae1a9f4 feat(runtime-host): add peer mesh membership
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
(maka) 01/01: feat(runtime-host): add peer mesh membership
Posted by wy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch feat/runtime-host-peer-mesh-membership
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 4cae1a9f4dba836222327e552d6b463ef670bc4a
Author: Wang <m4...@gmail.com>
AuthorDate: Fri Aug 28 08:50:41 2026 +0800
feat(runtime-host): add peer mesh membership
Establish an independent signed Mesh membership authority over the shared libp2p endpoint. Add bounded one-time invitations, durable roster state, authenticated control streams, and headless CLI operations without granting Runtime Host access.
Generated-by: OpenAI Codex
---
native/runtime-host-peer/Cargo.lock | 15 +
native/runtime-host-peer/Cargo.toml | 1 +
native/runtime-host-peer/src/bindings.rs | 107 ++--
native/runtime-host-peer/src/engine.rs | 269 ++++++++--
.../src/engine/application_stream.rs | 7 +
native/runtime-host-peer/src/engine/peer_stream.rs | 13 +-
.../cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv | 1 +
.../cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt | 11 +-
packages/cli/src/cli-core.ts | 13 +-
packages/cli/src/runtime-host-cli.ts | 116 ++++-
packages/cli/src/runtime-host-peer-mesh-command.ts | 131 +++++
packages/runtime-host/package.json | 1 +
.../src/__tests__/peer-listener.test.ts | 7 +
.../runtime-host/src/__tests__/peer-mesh.test.ts | 198 ++++++++
.../runtime-host/src/__tests__/peer-native.test.ts | 1 +
packages/runtime-host/src/client/peer-client.ts | 59 ++-
packages/runtime-host/src/peer-mesh/index.ts | 31 ++
packages/runtime-host/src/peer-mesh/model.ts | 355 ++++++++++++++
packages/runtime-host/src/peer-mesh/node.ts | 546 +++++++++++++++++++++
packages/runtime-host/src/peer-mesh/store.ts | 340 +++++++++++++
packages/runtime-host/src/transport/peer-native.ts | 9 +
packages/storage/package.json | 1 +
packages/storage/src/file-lifetime-owner.ts | 51 ++
23 files changed, 2181 insertions(+), 102 deletions(-)
diff --git a/native/runtime-host-peer/Cargo.lock b/native/runtime-host-peer/Cargo.lock
index 12bcd90ac..e311cd232 100644
--- a/native/runtime-host-peer/Cargo.lock
+++ b/native/runtime-host-peer/Cargo.lock
@@ -1563,6 +1563,20 @@ dependencies = [
"web-time",
]
+[[package]]
+name = "libp2p-stream"
+version = "0.4.0-alpha"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d6bd8025c80205ec2810cfb28b02f362ab48a01bee32c50ab5f12761e033464"
+dependencies = [
+ "futures",
+ "libp2p-core",
+ "libp2p-identity",
+ "libp2p-swarm",
+ "rand 0.8.8",
+ "tracing",
+]
+
[[package]]
name = "libp2p-swarm"
version = "0.47.1"
@@ -1700,6 +1714,7 @@ version = "0.0.0"
dependencies = [
"futures",
"libp2p",
+ "libp2p-stream",
"napi",
"napi-build",
"napi-derive",
diff --git a/native/runtime-host-peer/Cargo.toml b/native/runtime-host-peer/Cargo.toml
index 5b086ce9d..3b359be85 100644
--- a/native/runtime-host-peer/Cargo.toml
+++ b/native/runtime-host-peer/Cargo.toml
@@ -41,6 +41,7 @@ libp2p = { version = "0.56", features = [
"tokio",
"yamux",
] }
+libp2p-stream = "0.4.0-alpha"
napi = { version = "3.12", features = ["napi8", "tokio_rt"] }
napi-derive = "3.5"
tokio = { version = "1.53", features = ["fs", "io-util", "rt-multi-thread", "sync", "time"] }
diff --git a/native/runtime-host-peer/src/bindings.rs b/native/runtime-host-peer/src/bindings.rs
index 2d7603567..433f77a56 100644
--- a/native/runtime-host-peer/src/bindings.rs
+++ b/native/runtime-host-peer/src/bindings.rs
@@ -55,6 +55,7 @@ pub struct PeerEndpoint {
listen_addresses: Vec<String>,
commands: mpsc::Sender<EngineCommand>,
incoming: Arc<AsyncMutex<mpsc::Receiver<engine::PeerStream>>>,
+ mesh_incoming: Arc<AsyncMutex<mpsc::Receiver<engine::PeerStream>>>,
terminal: Arc<AsyncMutex<mpsc::Receiver<PeerError>>>,
thread: Arc<Mutex<Option<std::thread::JoinHandle<()>>>>,
}
@@ -73,43 +74,23 @@ impl PeerEndpoint {
#[napi]
pub async fn connect(&self, options: ConnectPeerOptions) -> Result<PeerStream> {
- let peer_id = parse_peer_id(&options.peer_id)?;
- let route_hints = parse_addresses(options.route_hints, "route hint")?;
- let coordination_relays = parse_addresses(
- options.coordination_relays.unwrap_or_default(),
- "coordination relay",
- )?;
- if !(1..=120_000).contains(&options.direct_deadline_ms) {
- return Err(Error::new(
- Status::InvalidArg,
- "direct deadline must be between 1 and 120000 milliseconds",
- ));
- }
- let (result_tx, result_rx) = oneshot::channel();
- self.commands
- .send(EngineCommand::Connect {
- options: engine::ConnectOptions {
- request_id: options.request_id,
- peer_id,
- route_hints,
- coordination_relays,
- deadline: Duration::from_millis(u64::from(options.direct_deadline_ms)),
- },
- result: result_tx,
- })
+ connect_peer(self, options, engine::StreamKind::Application).await
+ }
+
+ #[napi]
+ pub async fn connect_mesh_control(&self, options: ConnectPeerOptions) -> Result<PeerStream> {
+ connect_peer(self, options, engine::StreamKind::MeshControl).await
+ }
+
+ #[napi]
+ pub async fn accept_mesh_control(&self) -> Result<Option<PeerStream>> {
+ self.mesh_incoming
+ .lock()
.await
- .map_err(|_| {
- peer_error(PeerError {
- code: "peer_native_failed",
- message: "peer endpoint is closed".to_owned(),
- })
- })?;
- wrap_stream(
- result_rx
- .await
- .map_err(|_| native_closed_error())?
- .map_err(peer_error)?,
- )
+ .recv()
+ .await
+ .map(wrap_stream)
+ .transpose()
}
#[napi]
@@ -164,6 +145,52 @@ impl PeerEndpoint {
}
}
+async fn connect_peer(
+ endpoint: &PeerEndpoint,
+ options: ConnectPeerOptions,
+ stream_kind: engine::StreamKind,
+) -> Result<PeerStream> {
+ let peer_id = parse_peer_id(&options.peer_id)?;
+ let route_hints = parse_addresses(options.route_hints, "route hint")?;
+ let coordination_relays = parse_addresses(
+ options.coordination_relays.unwrap_or_default(),
+ "coordination relay",
+ )?;
+ if !(1..=120_000).contains(&options.direct_deadline_ms) {
+ return Err(Error::new(
+ Status::InvalidArg,
+ "direct deadline must be between 1 and 120000 milliseconds",
+ ));
+ }
+ let (result_tx, result_rx) = oneshot::channel();
+ endpoint
+ .commands
+ .send(EngineCommand::Connect {
+ options: engine::ConnectOptions {
+ request_id: options.request_id,
+ peer_id,
+ route_hints,
+ coordination_relays,
+ deadline: Duration::from_millis(u64::from(options.direct_deadline_ms)),
+ },
+ stream_kind,
+ result: result_tx,
+ })
+ .await
+ .map_err(|_| {
+ peer_error(PeerError {
+ code: "peer_native_failed",
+ message: "peer endpoint is closed".to_owned(),
+ })
+ })?;
+ wrap_stream(
+ result_rx
+ .await
+ .map_err(|_| native_closed_error())?
+ .map_err(peer_error)?,
+ )
+}
+
impl Drop for PeerEndpoint {
fn drop(&mut self) {
if Arc::strong_count(&self.thread) == 1 {
@@ -175,12 +202,18 @@ impl Drop for PeerEndpoint {
#[napi]
pub struct PeerStream {
+ peer_id: String,
incoming: Arc<AsyncMutex<IncomingStreamReceiver>>,
commands: mpsc::Sender<StreamCommand>,
}
#[napi]
impl PeerStream {
+ #[napi(getter)]
+ pub fn peer_id(&self) -> String {
+ self.peer_id.clone()
+ }
+
#[napi]
pub async fn read(&self) -> Result<Option<Buffer>> {
match self.incoming.lock().await.recv().await {
@@ -253,6 +286,7 @@ pub fn start_peer_endpoint(options: StartPeerEndpointOptions) -> Result<PeerEndp
.collect(),
commands: started.commands,
incoming: Arc::new(AsyncMutex::new(started.incoming)),
+ mesh_incoming: Arc::new(AsyncMutex::new(started.mesh_incoming)),
terminal: Arc::new(AsyncMutex::new(started.terminal)),
thread: Arc::new(Mutex::new(Some(started.thread))),
})
@@ -268,6 +302,7 @@ pub async fn ensure_peer_identity(key_path: String) -> Result<String> {
fn wrap_stream(stream: engine::PeerStream) -> Result<PeerStream> {
Ok(PeerStream {
+ peer_id: stream.peer_id.to_string(),
incoming: Arc::new(AsyncMutex::new(stream.incoming)),
commands: stream.commands,
})
diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs
index b39ac5178..9aeb542e4 100644
--- a/native/runtime-host-peer/src/engine.rs
+++ b/native/runtime-host-peer/src/engine.rs
@@ -52,6 +52,7 @@ use peer_stream::spawn_stream;
pub use peer_stream::{PeerStream, StreamCommand};
const APPLICATION_PROTOCOL: &str = "/maka/runtime-host/peer/1";
+const MESH_CONTROL_PROTOCOL: &str = "/maka/runtime-host/mesh-control/1";
const IDENTIFY_PROTOCOL: &str = "/maka/runtime-host/peer-identify/1";
const COMMAND_CAPACITY: usize = 32;
const INCOMING_STREAM_CAPACITY: usize = 16;
@@ -76,6 +77,7 @@ pub struct StartedEndpoint {
pub listen_addresses: Vec<Multiaddr>,
pub commands: mpsc::Sender<EngineCommand>,
pub incoming: mpsc::Receiver<PeerStream>,
+ pub mesh_incoming: mpsc::Receiver<PeerStream>,
pub terminal: mpsc::Receiver<PeerError>,
pub thread: thread::JoinHandle<()>,
}
@@ -91,6 +93,7 @@ pub struct ConnectOptions {
pub enum EngineCommand {
Connect {
options: ConnectOptions,
+ stream_kind: StreamKind,
result: oneshot::Sender<Result<PeerStream, PeerError>>,
},
CancelConnect {
@@ -125,11 +128,13 @@ struct Behaviour {
identify: identify::Behaviour,
ping: ping::Behaviour,
application_stream: application_stream::Behaviour,
+ mesh_control: libp2p_stream::Behaviour,
}
struct PendingConnect {
peer_id: PeerId,
result: oneshot::Sender<Result<PeerStream, PeerError>>,
+ stream_kind: StreamKind,
deadline: Instant,
opening: Option<tokio::task::JoinHandle<()>>,
dials: HashMap<ConnectionId, DialOrigin>,
@@ -140,7 +145,13 @@ struct PendingConnect {
}
#[derive(Clone, Copy, PartialEq, Eq)]
-enum DialOrigin {
+pub enum StreamKind {
+ Application,
+ MeshControl,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub(super) enum DialOrigin {
DirectRoute,
CoordinationRoute,
DirectConnection,
@@ -215,7 +226,20 @@ impl CoordinationRelay {
struct OpenedStream {
request_id: u32,
- result: Result<application_stream::OpenedApplicationStream, String>,
+ result: Result<OpenedPeerStream, String>,
+}
+
+struct OpenedPeerStream {
+ connection_id: Option<ConnectionId>,
+ stream: libp2p::swarm::Stream,
+}
+
+pub(super) enum StreamCompletion {
+ Application(ConnectionId),
+ MeshControl {
+ dials: HashMap<ConnectionId, DialOrigin>,
+ coordination_relay_peers: Vec<PeerId>,
+ },
}
pub async fn ensure_identity(key_path: PathBuf) -> Result<PeerId, PeerError> {
@@ -229,11 +253,18 @@ pub fn start(options: StartOptions) -> Result<StartedEndpoint, PeerError> {
let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1);
let (command_tx, command_rx) = mpsc::channel(COMMAND_CAPACITY);
let (incoming_tx, incoming_rx) = mpsc::channel(INCOMING_STREAM_CAPACITY);
+ let (mesh_incoming_tx, mesh_incoming_rx) = mpsc::channel(INCOMING_STREAM_CAPACITY);
let (terminal_tx, terminal_rx) = mpsc::channel(1);
let thread = thread::Builder::new()
.name("maka-runtime-host-peer".to_owned())
.spawn(move || {
- let result = run_endpoint(options, command_rx, incoming_tx, ready_tx.clone());
+ let result = run_endpoint(
+ options,
+ command_rx,
+ incoming_tx,
+ mesh_incoming_tx,
+ ready_tx.clone(),
+ );
if let Err(error) = result {
let _ = ready_tx.send(Err(error.clone()));
let _ = terminal_tx.blocking_send(error);
@@ -248,6 +279,7 @@ pub fn start(options: StartOptions) -> Result<StartedEndpoint, PeerError> {
listen_addresses: ready.1,
commands: command_tx,
incoming: incoming_rx,
+ mesh_incoming: mesh_incoming_rx,
terminal: terminal_rx,
thread,
})
@@ -257,6 +289,7 @@ fn run_endpoint(
options: StartOptions,
commands: mpsc::Receiver<EngineCommand>,
incoming_tx: mpsc::Sender<PeerStream>,
+ mesh_incoming_tx: mpsc::Sender<PeerStream>,
ready_tx: std::sync::mpsc::SyncSender<Result<(PeerId, Vec<Multiaddr>), PeerError>>,
) -> Result<(), PeerError> {
let runtime = tokio::runtime::Builder::new_multi_thread()
@@ -264,13 +297,20 @@ fn run_endpoint(
.thread_name("maka-peer-io")
.build()
.map_err(|error| PeerError::new("peer_native_failed", error.to_string()))?;
- runtime.block_on(run_endpoint_async(options, commands, incoming_tx, ready_tx))
+ runtime.block_on(run_endpoint_async(
+ options,
+ commands,
+ incoming_tx,
+ mesh_incoming_tx,
+ ready_tx,
+ ))
}
async fn run_endpoint_async(
options: StartOptions,
mut commands: mpsc::Receiver<EngineCommand>,
incoming_tx: mpsc::Sender<PeerStream>,
+ mesh_incoming_tx: mpsc::Sender<PeerStream>,
ready_tx: std::sync::mpsc::SyncSender<Result<(PeerId, Vec<Multiaddr>), PeerError>>,
) -> Result<(), PeerError> {
let key = match options.expected_peer_id {
@@ -287,7 +327,8 @@ async fn run_endpoint_async(
None => load_or_create_key(&options.key_path).await?,
};
let local_peer_id = PeerId::from(key.public());
- let (mut swarm, stream_control, mut incoming_streams) = build_swarm(key)?;
+ let (mut swarm, stream_control, mut incoming_streams, mesh_control, mut mesh_incoming) =
+ build_swarm(key)?;
let listen_addresses = if options.listen_addresses.is_empty() {
vec![
@@ -355,8 +396,8 @@ async fn run_endpoint_async(
let _ = ready_tx.send(Ok((local_peer_id, bound_addresses)));
let (opened_tx, mut opened_rx) = mpsc::channel::<OpenedStream>(COMMAND_CAPACITY);
- let (close_connection_tx, mut close_connection_rx) =
- mpsc::channel::<ConnectionId>(MAX_ESTABLISHED_CONNECTIONS as usize);
+ let (stream_completed_tx, mut stream_completed_rx) =
+ mpsc::channel::<StreamCompletion>(MAX_ESTABLISHED_CONNECTIONS as usize);
let mut direct = DirectConnectState::default();
let mut relayed = HashMap::<PeerId, HashSet<ConnectionId>>::new();
let mut external_candidate_ready = startup_external_candidate_ready;
@@ -366,10 +407,11 @@ async fn run_endpoint_async(
loop {
tokio::select! {
command = commands.recv() => match command {
- Some(EngineCommand::Connect { options, result }) => {
+ Some(EngineCommand::Connect { options, stream_kind, result }) => {
if direct.pending.contains_key(&options.request_id)
|| direct.pending.values().any(|connect| connect.peer_id == options.peer_id)
- || direct.active.contains_key(&options.peer_id)
+ || (stream_kind == StreamKind::Application
+ && direct.active.contains_key(&options.peer_id))
{
let _ = result.send(Err(PeerError::new(
"peer_connect_in_progress",
@@ -382,6 +424,7 @@ async fn run_endpoint_async(
&mut coordination_relays,
&options,
local_peer_id,
+ stream_kind,
) {
Ok(peers) => peers,
Err(error) => {
@@ -398,6 +441,7 @@ async fn run_endpoint_async(
direct.pending.insert(request_id, PendingConnect {
peer_id: options.peer_id,
result,
+ stream_kind,
deadline: Instant::now() + options.deadline,
opening: None,
dials: HashMap::new(),
@@ -415,11 +459,13 @@ async fn run_endpoint_async(
external_candidate_ready,
Instant::now(),
);
- maybe_open_direct_stream(
+ maybe_open_peer_stream(
request_id,
&mut direct.pending,
&direct.retiring_connections,
stream_control.clone(),
+ mesh_control.clone(),
+ swarm.is_connected(&options.peer_id),
opened_tx.clone(),
);
}
@@ -458,37 +504,91 @@ async fn run_endpoint_async(
},
Some(stream) = incoming_streams.recv() => {
let peer_stream = spawn_stream(
+ stream.peer_id,
stream.stream,
- Some((stream.connection_id, close_connection_tx.clone())),
+ Some((
+ StreamCompletion::Application(stream.connection_id),
+ stream_completed_tx.clone(),
+ )),
);
if incoming_tx.try_send(peer_stream).is_err() {
// Dropping the stream closes it. A slow Host cannot create an unbounded queue.
}
}
- Some(connection_id) = close_connection_rx.recv() => {
- direct.active.retain(|_, active| *active != connection_id);
- retire_established_connection(
- &mut swarm,
- &mut direct.retiring_connections,
- connection_id,
- );
+ Some((peer_id, stream)) = mesh_incoming.next() => {
+ let peer_stream = spawn_stream(peer_id, stream, None);
+ if mesh_incoming_tx.try_send(peer_stream).is_err() {
+ // Dropping the stream applies bounded backpressure to Mesh control callers.
+ }
+ }
+ Some(completed) = stream_completed_rx.recv() => {
+ match completed {
+ StreamCompletion::Application(connection_id) => {
+ direct.active.retain(|_, active| *active != connection_id);
+ retire_established_connection(
+ &mut swarm,
+ &mut direct.retiring_connections,
+ connection_id,
+ );
+ }
+ StreamCompletion::MeshControl { dials, coordination_relay_peers } => {
+ retire_control_dials(
+ &mut swarm,
+ &mut direct.retiring_connections,
+ dials,
+ &direct.active,
+ );
+ release_coordination_relays(
+ &mut swarm,
+ &mut coordination_relays,
+ &coordination_relay_peers,
+ &direct.active,
+ );
+ }
+ }
}
Some(opened) = opened_rx.recv() => {
if let Some(waiter) = direct.pending.remove(&opened.request_id) {
let result = match opened.result {
- Ok(opened) => {
- retire_direct_dials(
- &mut swarm,
- &mut direct.retiring_connections,
- waiter.dials,
- Some(opened.connection_id),
- );
- direct.active.insert(waiter.peer_id, opened.connection_id);
- Ok(spawn_stream(
+ Ok(opened) => match waiter.stream_kind {
+ StreamKind::Application => {
+ let connection_id = opened
+ .connection_id
+ .expect("application streams have a connection id");
+ retire_direct_dials(
+ &mut swarm,
+ &mut direct.retiring_connections,
+ waiter.dials,
+ Some(connection_id),
+ );
+ direct.active.insert(waiter.peer_id, connection_id);
+ release_coordination_relays(
+ &mut swarm,
+ &mut coordination_relays,
+ &waiter.coordination_relay_peers,
+ &direct.active,
+ );
+ Ok(spawn_stream(
+ waiter.peer_id,
+ opened.stream,
+ Some((
+ StreamCompletion::Application(connection_id),
+ stream_completed_tx.clone(),
+ )),
+ ))
+ }
+ StreamKind::MeshControl => Ok(spawn_stream(
+ waiter.peer_id,
opened.stream,
- Some((opened.connection_id, close_connection_tx.clone())),
- ))
- }
+ Some((
+ StreamCompletion::MeshControl {
+ dials: waiter.dials,
+ coordination_relay_peers: waiter.coordination_relay_peers,
+ },
+ stream_completed_tx.clone(),
+ )),
+ )),
+ },
Err(message) => {
retire_direct_dials(
&mut swarm,
@@ -496,21 +596,27 @@ async fn run_endpoint_async(
waiter.dials,
None,
);
- Err(PeerError::new("direct_path_unavailable", message))
+ release_coordination_relays(
+ &mut swarm,
+ &mut coordination_relays,
+ &waiter.coordination_relay_peers,
+ &direct.active,
+ );
+ let code = match waiter.stream_kind {
+ StreamKind::Application => "direct_path_unavailable",
+ StreamKind::MeshControl => "mesh_control_unavailable",
+ };
+ Err(PeerError::new(code, message))
}
};
- release_coordination_relays(
- &mut swarm,
- &mut coordination_relays,
- &waiter.coordination_relay_peers,
- &direct.active,
- );
let _ = waiter.result.send(result);
- } else if let Ok(opened) = opened.result {
+ } else if let Ok(opened) = opened.result
+ && let Some(connection_id) = opened.connection_id
+ {
retire_established_connection(
&mut swarm,
&mut direct.retiring_connections,
- opened.connection_id,
+ connection_id,
);
}
}
@@ -531,11 +637,17 @@ async fn run_endpoint_async(
);
let requests = direct.pending.keys().copied().collect::<Vec<_>>();
for request_id in requests {
- maybe_open_direct_stream(
+ let peer_connected = direct
+ .pending
+ .get(&request_id)
+ .is_some_and(|pending| swarm.is_connected(&pending.peer_id));
+ maybe_open_peer_stream(
request_id,
&mut direct.pending,
&direct.retiring_connections,
stream_control.clone(),
+ mesh_control.clone(),
+ peer_connected,
opened_tx.clone(),
);
}
@@ -592,6 +704,8 @@ type BuiltSwarm = (
Swarm<Behaviour>,
application_stream::Control,
mpsc::Receiver<application_stream::InboundStream>,
+ libp2p_stream::Control,
+ libp2p_stream::IncomingStreams,
);
fn build_swarm(key: identity::Keypair) -> Result<BuiltSwarm, PeerError> {
@@ -599,6 +713,11 @@ fn build_swarm(key: identity::Keypair) -> Result<BuiltSwarm, PeerError> {
StreamProtocol::new(APPLICATION_PROTOCOL),
INCOMING_STREAM_CAPACITY,
);
+ let mesh_stream = libp2p_stream::Behaviour::new();
+ let mut mesh_control = mesh_stream.new_control();
+ let mesh_incoming = mesh_control
+ .accept(StreamProtocol::new(MESH_CONTROL_PROTOCOL))
+ .map_err(native_error)?;
let swarm = SwarmBuilder::with_existing_identity(key)
.with_tokio()
.with_tcp(
@@ -630,10 +749,11 @@ fn build_swarm(key: identity::Keypair) -> Result<BuiltSwarm, PeerError> {
)),
ping: ping::Behaviour::new(ping::Config::new()),
application_stream,
+ mesh_control: mesh_stream,
})
.map_err(native_error)
.map(|builder| builder.build())?;
- Ok((swarm, control, incoming))
+ Ok((swarm, control, incoming, mesh_control, mesh_incoming))
}
fn start_connect(
@@ -641,10 +761,15 @@ fn start_connect(
coordination_relays: &mut HashMap<PeerId, CoordinationRelay>,
options: &ConnectOptions,
local_peer_id: PeerId,
+ stream_kind: StreamKind,
) -> Result<StartedConnect, PeerError> {
if options.route_hints.is_empty() && options.coordination_relays.is_empty() {
+ let code = match stream_kind {
+ StreamKind::Application => "direct_path_unavailable",
+ StreamKind::MeshControl => "mesh_control_unavailable",
+ };
return Err(PeerError::new(
- "direct_path_unavailable",
+ code,
"the peer profile has no route hints or coordination relays",
));
}
@@ -691,26 +816,49 @@ fn start_connect(
})
}
-fn maybe_open_direct_stream(
+fn maybe_open_peer_stream(
request_id: u32,
pending: &mut HashMap<u32, PendingConnect>,
retiring_connections: &HashSet<ConnectionId>,
- mut control: application_stream::Control,
+ mut application_control: application_stream::Control,
+ mut mesh_control: libp2p_stream::Control,
+ peer_connected: bool,
opened_tx: mpsc::Sender<OpenedStream>,
) {
let Some(waiter) = pending.get_mut(&request_id) else {
return;
};
let peer_id = waiter.peer_id;
- if waiter.opening.is_some() || !control.has_connection(peer_id, retiring_connections) {
+ let available = match waiter.stream_kind {
+ StreamKind::Application => {
+ application_control.has_connection(peer_id, retiring_connections)
+ }
+ StreamKind::MeshControl => peer_connected,
+ };
+ if waiter.opening.is_some() || !available {
return;
}
+ let stream_kind = waiter.stream_kind;
let retiring_connections = retiring_connections.clone();
waiter.opening = Some(tokio::spawn(async move {
- let result = control
- .open_stream(peer_id, &retiring_connections)
- .await
- .map_err(|error| error.to_string());
+ let result = match stream_kind {
+ StreamKind::Application => application_control
+ .open_stream(peer_id, &retiring_connections)
+ .await
+ .map(|opened| OpenedPeerStream {
+ connection_id: Some(opened.connection_id),
+ stream: opened.stream,
+ })
+ .map_err(|error| error.to_string()),
+ StreamKind::MeshControl => mesh_control
+ .open_stream(peer_id, StreamProtocol::new(MESH_CONTROL_PROTOCOL))
+ .await
+ .map(|stream| OpenedPeerStream {
+ connection_id: None,
+ stream,
+ })
+ .map_err(|error| error.to_string()),
+ };
let _ = opened_tx.send(OpenedStream { request_id, result }).await;
}));
}
@@ -826,11 +974,9 @@ fn handle_swarm_event(
remote_peer_id,
result: Err(_),
})) => {
- if direct
- .pending
- .values()
- .any(|connect| connect.peer_id == remote_peer_id)
- && let Some(connection_ids) = relayed.remove(&remote_peer_id)
+ if direct.pending.values().any(|connect| {
+ connect.peer_id == remote_peer_id && connect.stream_kind == StreamKind::Application
+ }) && let Some(connection_ids) = relayed.remove(&remote_peer_id)
{
for connection_id in connection_ids {
let _ = swarm.close_connection(connection_id);
@@ -1119,6 +1265,21 @@ fn retire_direct_dials(
}
}
+fn retire_control_dials(
+ swarm: &mut Swarm<Behaviour>,
+ retiring: &mut HashSet<ConnectionId>,
+ dials: HashMap<ConnectionId, DialOrigin>,
+ active: &HashMap<PeerId, ConnectionId>,
+) {
+ for connection_id in dials.into_keys() {
+ if active.values().any(|active| *active == connection_id) {
+ continue;
+ }
+ retiring.insert(connection_id);
+ let _ = swarm.close_connection(connection_id);
+ }
+}
+
fn retire_established_connection(
swarm: &mut Swarm<Behaviour>,
retiring: &mut HashSet<ConnectionId>,
diff --git a/native/runtime-host-peer/src/engine/application_stream.rs b/native/runtime-host-peer/src/engine/application_stream.rs
index 7ed44f0ff..3474736e2 100644
--- a/native/runtime-host-peer/src/engine/application_stream.rs
+++ b/native/runtime-host-peer/src/engine/application_stream.rs
@@ -57,6 +57,7 @@ pub(super) struct Behaviour {
}
pub(super) struct InboundStream {
+ pub(super) peer_id: PeerId,
pub(super) connection_id: ConnectionId,
pub(super) stream: Stream,
}
@@ -87,6 +88,7 @@ impl Behaviour {
let (sender, receiver) = mpsc::channel(OUTBOUND_COMMAND_CAPACITY);
lock(&self.shared).insert(connection_id, peer_id, sender);
Handler::direct(
+ peer_id,
connection_id,
self.protocol.clone(),
self.incoming.clone(),
@@ -257,6 +259,7 @@ fn lock(shared: &Arc<Mutex<DirectConnections>>) -> MutexGuard<'_, DirectConnecti
}
pub(super) struct Handler {
+ peer_id: Option<PeerId>,
connection_id: Option<ConnectionId>,
protocol: Option<StreamProtocol>,
incoming: Option<mpsc::Sender<InboundStream>>,
@@ -268,12 +271,14 @@ pub(super) struct Handler {
impl Handler {
fn direct(
+ peer_id: PeerId,
connection_id: ConnectionId,
protocol: StreamProtocol,
incoming: mpsc::Sender<InboundStream>,
commands: mpsc::Receiver<NewStream>,
) -> Self {
Self {
+ peer_id: Some(peer_id),
connection_id: Some(connection_id),
protocol: Some(protocol),
incoming: Some(incoming),
@@ -286,6 +291,7 @@ impl Handler {
fn relayed() -> Self {
Self {
+ peer_id: None,
connection_id: None,
protocol: None,
incoming: None,
@@ -373,6 +379,7 @@ impl ConnectionHandler for Handler {
if self.incoming.as_ref().is_none_or(|incoming| {
incoming
.try_send(InboundStream {
+ peer_id: self.peer_id.expect("direct handlers have a peer id"),
connection_id: self
.connection_id
.expect("direct handlers have a connection id"),
diff --git a/native/runtime-host-peer/src/engine/peer_stream.rs b/native/runtime-host-peer/src/engine/peer_stream.rs
index 6cd055648..4a99dd92e 100644
--- a/native/runtime-host-peer/src/engine/peer_stream.rs
+++ b/native/runtime-host-peer/src/engine/peer_stream.rs
@@ -18,15 +18,16 @@
*/
use futures::{AsyncReadExt as _, AsyncWriteExt as _};
-use libp2p::swarm::ConnectionId;
+use libp2p::PeerId;
use tokio::sync::{mpsc, oneshot};
-use super::PeerError;
+use super::{PeerError, StreamCompletion};
const QUEUE_CAPACITY: usize = 64;
const CHUNK_BYTES: usize = 64 * 1024;
pub struct PeerStream {
+ pub peer_id: PeerId,
pub incoming: mpsc::Receiver<Result<Vec<u8>, PeerError>>,
pub commands: mpsc::Sender<StreamCommand>,
}
@@ -43,8 +44,9 @@ pub enum StreamCommand {
}
pub(super) fn spawn_stream(
+ peer_id: PeerId,
stream: libp2p::swarm::Stream,
- close_connection: Option<(ConnectionId, mpsc::Sender<ConnectionId>)>,
+ completion: Option<(StreamCompletion, mpsc::Sender<StreamCompletion>)>,
) -> PeerStream {
let (incoming_tx, incoming_rx) = mpsc::channel(QUEUE_CAPACITY);
let (command_tx, mut command_rx) = mpsc::channel(QUEUE_CAPACITY);
@@ -86,11 +88,12 @@ pub(super) fn spawn_stream(
}
}
}
- if let Some((connection_id, close)) = close_connection {
- let _ = close.send(connection_id).await;
+ if let Some((completion, completed)) = completion {
+ let _ = completed.send(completion).await;
}
});
PeerStream {
+ peer_id,
incoming: incoming_rx,
commands: command_tx,
}
diff --git a/packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv b/packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv
index b5b200eea..71e73b070 100644
--- a/packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv
+++ b/packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv
@@ -105,6 +105,7 @@ libp2p-noise@0.46.1 X
libp2p-ping@0.47.0 X
libp2p-quic@0.13.1 X
libp2p-relay@0.21.1 X
+libp2p-stream@0.4.0-alpha X
libp2p-swarm@0.47.1 X
libp2p-swarm-derive@0.35.1 X
libp2p-tcp@0.44.1 X
diff --git a/packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt b/packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt
index b6cab8d19..f79330a82 100644
--- a/packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt
+++ b/packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt
@@ -5,8 +5,8 @@ Generated by scripts/generate-runtime-host-peer-notices.mjs from the exact
four-target production dependency inventory. Do not edit this file by hand.
Manifest: native/runtime-host-peer/Cargo.toml
-Cargo.lock SHA-256: 07524c9f4eac723344e6fd8746a2f7cfed692feaed1785ecbfd45e0ecbe20981
-Inventory SHA-256: eb4b2941cf0ed87153aff3ac5d32697dde3ba105e388ff96b9584c65882834f1
+Cargo.lock SHA-256: a8e746b4231e3011af210e7b5b3dd5cb48353926f4ba52d769346646aceeab37
+Inventory SHA-256: ab88060b713dfda096d760555f31b561f7b5371d8acd1cedd56fc8dd64917514
Packages
--------
@@ -713,6 +713,12 @@ SPDX license: MIT
Source: https://github.com/libp2p/rust-libp2p
License text: 9f8c060ac8678c29279330e8d89ab93030734c72f99801af50e32872b5c29fbe (MIT.txt)
+libp2p-stream@0.4.0-alpha
+-------------------------
+SPDX license: MIT
+Source: https://github.com/libp2p/rust-libp2p
+License text: 80f847ab8b3d1802dcf33aad57b13d3e5cbae4969fb5990a5ff24ac7ae98ddb6 (MIT.txt)
+
libp2p-swarm@0.47.1
-------------------
SPDX license: MIT
@@ -8606,6 +8612,7 @@ Used by:
- libp2p-allow-block-list@0.6.0 (MIT.txt)
- libp2p-connection-limits@0.6.0 (MIT.txt)
- libp2p-identity@0.2.14 (MIT.txt)
+- libp2p-stream@0.4.0-alpha (MIT.txt)
- libp2p-tls@0.6.2 (MIT.txt)
Copyright notice: not included in the published crate
diff --git a/packages/cli/src/cli-core.ts b/packages/cli/src/cli-core.ts
index 548edbcc7..cf33710af 100644
--- a/packages/cli/src/cli-core.ts
+++ b/packages/cli/src/cli-core.ts
@@ -141,6 +141,7 @@ function helpText(cliCommand: string): string {
` ${cliCommand} runtime-host service status|start|stop|restart|logs [--json]`,
` ${cliCommand} runtime-host service uninstall --expected-service-id <id> --expected-root-path <path> --expected-root-id <id> [--allow-interrupt-active-tasks]`,
` ${cliCommand} runtime-host service peer enable|disable|status|rotate|descriptor [options]`,
+ ` ${cliCommand} runtime-host mesh create|status|invite|join|remove|close|serve [options]`,
` ${cliCommand} runtime-host service retire --expected-service-id <id> --expected-root-path <path> --expected-root-id <id> [--allow-interrupt-active-tasks]`,
` ${cliCommand} runtime-host service check-update --target <latest|next|version> [--json]`,
` ${cliCommand} runtime-host service update [--target <latest|next|version>] --expected-service-id <id> --expected-root-path <path> --expected-root-id <id> [--allow-interrupt-active-tasks]`,
@@ -233,7 +234,10 @@ export async function runMakaCli(
const dataRoots = resolveMakaDataRoots({ profileName: options.dataProfileName });
await configureRuntimeHostPeerClient({
cliPath: process.argv[1] ?? '',
- clientDataRoot: dataRoots.clientDataRoot,
+ clientDataRoot:
+ command.kind === 'runtime-host-peer-mesh' && command.clientDataRoot
+ ? command.clientDataRoot
+ : dataRoots.clientDataRoot,
});
switch (command.kind) {
case 'run': {
@@ -366,6 +370,13 @@ export async function runMakaCli(
...(command.allowInterruptActiveTasks ? { allowInterruptActiveTasks: true } : {}),
});
}
+ case 'runtime-host-peer-mesh': {
+ const { runRuntimeHostPeerMeshCli } = await import('./runtime-host-peer-mesh-command.js');
+ return runRuntimeHostPeerMeshCli({
+ ...command,
+ clientDataRoot: command.clientDataRoot ?? dataRoots.clientDataRoot,
+ });
+ }
case 'runtime-host-service-update': {
const { runManagedRuntimeHostSelectedUpdateCli, runManagedRuntimeHostUpdateCli } =
await import('./runtime-host-update-command.js');
diff --git a/packages/cli/src/runtime-host-cli.ts b/packages/cli/src/runtime-host-cli.ts
index c69e71f7a..bc0b7eb4d 100644
--- a/packages/cli/src/runtime-host-cli.ts
+++ b/packages/cli/src/runtime-host-cli.ts
@@ -112,6 +112,18 @@ export type RuntimeHostCliCommand =
expectedTarget?: RuntimeHostManagedServiceTarget;
allowInterruptActiveTasks?: true;
}
+ | {
+ kind: 'runtime-host-peer-mesh';
+ action: 'create' | 'status' | 'invite' | 'join' | 'remove' | 'close' | 'serve';
+ json: boolean;
+ clientDataRoot?: string;
+ listenAddresses: string[];
+ coordinationRelays: string[];
+ meshId?: string;
+ peerId?: string;
+ invitation?: string;
+ ttlMs?: number;
+ }
| {
kind: 'runtime-host-service-check-update';
json: boolean;
@@ -236,13 +248,115 @@ export function parseRuntimeHostCommand(argv: string[]): RuntimeHostCliCommand {
return parseCapabilityProviderCommand(argv.slice(1));
}
if (argv[0] === 'profile') return parseProfileCommand(argv.slice(1));
+ if (argv[0] === 'mesh') return parsePeerMeshCommand(argv.slice(1));
return error(
argv[0]
? `Unexpected runtime-host command: ${argv[0]}`
- : 'runtime-host requires the serve, setup, service, access, project, profile, or capability-provider command',
+ : 'runtime-host requires the serve, setup, service, access, project, profile, mesh, or capability-provider command',
);
}
+function parsePeerMeshCommand(argv: string[]): RuntimeHostCliCommand {
+ const action = argv[0];
+ if (
+ action !== 'create' &&
+ action !== 'status' &&
+ action !== 'invite' &&
+ action !== 'join' &&
+ action !== 'remove' &&
+ action !== 'close' &&
+ action !== 'serve'
+ ) {
+ return error(
+ action
+ ? `Unexpected runtime-host mesh command: ${action}`
+ : 'runtime-host mesh requires create, status, invite, join, remove, close, or serve',
+ );
+ }
+ let json = false;
+ let clientDataRoot: string | undefined;
+ let meshId: string | undefined;
+ let peerId: string | undefined;
+ let invitation: string | undefined;
+ let ttlMs: number | undefined;
+ const listenAddresses: string[] = [];
+ const coordinationRelays: string[] = [];
+ for (let index = 1; index < argv.length; index += 1) {
+ const argument = argv[index];
+ if (argument === '--json') {
+ if (json) return error('Duplicate --json');
+ json = true;
+ continue;
+ }
+ if (
+ argument !== '--client-data-root' &&
+ argument !== '--mesh' &&
+ argument !== '--peer' &&
+ argument !== '--invitation' &&
+ argument !== '--ttl-seconds' &&
+ argument !== '--listen' &&
+ argument !== '--coordination-relay'
+ ) {
+ return error(`Unexpected argument: ${argument ?? ''}`);
+ }
+ const value = optionValue(argv, index, argument);
+ if (typeof value !== 'string') return value;
+ if (argument === '--client-data-root') {
+ if (clientDataRoot) return error('Duplicate --client-data-root');
+ if (!isSafeAbsolutePath(value)) return error('--client-data-root must be an absolute path');
+ clientDataRoot = value;
+ }
+ if (argument === '--mesh') {
+ if (meshId) return error('Duplicate --mesh');
+ meshId = value;
+ }
+ if (argument === '--peer') {
+ if (peerId) return error('Duplicate --peer');
+ peerId = value;
+ }
+ if (argument === '--invitation') {
+ if (invitation) return error('Duplicate --invitation');
+ invitation = value;
+ }
+ if (argument === '--ttl-seconds') {
+ if (ttlMs !== undefined) return error('Duplicate --ttl-seconds');
+ ttlMs = Number(value) * 1_000;
+ }
+ if (argument === '--listen') listenAddresses.push(value);
+ if (argument === '--coordination-relay') coordinationRelays.push(value);
+ index += 1;
+ }
+ if (ttlMs !== undefined && (!Number.isSafeInteger(ttlMs) || ttlMs < 1_000)) {
+ return error('--ttl-seconds must be a positive whole number');
+ }
+ if ((action === 'invite' || action === 'remove' || action === 'close') && !meshId) {
+ return error(`runtime-host mesh ${action} requires --mesh`);
+ }
+ if (action === 'join' && !invitation) {
+ return error('runtime-host mesh join requires --invitation');
+ }
+ if (action === 'remove' && !peerId) {
+ return error('runtime-host mesh remove requires --peer');
+ }
+ if (ttlMs !== undefined && action !== 'invite') {
+ return error('--ttl-seconds is only valid with mesh invite');
+ }
+ if (invitation && action !== 'join') return error('--invitation is only valid with mesh join');
+ if (peerId && action !== 'remove') return error('--peer is only valid with mesh remove');
+ return {
+ kind: 'runtime-host-peer-mesh',
+ action,
+ json,
+ listenAddresses,
+ coordinationRelays,
+ ...(clientDataRoot ? { clientDataRoot } : {}),
+ ...(meshId ? { meshId } : {}),
+ ...(peerId ? { peerId } : {}),
+ ...(invitation ? { invitation } : {}),
+ ...(ttlMs === undefined ? {} : { ttlMs }),
+ };
+}
+
function parseSetupCommand(argv: string[]): RuntimeHostCliCommand {
let principalId: string | undefined;
let preset: 'desktop-client' | 'terminal-client' | undefined;
diff --git a/packages/cli/src/runtime-host-peer-mesh-command.ts b/packages/cli/src/runtime-host-peer-mesh-command.ts
new file mode 100644
index 000000000..0d63ac465
--- /dev/null
+++ b/packages/cli/src/runtime-host-peer-mesh-command.ts
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { join } from 'node:path';
+import { createRuntimeHostPeerClientFromEnvironment } from '@maka/runtime-host/client';
+import {
+ decodePeerMeshInvitation,
+ openPeerMeshNode,
+ type PeerMeshInvitationV1,
+ type PeerMeshStatus,
+} from '@maka/runtime-host/peer-mesh';
+
+const INVITATION_PREFIX = 'maka-mesh-v1.';
+
+export interface RuntimeHostPeerMeshCliOptions {
+ readonly action: 'create' | 'status' | 'invite' | 'join' | 'remove' | 'close' | 'serve';
+ readonly json: boolean;
+ readonly clientDataRoot: string;
+ readonly listenAddresses: readonly string[];
+ readonly coordinationRelays: readonly string[];
+ readonly meshId?: string;
+ readonly peerId?: string;
+ readonly invitation?: string;
+ readonly ttlMs?: number;
+}
+
+export async function runRuntimeHostPeerMeshCli(
+ options: RuntimeHostPeerMeshCliOptions,
+): Promise<number> {
+ const peer = createRuntimeHostPeerClientFromEnvironment(process.env, {
+ listenAddresses: options.listenAddresses,
+ coordinationRelays: options.coordinationRelays,
+ });
+ const node = await openPeerMeshNode({
+ dataRoot: join(options.clientDataRoot, 'peer-mesh'),
+ peer,
+ coordinationRelays: options.coordinationRelays,
+ });
+ const shutdown = () => void peer.close();
+ if (options.action === 'serve') {
+ process.once('SIGINT', shutdown);
+ process.once('SIGTERM', shutdown);
+ }
+ try {
+ switch (options.action) {
+ case 'create':
+ writeResult(options, await node.create());
+ break;
+ case 'status':
+ writeResult(options, node.status());
+ break;
+ case 'invite': {
+ const invitation = await node.invite(options.meshId!, {
+ ...(options.ttlMs === undefined ? {} : { ttlMs: options.ttlMs }),
+ });
+ writeResult(options, {
+ meshId: invitation.meshId,
+ expiresAt: invitation.expiresAt,
+ invitation: encodeInvitation(invitation),
+ });
+ break;
+ }
+ case 'join':
+ writeResult(options, await node.join(decodeInvitation(options.invitation!)));
+ break;
+ case 'remove':
+ writeResult(options, await node.remove(options.meshId!, options.peerId!));
+ break;
+ case 'close':
+ writeResult(options, await node.closeMesh(options.meshId!));
+ break;
+ case 'serve':
+ writeResult(options, { serving: true, meshes: node.status() });
+ await node.serve();
+ break;
+ }
+ return 0;
+ } catch (error) {
+ process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
+ return 1;
+ } finally {
+ process.off('SIGINT', shutdown);
+ process.off('SIGTERM', shutdown);
+ await node.close();
+ await peer.close();
+ }
+}
+
+function encodeInvitation(invitation: PeerMeshInvitationV1): string {
+ return `${INVITATION_PREFIX}${Buffer.from(JSON.stringify(invitation)).toString('base64url')}`;
+}
+
+function decodeInvitation(value: string): PeerMeshInvitationV1 {
+ if (!value.startsWith(INVITATION_PREFIX)) throw new Error('Invalid Peer Mesh invitation code');
+ try {
+ return decodePeerMeshInvitation(
+ JSON.parse(
+ Buffer.from(value.slice(INVITATION_PREFIX.length), 'base64url').toString('utf8'),
+ ) as unknown,
+ );
+ } catch (error) {
+ throw new Error('Invalid Peer Mesh invitation code', { cause: error });
+ }
+}
+
+function writeResult(
+ options: RuntimeHostPeerMeshCliOptions,
+ value: PeerMeshStatus | readonly PeerMeshStatus[] | Readonly<Record<string, unknown>>,
+): void {
+ if (options.json) {
+ process.stdout.write(`${JSON.stringify({ schemaVersion: 1, ok: true, result: value })}\n`);
+ return;
+ }
+ process.stdout.write(`${JSON.stringify(value, null, 2)}\n`);
+}
diff --git a/packages/runtime-host/package.json b/packages/runtime-host/package.json
index d7591a76c..32d639dee 100644
--- a/packages/runtime-host/package.json
+++ b/packages/runtime-host/package.json
@@ -9,6 +9,7 @@
"./adapter": "./dist/adapter/index.js",
"./protocol": "./dist/protocol/index.js",
"./client": "./dist/client/index.js",
+ "./peer-mesh": "./dist/peer-mesh/index.js",
"./operator": "./dist/operator/index.js",
"./execution-candidate-main": "./dist/execution-candidate-main.js",
"./server": "./dist/server/index.js",
diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts
index 72db11529..024d6225d 100644
--- a/packages/runtime-host/src/__tests__/peer-listener.test.ts
+++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts
@@ -75,6 +75,7 @@ test('rechecks peer authority at admission after the authentication response is
let aborted = false;
let reads = 0;
const stream: RuntimeHostPeerNativeStream = {
+ peerId: 'remote-peer',
read: async () => (reads++ === 0 ? Buffer.from('{"v":1,"credential":"revoked"}\n') : null),
write: async () => writeReleased,
close: async () => undefined,
@@ -112,8 +113,12 @@ function endpointWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerNa
connect: async () => {
throw new Error('not used');
},
+ connectMeshControl: async () => {
+ throw new Error('not used');
+ },
cancelConnect: async () => false,
accept: async () => streams.shift() ?? null,
+ acceptMeshControl: async () => null,
close: async () => undefined,
};
}
@@ -125,6 +130,7 @@ function pendingStream(): RuntimeHostPeerNativeStream & { readonly aborted: bool
});
let aborted = false;
return {
+ peerId: 'remote-peer',
get aborted() {
return aborted;
},
@@ -146,6 +152,7 @@ function recordingStream(initial: Buffer): RuntimeHostPeerNativeStream & {
let closed = false;
const writes: Buffer[] = [];
return {
+ peerId: 'remote-peer',
get writes() {
return writes;
},
diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts
new file mode 100644
index 000000000..f51d93dce
--- /dev/null
+++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { test } from 'node:test';
+import type { RuntimeHostPeerClient, RuntimeHostPeerConnectInput } from '../client/peer-client.js';
+import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js';
+import {
+ decodeSignedPeerMeshRoster,
+ generatePeerMeshAuthorityKeyPair,
+ peerMeshId,
+ signPeerMeshRoster,
+} from '../peer-mesh/model.js';
+import { openPeerMeshNode, type PeerMeshNode } from '../peer-mesh/node.js';
+
+test('authenticates three peers, consumes invitations once, and keeps authority state private', async () => {
+ const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-'));
+ const network = new MemoryPeerNetwork();
+ const peers = ['peer-a', 'peer-b', 'peer-c'].map((peerId) => network.create(peerId));
+ const nodes: PeerMeshNode[] = [];
+ try {
+ for (const [index, peer] of peers.entries()) {
+ nodes.push(await openPeerMeshNode({ dataRoot: join(root, String(index)), peer }));
+ }
+ const [authority, memberB, memberC] = nodes as [PeerMeshNode, PeerMeshNode, PeerMeshNode];
+ const mesh = await authority.create();
+ const serving = authority.serve();
+
+ const contested = await authority.invite(mesh.roster.roster.meshId);
+ const attempts = await Promise.allSettled([memberB.join(contested), memberC.join(contested)]);
+ assert.equal(attempts.filter(({ status }) => status === 'fulfilled').length, 1);
+ assert.equal(attempts.filter(({ status }) => status === 'rejected').length, 1);
+
+ const loser = attempts[0]?.status === 'rejected' ? memberB : memberC;
+ await loser.join(await authority.invite(mesh.roster.roster.meshId));
+ const current = authority.status()[0];
+ assert.deepEqual(current?.roster.roster.members, ['peer-a', 'peer-b', 'peer-c']);
+ assert.equal('authorityPrivateKey' in (current ?? {}), false);
+
+ await authority.remove(mesh.roster.roster.meshId, 'peer-b');
+ assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a', 'peer-c']);
+
+ await peers[0]!.close();
+ await serving;
+ } finally {
+ await Promise.allSettled(nodes.map((node) => node.close()));
+ await Promise.allSettled(peers.map((peer) => peer.close()));
+ await rm(root, { recursive: true, force: true });
+ }
+});
+
+test('rejects a modified authority-signed roster', () => {
+ const keys = generatePeerMeshAuthorityKeyPair();
+ const signed = signPeerMeshRoster(
+ {
+ version: 1,
+ meshId: peerMeshId(keys.publicKey),
+ revision: 1,
+ members: ['peer-a'],
+ closed: false,
+ },
+ keys,
+ );
+ assert.throws(() =>
+ decodeSignedPeerMeshRoster({
+ ...signed,
+ roster: { ...signed.roster, members: ['peer-b'] },
+ }),
+ );
+});
+
+class MemoryPeerNetwork {
+ readonly #peers = new Map<string, MemoryPeerClient>();
+
+ create(peerId: string): MemoryPeerClient {
+ const peer = new MemoryPeerClient(peerId, this.#peers);
+ this.#peers.set(peerId, peer);
+ return peer;
+ }
+}
+
+class MemoryPeerClient implements RuntimeHostPeerClient {
+ readonly #incoming: Array<RuntimeHostPeerNativeStream | null> = [];
+ readonly #waiters: Array<(stream: RuntimeHostPeerNativeStream | null) => void> = [];
+ #closed = false;
+
+ constructor(
+ private readonly peerId: string,
+ private readonly peers: ReadonlyMap<string, MemoryPeerClient>,
+ ) {}
+
+ identity() {
+ return { peerId: this.peerId, listenAddresses: [`/memory/${this.peerId}`] } as const;
+ }
+
+ connect(): Promise<RuntimeHostPeerNativeStream> {
+ throw new Error('Application streams are not used by this test');
+ }
+
+ async connectMeshControl(
+ input: RuntimeHostPeerConnectInput,
+ ): Promise<RuntimeHostPeerNativeStream> {
+ const remote = this.peers.get(input.peerId);
+ if (!remote) throw new Error('Peer is unavailable');
+ const [localStream, remoteStream] = memoryStreamPair(this.peerId, input.peerId);
+ remote.accept(remoteStream);
+ return localStream;
+ }
+
+ acceptMeshControl(): Promise<RuntimeHostPeerNativeStream | null> {
+ const stream = this.#incoming.shift();
+ if (stream !== undefined) return Promise.resolve(stream);
+ return new Promise((resolve) => this.#waiters.push(resolve));
+ }
+
+ close(): Promise<void> {
+ if (this.#closed) return Promise.resolve();
+ this.#closed = true;
+ this.accept(null);
+ return Promise.resolve();
+ }
+
+ accept(stream: RuntimeHostPeerNativeStream | null): void {
+ const waiter = this.#waiters.shift();
+ if (waiter) waiter(stream);
+ else this.#incoming.push(stream);
+ }
+}
+
+function memoryStreamPair(
+ localPeerId: string,
+ remotePeerId: string,
+): [RuntimeHostPeerNativeStream, RuntimeHostPeerNativeStream] {
+ const local = new MemoryStream(remotePeerId);
+ const remote = new MemoryStream(localPeerId);
+ local.connect(remote);
+ remote.connect(local);
+ return [local, remote];
+}
+
+class MemoryStream implements RuntimeHostPeerNativeStream {
+ readonly #incoming: Array<Buffer | null> = [];
+ readonly #waiters: Array<(chunk: Buffer | null) => void> = [];
+ #remote: MemoryStream | undefined;
+ #closed = false;
+
+ constructor(readonly peerId: string) {}
+
+ connect(remote: MemoryStream): void {
+ this.#remote = remote;
+ }
+
+ read(): Promise<Buffer | null> {
+ const chunk = this.#incoming.shift();
+ if (chunk !== undefined) return Promise.resolve(chunk);
+ return new Promise((resolve) => this.#waiters.push(resolve));
+ }
+
+ async write(bytes: Buffer): Promise<void> {
+ if (this.#closed || !this.#remote) throw new Error('Stream is closed');
+ this.#remote.push(Buffer.from(bytes));
+ }
+
+ async close(): Promise<void> {
+ if (this.#closed) return;
+ this.#closed = true;
+ this.#remote?.push(null);
+ }
+
+ abort(): void {
+ void this.close();
+ }
+
+ push(chunk: Buffer | null): void {
+ const waiter = this.#waiters.shift();
+ if (waiter) waiter(chunk);
+ else this.#incoming.push(chunk);
+ }
+}
diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts
index 423567335..e6470a2dd 100644
--- a/packages/runtime-host/src/__tests__/peer-native.test.ts
+++ b/packages/runtime-host/src/__tests__/peer-native.test.ts
@@ -156,6 +156,7 @@ test('bounds and separates the peer credential preface from Runtime Host frames'
function streamWith(chunk: Buffer): RuntimeHostPeerNativeStream {
let pending: Buffer | null = chunk;
return {
+ peerId: 'remote-peer',
read: async () => {
const value = pending;
pending = null;
diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts
index df6707e02..e750146a7 100644
--- a/packages/runtime-host/src/client/peer-client.ts
+++ b/packages/runtime-host/src/client/peer-client.ts
@@ -33,15 +33,25 @@ export interface RuntimeHostPeerConnectInput {
}
export interface RuntimeHostPeerClient {
+ identity(): Readonly<{ peerId: string; listenAddresses: readonly string[] }>;
connect(
input: RuntimeHostPeerConnectInput,
signal?: AbortSignal,
): Promise<RuntimeHostPeerNativeStream>;
+ connectMeshControl(
+ input: RuntimeHostPeerConnectInput,
+ signal?: AbortSignal,
+ ): Promise<RuntimeHostPeerNativeStream>;
+ acceptMeshControl(): Promise<RuntimeHostPeerNativeStream | null>;
close(): Promise<void>;
}
export function createRuntimeHostPeerClientFromEnvironment(
environment: NodeJS.ProcessEnv = process.env,
+ options: {
+ readonly listenAddresses?: readonly string[];
+ readonly coordinationRelays?: readonly string[];
+ } = {},
): RuntimeHostPeerClient {
const nativePath = environment.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH;
const keyPath = environment.MAKA_RUNTIME_HOST_PEER_KEY_PATH;
@@ -51,12 +61,14 @@ export function createRuntimeHostPeerClientFromEnvironment(
'Experimental direct peer requires MAKA_RUNTIME_HOST_PEER_NATIVE_PATH and MAKA_RUNTIME_HOST_PEER_KEY_PATH',
);
}
- return createRuntimeHostPeerClient({ nativePath, keyPath });
+ return createRuntimeHostPeerClient({ nativePath, keyPath, ...options });
}
export function createRuntimeHostPeerClient(input: {
readonly nativePath: string;
readonly keyPath: string;
+ readonly listenAddresses?: readonly string[];
+ readonly coordinationRelays?: readonly string[];
}): RuntimeHostPeerClient {
return new RuntimeHostPeerClientImpl(input);
}
@@ -64,6 +76,8 @@ export function createRuntimeHostPeerClient(input: {
class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient {
readonly #nativePath: string;
readonly #keyPath: string;
+ readonly #listenAddresses: readonly string[] | undefined;
+ readonly #coordinationRelays: readonly string[] | undefined;
#endpoint: RuntimeHostPeerNativeEndpoint | undefined;
#draining: Promise<void> | undefined;
#terminalError: Error | undefined;
@@ -71,19 +85,56 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient {
#closed = false;
#closeTask: Promise<void> | undefined;
- constructor(input: { readonly nativePath: string; readonly keyPath: string }) {
+ constructor(input: {
+ readonly nativePath: string;
+ readonly keyPath: string;
+ readonly listenAddresses?: readonly string[];
+ readonly coordinationRelays?: readonly string[];
+ }) {
this.#nativePath = input.nativePath;
this.#keyPath = input.keyPath;
+ this.#listenAddresses = input.listenAddresses;
+ this.#coordinationRelays = input.coordinationRelays;
+ }
+
+ identity(): Readonly<{ peerId: string; listenAddresses: readonly string[] }> {
+ const endpoint = this.#requireEndpoint();
+ return Object.freeze({
+ peerId: endpoint.peerId,
+ listenAddresses: Object.freeze([...endpoint.listenAddresses]),
+ });
}
async connect(
input: RuntimeHostPeerConnectInput,
signal?: AbortSignal,
+ ): Promise<RuntimeHostPeerNativeStream> {
+ return this.#connect(input, signal, 'application');
+ }
+
+ async connectMeshControl(
+ input: RuntimeHostPeerConnectInput,
+ signal?: AbortSignal,
+ ): Promise<RuntimeHostPeerNativeStream> {
+ return this.#connect(input, signal, 'mesh-control');
+ }
+
+ acceptMeshControl(): Promise<RuntimeHostPeerNativeStream | null> {
+ return this.#requireEndpoint().acceptMeshControl();
+ }
+
+ async #connect(
+ input: RuntimeHostPeerConnectInput,
+ signal: AbortSignal | undefined,
+ kind: 'application' | 'mesh-control',
): Promise<RuntimeHostPeerNativeStream> {
signal?.throwIfAborted();
const endpoint = this.#requireEndpoint();
const requestId = this.#allocateRequestId();
- const connection = endpoint.connect({ ...input, requestId });
+ const connection = endpoint[kind === 'application' ? 'connect' : 'connectMeshControl']({
+ ...input,
+ requestId,
+ });
let settled = false;
const cancel = () => {
void cancelPeerConnect(endpoint, requestId, () => settled);
@@ -125,6 +176,8 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient {
const endpoint = startRuntimeHostPeerEndpoint({
nativePath: this.#nativePath,
keyPath: this.#keyPath,
+ ...(this.#listenAddresses ? { listenAddresses: this.#listenAddresses } : {}),
+ ...(this.#coordinationRelays ? { coordinationRelays: this.#coordinationRelays } : {}),
});
this.#endpoint = endpoint;
this.#draining = this.#drainInbound(endpoint);
diff --git a/packages/runtime-host/src/peer-mesh/index.ts b/packages/runtime-host/src/peer-mesh/index.ts
new file mode 100644
index 000000000..74e762e9c
--- /dev/null
+++ b/packages/runtime-host/src/peer-mesh/index.ts
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+export {
+ decodePeerMeshInvitation,
+ type PeerMeshAuthorityTarget,
+ type PeerMeshInvitationV1,
+ type PeerMeshRosterV1,
+ type SignedPeerMeshRosterV1,
+} from './model.js';
+export {
+ openPeerMeshNode,
+ type PeerMeshNode,
+ type PeerMeshStatus,
+} from './node.js';
diff --git a/packages/runtime-host/src/peer-mesh/model.ts b/packages/runtime-host/src/peer-mesh/model.ts
new file mode 100644
index 000000000..e514e212c
--- /dev/null
+++ b/packages/runtime-host/src/peer-mesh/model.ts
@@ -0,0 +1,355 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import {
+ createHash,
+ createPrivateKey,
+ createPublicKey,
+ generateKeyPairSync,
+ randomBytes,
+ randomUUID,
+ sign,
+ timingSafeEqual,
+ verify,
+} from 'node:crypto';
+
+export const PEER_MESH_MAX_MEMBERS = 64;
+export const PEER_MESH_MAX_MESHES = 16;
+export const PEER_MESH_MAX_PENDING_INVITATIONS = 32;
+export const PEER_MESH_MAX_ROUTE_HINTS = 16;
+
+export interface PeerMeshRosterV1 {
+ readonly version: 1;
+ readonly meshId: string;
+ readonly revision: number;
+ readonly members: readonly string[];
+ readonly closed: boolean;
+}
+
+export interface SignedPeerMeshRosterV1 {
+ readonly roster: PeerMeshRosterV1;
+ readonly authorityPublicKey: string;
+ readonly signature: string;
+}
+
+export interface PeerMeshAuthorityTarget {
+ readonly peerId: string;
+ readonly routeHints: readonly string[];
+ readonly coordinationRelays: readonly string[];
+}
+
+export interface PeerMeshInvitationV1 extends PeerMeshAuthorityTarget {
+ readonly version: 1;
+ readonly meshId: string;
+ readonly authorityPublicKey: string;
+ readonly invitationId: string;
+ readonly secret: string;
+ readonly expiresAt: number;
+}
+
+export interface PeerMeshAuthorityKeyPair {
+ readonly publicKey: string;
+ readonly privateKey: string;
+}
+
+export function generatePeerMeshAuthorityKeyPair(): PeerMeshAuthorityKeyPair {
+ const { publicKey, privateKey } = generateKeyPairSync('ed25519');
+ return Object.freeze({
+ publicKey: publicKey.export({ format: 'der', type: 'spki' }).toString('base64url'),
+ privateKey: privateKey.export({ format: 'der', type: 'pkcs8' }).toString('base64url'),
+ });
+}
+
+export function peerMeshId(authorityPublicKey: string): string {
+ decodePublicKey(authorityPublicKey);
+ return `mesh_${createHash('sha256').update(authorityPublicKey).digest('base64url')}`;
+}
+
+export function signPeerMeshRoster(
+ roster: PeerMeshRosterV1,
+ keys: PeerMeshAuthorityKeyPair,
+): SignedPeerMeshRosterV1 {
+ const canonical = canonicalPeerMeshRoster(roster);
+ if (canonical.meshId !== peerMeshId(keys.publicKey)) {
+ throw new Error('Peer Mesh roster does not belong to the authority key');
+ }
+ const privateKey = createPrivateKey({
+ key: Buffer.from(keys.privateKey, 'base64url'),
+ format: 'der',
+ type: 'pkcs8',
+ });
+ if (privateKey.asymmetricKeyType !== 'ed25519') {
+ throw new Error('Peer Mesh authority key must be Ed25519');
+ }
+ return Object.freeze({
+ roster: canonical,
+ authorityPublicKey: keys.publicKey,
+ signature: sign(null, encodeRoster(canonical), privateKey).toString('base64url'),
+ });
+}
+
+export function validatePeerMeshAuthorityKeyPair(keys: PeerMeshAuthorityKeyPair): void {
+ let privateKey;
+ try {
+ privateKey = createPrivateKey({
+ key: decodeCanonicalBase64Url(keys.privateKey, 'authority private key'),
+ format: 'der',
+ type: 'pkcs8',
+ });
+ } catch (error) {
+ throw new Error('Invalid Peer Mesh authority private key', { cause: error });
+ }
+ if (privateKey.asymmetricKeyType !== 'ed25519') {
+ throw new Error('Peer Mesh authority key must be Ed25519');
+ }
+ const derived = createPublicKey(privateKey).export({ format: 'der', type: 'spki' });
+ if (derived.toString('base64url') !== keys.publicKey) {
+ throw new Error('Peer Mesh authority private key does not match its public key');
+ }
+}
+
+export function decodeSignedPeerMeshRoster(value: unknown): SignedPeerMeshRosterV1 {
+ const record = exactObject(value, 'signed Peer Mesh roster', [
+ 'roster',
+ 'authorityPublicKey',
+ 'signature',
+ ]);
+ const authorityPublicKey = string(record.authorityPublicKey, 'authorityPublicKey', 256);
+ const roster = canonicalPeerMeshRoster(record.roster);
+ if (roster.meshId !== peerMeshId(authorityPublicKey)) {
+ throw new Error('Peer Mesh roster has the wrong authority');
+ }
+ const signature = string(record.signature, 'signature', 128);
+ const signatureBytes = decodeCanonicalBase64Url(signature, 'roster signature');
+ if (signatureBytes.length !== 64) throw new Error('Invalid Peer Mesh roster signature');
+ const verified = verify(
+ null,
+ encodeRoster(roster),
+ decodePublicKey(authorityPublicKey),
+ signatureBytes,
+ );
+ if (!verified) throw new Error('Peer Mesh roster signature is invalid');
+ return Object.freeze({ roster, authorityPublicKey, signature });
+}
+
+export function createPeerMeshInvitationSecret(): string {
+ return randomBytes(32).toString('base64url');
+}
+
+export function createPeerMeshInvitationId(): string {
+ return randomUUID();
+}
+
+export function peerMeshInvitationSecretDigest(secret: string): string {
+ return createHash('sha256').update(decodeSecret(secret)).digest('base64url');
+}
+
+export function matchesPeerMeshInvitationSecret(secret: string, expectedDigest: string): boolean {
+ const actual = Buffer.from(peerMeshInvitationSecretDigest(secret), 'base64url');
+ const expected = Buffer.from(expectedDigest, 'base64url');
+ return actual.length === expected.length && timingSafeEqual(actual, expected);
+}
+
+export function decodePeerMeshInvitation(value: unknown): PeerMeshInvitationV1 {
+ const record = exactObject(value, 'Peer Mesh invitation', [
+ 'version',
+ 'meshId',
+ 'authorityPublicKey',
+ 'invitationId',
+ 'secret',
+ 'expiresAt',
+ 'peerId',
+ 'routeHints',
+ 'coordinationRelays',
+ ]);
+ if (record.version !== 1) throw new Error('Unsupported Peer Mesh invitation version');
+ const authorityPublicKey = string(record.authorityPublicKey, 'authorityPublicKey', 256);
+ const meshId = string(record.meshId, 'meshId', 128);
+ if (meshId !== peerMeshId(authorityPublicKey)) {
+ throw new Error('Peer Mesh invitation has the wrong authority');
+ }
+ const expiresAt = integer(record.expiresAt, 'expiresAt', 1);
+ return Object.freeze({
+ version: 1,
+ meshId,
+ authorityPublicKey,
+ invitationId: string(record.invitationId, 'invitationId', 64),
+ secret: validateSecret(record.secret),
+ expiresAt,
+ ...decodeAuthorityTarget({
+ peerId: record.peerId,
+ routeHints: record.routeHints,
+ coordinationRelays: record.coordinationRelays,
+ }),
+ });
+}
+
+export function canonicalPeerMeshRoster(value: unknown): PeerMeshRosterV1 {
+ const record = exactObject(value, 'Peer Mesh roster', [
+ 'version',
+ 'meshId',
+ 'revision',
+ 'members',
+ 'closed',
+ ]);
+ if (record.version !== 1) throw new Error('Unsupported Peer Mesh roster version');
+ const members = stringArray(record.members, 'members', PEER_MESH_MAX_MEMBERS, 256)
+ .map((member) => token(member, 'member', 256))
+ .sort();
+ if (members.length === 0 || new Set(members).size !== members.length) {
+ throw new Error('Peer Mesh roster members must be unique and non-empty');
+ }
+ if (typeof record.closed !== 'boolean') throw new Error('Invalid Peer Mesh roster closed');
+ return Object.freeze({
+ version: 1,
+ meshId: string(record.meshId, 'meshId', 128),
+ revision: integer(record.revision, 'revision', 1),
+ members: Object.freeze(members),
+ closed: record.closed,
+ });
+}
+
+export function decodeAuthorityTarget(value: unknown): PeerMeshAuthorityTarget {
+ const record = exactObject(value, 'Peer Mesh authority target', [
+ 'peerId',
+ 'routeHints',
+ 'coordinationRelays',
+ ]);
+ return Object.freeze({
+ peerId: token(record.peerId, 'peerId', 256),
+ routeHints: Object.freeze(addressArray(record.routeHints, 'routeHints')),
+ coordinationRelays: Object.freeze(
+ addressArray(record.coordinationRelays, 'coordinationRelays'),
+ ),
+ });
+}
+
+function encodeRoster(roster: PeerMeshRosterV1): Buffer {
+ return Buffer.from(
+ `maka.peer-mesh.roster.v1\n${JSON.stringify({
+ closed: roster.closed,
+ members: roster.members,
+ meshId: roster.meshId,
+ revision: roster.revision,
+ version: roster.version,
+ })}`,
+ );
+}
+
+function decodePublicKey(encoded: string) {
+ try {
+ const key = createPublicKey({
+ key: decodeCanonicalBase64Url(encoded, 'authority public key'),
+ format: 'der',
+ type: 'spki',
+ });
+ if (key.asymmetricKeyType !== 'ed25519') {
+ throw new Error('Peer Mesh authority key must be Ed25519');
+ }
+ return key;
+ } catch (error) {
+ throw new Error('Invalid Peer Mesh authority public key', { cause: error });
+ }
+}
+
+function validateSecret(value: unknown): string {
+ const secret = string(value, 'secret', 64);
+ decodeSecret(secret);
+ return secret;
+}
+
+function decodeSecret(secret: string): Buffer {
+ const bytes = decodeCanonicalBase64Url(secret, 'invitation secret');
+ if (bytes.length !== 32) {
+ throw new Error('Invalid Peer Mesh invitation secret');
+ }
+ return bytes;
+}
+
+function decodeCanonicalBase64Url(value: string, label: string): Buffer {
+ const bytes = Buffer.from(value, 'base64url');
+ if (bytes.toString('base64url') !== value) {
+ throw new Error(`Invalid Peer Mesh ${label}`);
+ }
+ return bytes;
+}
+
+function object(value: unknown, label: string): Record<string, unknown> {
+ if (!value || typeof value !== 'object' || Array.isArray(value)) {
+ throw new Error(`Invalid ${label}`);
+ }
+ return value as Record<string, unknown>;
+}
+
+function exactObject(
+ value: unknown,
+ label: string,
+ keys: readonly string[],
+): Record<string, unknown> {
+ const record = object(value, label);
+ if (
+ Object.keys(record).length !== keys.length ||
+ keys.some((key) => !Object.hasOwn(record, key))
+ ) {
+ throw new Error(`Invalid ${label}`);
+ }
+ return record;
+}
+
+function string(value: unknown, label: string, max: number): string {
+ if (typeof value !== 'string' || value.length === 0 || value.length > max) {
+ throw new Error(`Invalid Peer Mesh ${label}`);
+ }
+ return value;
+}
+
+function token(value: unknown, label: string, max: number): string {
+ const result = string(value, label, max);
+ if (/\s|[\u0000-\u001f\u007f]/u.test(result)) {
+ throw new Error(`Invalid Peer Mesh ${label}`);
+ }
+ return result;
+}
+
+function integer(value: unknown, label: string, minimum: number): number {
+ if (!Number.isSafeInteger(value) || (value as number) < minimum) {
+ throw new Error(`Invalid Peer Mesh ${label}`);
+ }
+ return value as number;
+}
+
+function stringArray(value: unknown, label: string, maxItems: number, maxLength: number): string[] {
+ if (!Array.isArray(value) || value.length > maxItems) {
+ throw new Error(`Invalid Peer Mesh ${label}`);
+ }
+ return value.map((item) => string(item, label, maxLength));
+}
+
+function addressArray(value: unknown, label: string): string[] {
+ const addresses = stringArray(value, label, PEER_MESH_MAX_ROUTE_HINTS, 1024);
+ if (
+ addresses.some(
+ (address) => !address.startsWith('/') || /\s|[\u0000-\u001f\u007f]/u.test(address),
+ ) ||
+ new Set(addresses).size !== addresses.length
+ ) {
+ throw new Error(`Invalid Peer Mesh ${label}`);
+ }
+ return addresses;
+}
diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts
new file mode 100644
index 000000000..c5a3e7ee1
--- /dev/null
+++ b/packages/runtime-host/src/peer-mesh/node.ts
@@ -0,0 +1,546 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { RuntimeHostPeerClient } from '../client/peer-client.js';
+import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js';
+import {
+ canonicalPeerMeshRoster,
+ createPeerMeshInvitationId,
+ createPeerMeshInvitationSecret,
+ decodePeerMeshInvitation,
+ decodeSignedPeerMeshRoster,
+ generatePeerMeshAuthorityKeyPair,
+ matchesPeerMeshInvitationSecret,
+ PEER_MESH_MAX_MEMBERS,
+ PEER_MESH_MAX_MESHES,
+ PEER_MESH_MAX_PENDING_INVITATIONS,
+ peerMeshId,
+ peerMeshInvitationSecretDigest,
+ signPeerMeshRoster,
+ type PeerMeshAuthorityTarget,
+ type PeerMeshInvitationV1,
+ type SignedPeerMeshRosterV1,
+} from './model.js';
+import {
+ authorityKeys,
+ openPeerMeshStateStore,
+ type PeerMeshStateStore,
+ type PeerMeshStateV1,
+} from './store.js';
+
+const CONTROL_FRAME_MAX_BYTES = 64 * 1024;
+const DEFAULT_INVITATION_TTL_MS = 15 * 60 * 1_000;
+const CONNECT_DEADLINE_MS = 30_000;
+const CONTROL_REQUEST_DEADLINE_MS = 10_000;
+const MAX_ACTIVE_CONTROL_STREAMS = 32;
+
+interface RedeemInvitationRequest {
+ readonly version: 1;
+ readonly kind: 'redeem-invitation';
+ readonly meshId: string;
+ readonly invitationId: string;
+ readonly secret: string;
+}
+
+type RedeemInvitationResponse =
+ | {
+ readonly version: 1;
+ readonly kind: 'invitation-redeemed';
+ readonly roster: SignedPeerMeshRosterV1;
+ }
+ | {
+ readonly version: 1;
+ readonly kind: 'invitation-rejected';
+ readonly reason: RedeemInvitationRejectionReason;
+ };
+
+type RedeemInvitationRejectionReason = 'invalid' | 'expired' | 'closed' | 'full';
+
+export interface PeerMeshNode {
+ status(): readonly PeerMeshStatus[];
+ create(): Promise<PeerMeshStatus>;
+ invite(meshId: string, input?: { readonly ttlMs?: number }): Promise<PeerMeshInvitationV1>;
+ join(invitation: PeerMeshInvitationV1, signal?: AbortSignal): Promise<PeerMeshStatus>;
+ remove(meshId: string, peerId: string): Promise<PeerMeshStatus>;
+ closeMesh(meshId: string): Promise<PeerMeshStatus>;
+ serve(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface PeerMeshStatus {
+ readonly role: 'authority' | 'member';
+ readonly localPeerId: string;
+ readonly authority: PeerMeshAuthorityTarget;
+ readonly roster: SignedPeerMeshRosterV1;
+ readonly pendingInvitationCount: number;
+}
+
+export async function openPeerMeshNode(input: {
+ readonly dataRoot: string;
+ readonly peer: RuntimeHostPeerClient;
+ readonly coordinationRelays?: readonly string[];
+ readonly now?: () => number;
+}): Promise<PeerMeshNode> {
+ const store = await openPeerMeshStateStore(input.dataRoot);
+ try {
+ const states = store.read();
+ if (states.some(({ localPeerId }) => localPeerId !== input.peer.identity().peerId)) {
+ throw new Error('Peer Mesh state belongs to a different peer identity');
+ }
+ return new PeerMeshNodeImpl({ ...input, store });
+ } catch (error) {
+ await store.close();
+ throw error;
+ }
+}
+
+class PeerMeshNodeImpl implements PeerMeshNode {
+ readonly #store: PeerMeshStateStore;
+ readonly #peer: RuntimeHostPeerClient;
+ readonly #coordinationRelays: readonly string[];
+ readonly #now: () => number;
+ readonly #activeControlStreams = new Set<RuntimeHostPeerNativeStream>();
+ #closed = false;
+
+ constructor(input: {
+ readonly store: PeerMeshStateStore;
+ readonly peer: RuntimeHostPeerClient;
+ readonly coordinationRelays?: readonly string[];
+ readonly now?: () => number;
+ }) {
+ this.#store = input.store;
+ this.#peer = input.peer;
+ this.#coordinationRelays = input.coordinationRelays ?? [];
+ this.#now = input.now ?? Date.now;
+ }
+
+ status(): readonly PeerMeshStatus[] {
+ return Object.freeze(this.#store.read().map(peerMeshStatus));
+ }
+
+ async create(): Promise<PeerMeshStatus> {
+ const identity = this.#peer.identity();
+ const state = await this.#store.mutate((current) => {
+ if (current.length >= PEER_MESH_MAX_MESHES)
+ throw new Error('This peer belongs to too many Peer Meshes');
+ const keys = generatePeerMeshAuthorityKeyPair();
+ const roster = signPeerMeshRoster(
+ canonicalPeerMeshRoster({
+ version: 1,
+ meshId: peerMeshId(keys.publicKey),
+ revision: 1,
+ members: [identity.peerId],
+ closed: false,
+ }),
+ keys,
+ );
+ const state: PeerMeshStateV1 = {
+ version: 1,
+ role: 'authority',
+ localPeerId: identity.peerId,
+ authority: {
+ peerId: identity.peerId,
+ routeHints: identity.listenAddresses,
+ coordinationRelays: this.#coordinationRelays,
+ },
+ roster,
+ authorityPrivateKey: keys.privateKey,
+ pendingInvitations: [],
+ };
+ return { state: [...current, state], result: state };
+ });
+ return peerMeshStatus(state);
+ }
+
+ invite(meshId: string, input: { readonly ttlMs?: number } = {}): Promise<PeerMeshInvitationV1> {
+ const now = this.#now();
+ const identity = this.#peer.identity();
+ const ttlMs = input.ttlMs ?? DEFAULT_INVITATION_TTL_MS;
+ if (!Number.isSafeInteger(ttlMs) || ttlMs < 1_000 || ttlMs > 24 * 60 * 60 * 1_000) {
+ return Promise.reject(
+ new Error('Peer Mesh invitation TTL must be between 1 second and 1 day'),
+ );
+ }
+ return this.#store.mutate((current) => {
+ const state = requireAuthority(current, meshId);
+ if (state.localPeerId !== identity.peerId) {
+ throw new Error('Peer Mesh state belongs to a different peer identity');
+ }
+ if (state.roster.roster.closed) throw new Error('Peer Mesh is closed');
+ const pendingInvitations = state.pendingInvitations.filter(
+ ({ expiresAt }) => expiresAt > now,
+ );
+ if (pendingInvitations.length >= PEER_MESH_MAX_PENDING_INVITATIONS)
+ throw new Error('Peer Mesh has too many pending invitations');
+ const invitationId = createPeerMeshInvitationId();
+ const secret = createPeerMeshInvitationSecret();
+ const expiresAt = now + ttlMs;
+ const invitation: PeerMeshInvitationV1 = {
+ version: 1,
+ meshId: state.roster.roster.meshId,
+ authorityPublicKey: state.roster.authorityPublicKey,
+ invitationId,
+ secret,
+ expiresAt,
+ peerId: identity.peerId,
+ routeHints: identity.listenAddresses,
+ coordinationRelays: this.#coordinationRelays,
+ };
+ return {
+ state: replaceMesh(current, {
+ ...state,
+ authority: {
+ peerId: identity.peerId,
+ routeHints: identity.listenAddresses,
+ coordinationRelays: this.#coordinationRelays,
+ },
+ pendingInvitations: [
+ ...pendingInvitations,
+ { invitationId, secretDigest: peerMeshInvitationSecretDigest(secret), expiresAt },
+ ],
+ }),
+ result: Object.freeze(invitation),
+ };
+ });
+ }
+
+ async join(invitationValue: PeerMeshInvitationV1, signal?: AbortSignal): Promise<PeerMeshStatus> {
+ const invitation = decodePeerMeshInvitation(invitationValue);
+ if (findMesh(this.#store.read(), invitation.meshId)) {
+ throw new Error('This peer already belongs to that Peer Mesh');
+ }
+ if (invitation.expiresAt <= this.#now()) throw new Error('Peer Mesh invitation has expired');
+ const stream = await this.#peer.connectMeshControl(
+ {
+ peerId: invitation.peerId,
+ routeHints: invitation.routeHints,
+ coordinationRelays: invitation.coordinationRelays,
+ directDeadlineMs: CONNECT_DEADLINE_MS,
+ },
+ signal,
+ );
+ try {
+ const request: RedeemInvitationRequest = {
+ version: 1,
+ kind: 'redeem-invitation',
+ meshId: invitation.meshId,
+ invitationId: invitation.invitationId,
+ secret: invitation.secret,
+ };
+ await writeFrame(stream, request);
+ const response = decodeRedeemResponse(await readFrame(stream));
+ if (response.kind === 'invitation-rejected') {
+ throw new Error(`Peer Mesh invitation was rejected: ${response.reason}`);
+ }
+ const roster = decodeSignedPeerMeshRoster(response.roster);
+ const identity = this.#peer.identity();
+ if (
+ roster.roster.meshId !== invitation.meshId ||
+ roster.authorityPublicKey !== invitation.authorityPublicKey ||
+ !roster.roster.members.includes(identity.peerId)
+ ) {
+ throw new Error('Peer Mesh authority returned an unrelated roster');
+ }
+ const state: PeerMeshStateV1 = {
+ version: 1,
+ role: 'member',
+ localPeerId: identity.peerId,
+ authority: {
+ peerId: invitation.peerId,
+ routeHints: invitation.routeHints,
+ coordinationRelays: invitation.coordinationRelays,
+ },
+ roster,
+ pendingInvitations: [],
+ };
+ const joined = await this.#store.mutate((current) => {
+ if (findMesh(current, invitation.meshId)) {
+ throw new Error('This peer already belongs to that Peer Mesh');
+ }
+ if (current.length >= PEER_MESH_MAX_MESHES)
+ throw new Error('This peer belongs to too many Peer Meshes');
+ return { state: [...current, state], result: state };
+ });
+ return peerMeshStatus(joined);
+ } finally {
+ await stream.close().catch(() => undefined);
+ }
+ }
+
+ remove(meshId: string, peerId: string): Promise<PeerMeshStatus> {
+ return this.#updateAuthorityRoster(meshId, false, (state) => {
+ if (peerId === state.localPeerId) throw new Error('Peer Mesh authority cannot remove itself');
+ const members = state.roster.roster.members.filter((member) => member !== peerId);
+ if (members.length === state.roster.roster.members.length) {
+ throw new Error('Peer is not a member of this Peer Mesh');
+ }
+ return { members, closed: false };
+ });
+ }
+
+ closeMesh(meshId: string): Promise<PeerMeshStatus> {
+ return this.#updateAuthorityRoster(meshId, true, (state) => ({
+ members: state.roster.roster.members,
+ closed: true,
+ }));
+ }
+
+ async serve(): Promise<void> {
+ while (!this.#closed) {
+ const stream = await this.#peer.acceptMeshControl();
+ if (!stream) return;
+ if (this.#closed || this.#activeControlStreams.size >= MAX_ACTIVE_CONTROL_STREAMS) {
+ stream.abort();
+ if (this.#closed) return;
+ continue;
+ }
+ this.#activeControlStreams.add(stream);
+ void this.#handleIncoming(stream).finally(() => this.#activeControlStreams.delete(stream));
+ }
+ }
+
+ async close(): Promise<void> {
+ if (this.#closed) return;
+ this.#closed = true;
+ for (const stream of this.#activeControlStreams) stream.abort();
+ this.#activeControlStreams.clear();
+ return this.#store.close();
+ }
+
+ #updateAuthorityRoster(
+ meshId: string,
+ closedIsSuccess: boolean,
+ update: (state: PeerMeshStateV1) => {
+ readonly members: readonly string[];
+ readonly closed: boolean;
+ },
+ ): Promise<PeerMeshStatus> {
+ return this.#store.mutate((current) => {
+ const state = requireAuthority(current, meshId);
+ if (state.roster.roster.closed) {
+ if (closedIsSuccess) return { state: current, result: peerMeshStatus(state) };
+ throw new Error('Peer Mesh is closed');
+ }
+ const next = update(state);
+ const roster = signPeerMeshRoster(
+ {
+ version: 1,
+ meshId: state.roster.roster.meshId,
+ revision: state.roster.roster.revision + 1,
+ members: next.members,
+ closed: next.closed,
+ },
+ authorityKeys(state),
+ );
+ const updated = {
+ ...state,
+ roster,
+ ...(next.closed ? { pendingInvitations: [] } : {}),
+ };
+ return { state: replaceMesh(current, updated), result: peerMeshStatus(updated) };
+ });
+ }
+
+ async #handleIncoming(stream: RuntimeHostPeerNativeStream): Promise<void> {
+ const deadline = setTimeout(() => stream.abort(), CONTROL_REQUEST_DEADLINE_MS);
+ try {
+ const request = decodeRedeemRequest(await readFrame(stream));
+ const response = await this.#redeem(request, stream.peerId);
+ await writeFrame(stream, response);
+ await stream.close();
+ } catch {
+ stream.abort();
+ } finally {
+ clearTimeout(deadline);
+ }
+ }
+
+ #redeem(
+ request: RedeemInvitationRequest,
+ remotePeerId: string,
+ ): Promise<RedeemInvitationResponse> {
+ const now = this.#now();
+ return this.#store.mutate<RedeemInvitationResponse>((current) => {
+ const state = findMesh(current, request.meshId);
+ if (!state || state.role !== 'authority')
+ return { state: current, result: rejected('invalid') };
+ const invitation = state.pendingInvitations.find(
+ ({ invitationId }) => invitationId === request.invitationId,
+ );
+ if (
+ request.meshId !== state.roster.roster.meshId ||
+ !invitation ||
+ !matchesPeerMeshInvitationSecret(request.secret, invitation.secretDigest)
+ ) {
+ return { state: current, result: rejected('invalid') };
+ }
+ const remaining = state.pendingInvitations.filter(
+ ({ invitationId, expiresAt }) => invitationId !== request.invitationId && expiresAt > now,
+ );
+ if (invitation.expiresAt <= now) {
+ return {
+ state: replaceMesh(current, { ...state, pendingInvitations: remaining }),
+ result: rejected('expired'),
+ };
+ }
+ if (state.roster.roster.closed) {
+ return {
+ state: replaceMesh(current, { ...state, pendingInvitations: remaining }),
+ result: rejected('closed'),
+ };
+ }
+ if (
+ !state.roster.roster.members.includes(remotePeerId) &&
+ state.roster.roster.members.length >= PEER_MESH_MAX_MEMBERS
+ ) {
+ return {
+ state: replaceMesh(current, { ...state, pendingInvitations: remaining }),
+ result: rejected('full'),
+ };
+ }
+ if (state.roster.roster.members.includes(remotePeerId)) {
+ return {
+ state: replaceMesh(current, { ...state, pendingInvitations: remaining }),
+ result: { version: 1, kind: 'invitation-redeemed', roster: state.roster },
+ };
+ }
+ const members = [...state.roster.roster.members, remotePeerId].sort();
+ const roster = signPeerMeshRoster(
+ {
+ ...state.roster.roster,
+ revision: state.roster.roster.revision + 1,
+ members,
+ },
+ authorityKeys(state),
+ );
+ return {
+ state: replaceMesh(current, { ...state, roster, pendingInvitations: remaining }),
+ result: { version: 1, kind: 'invitation-redeemed', roster },
+ };
+ });
+ }
+}
+
+function peerMeshStatus(state: PeerMeshStateV1): PeerMeshStatus {
+ return Object.freeze({
+ role: state.role,
+ localPeerId: state.localPeerId,
+ authority: state.authority,
+ roster: state.roster,
+ pendingInvitationCount: state.pendingInvitations.length,
+ });
+}
+
+function requireAuthority(states: readonly PeerMeshStateV1[], meshId: string): PeerMeshStateV1 {
+ const state = findMesh(states, meshId);
+ if (!state || state.role !== 'authority')
+ throw new Error('Peer Mesh operation requires authority');
+ return state;
+}
+
+function findMesh(states: readonly PeerMeshStateV1[], meshId: string): PeerMeshStateV1 | undefined {
+ return states.find(({ roster }) => roster.roster.meshId === meshId);
+}
+
+function replaceMesh(
+ states: readonly PeerMeshStateV1[],
+ next: PeerMeshStateV1,
+): readonly PeerMeshStateV1[] {
+ return states.map((state) =>
+ state.roster.roster.meshId === next.roster.roster.meshId ? next : state,
+ );
+}
+
+function rejected(reason: RedeemInvitationRejectionReason) {
+ return { version: 1, kind: 'invitation-rejected', reason } as const;
+}
+
+function decodeRedeemRequest(value: unknown): RedeemInvitationRequest {
+ const record = recordValue(value);
+ if (record.version !== 1 || record.kind !== 'redeem-invitation') {
+ throw new Error('Unsupported Peer Mesh control request');
+ }
+ return {
+ version: 1,
+ kind: 'redeem-invitation',
+ meshId: requiredString(record.meshId, 128),
+ invitationId: requiredString(record.invitationId, 64),
+ secret: requiredString(record.secret, 64),
+ };
+}
+
+function decodeRedeemResponse(value: unknown): RedeemInvitationResponse {
+ const record = recordValue(value);
+ if (record.version !== 1) throw new Error('Unsupported Peer Mesh control response');
+ if (record.kind === 'invitation-redeemed') {
+ return {
+ version: 1,
+ kind: 'invitation-redeemed',
+ roster: decodeSignedPeerMeshRoster(record.roster),
+ };
+ }
+ if (
+ record.kind === 'invitation-rejected' &&
+ (record.reason === 'invalid' ||
+ record.reason === 'expired' ||
+ record.reason === 'closed' ||
+ record.reason === 'full')
+ ) {
+ return { version: 1, kind: 'invitation-rejected', reason: record.reason };
+ }
+ throw new Error('Invalid Peer Mesh control response');
+}
+
+async function writeFrame(stream: RuntimeHostPeerNativeStream, value: unknown): Promise<void> {
+ const bytes = Buffer.from(`${JSON.stringify(value)}\n`);
+ if (bytes.length > CONTROL_FRAME_MAX_BYTES)
+ throw new Error('Peer Mesh control frame is too large');
+ await stream.write(bytes);
+}
+
+async function readFrame(stream: RuntimeHostPeerNativeStream): Promise<unknown> {
+ let buffered = Buffer.alloc(0);
+ for (;;) {
+ const chunk = await stream.read();
+ if (!chunk) throw new Error('Peer Mesh control stream ended before a frame arrived');
+ buffered = Buffer.concat([buffered, chunk]);
+ if (buffered.length > CONTROL_FRAME_MAX_BYTES)
+ throw new Error('Peer Mesh control frame is too large');
+ const newline = buffered.indexOf(0x0a);
+ if (newline < 0) continue;
+ if (buffered.subarray(newline + 1).some((byte) => byte > 0x20)) {
+ throw new Error('Peer Mesh control stream contained multiple frames');
+ }
+ return JSON.parse(buffered.subarray(0, newline).toString('utf8')) as unknown;
+ }
+}
+
+function recordValue(value: unknown): Record<string, unknown> {
+ if (!value || typeof value !== 'object' || Array.isArray(value)) {
+ throw new Error('Invalid Peer Mesh control frame');
+ }
+ return value as Record<string, unknown>;
+}
+
+function requiredString(value: unknown, maxLength: number): string {
+ if (typeof value !== 'string' || value.length === 0 || value.length > maxLength) {
+ throw new Error('Invalid Peer Mesh control value');
+ }
+ return value;
+}
diff --git a/packages/runtime-host/src/peer-mesh/store.ts b/packages/runtime-host/src/peer-mesh/store.ts
new file mode 100644
index 000000000..bc27162b2
--- /dev/null
+++ b/packages/runtime-host/src/peer-mesh/store.ts
@@ -0,0 +1,340 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { chmod, lstat, mkdir, open, readFile, rename, unlink } from 'node:fs/promises';
+import { dirname, join } from 'node:path';
+import {
+ acquireFileLifetimeOwner,
+ type FileLifetimeOwner,
+} from '@maka/storage/file-lifetime-owner';
+import {
+ decodeAuthorityTarget,
+ decodeSignedPeerMeshRoster,
+ PEER_MESH_MAX_MESHES,
+ PEER_MESH_MAX_PENDING_INVITATIONS,
+ type PeerMeshAuthorityKeyPair,
+ type PeerMeshAuthorityTarget,
+ type SignedPeerMeshRosterV1,
+ validatePeerMeshAuthorityKeyPair,
+} from './model.js';
+
+const STATE_FILE = 'peer-mesh.json';
+const LOCK_FILE = 'peer-mesh.owner';
+const MAX_STATE_BYTES = 1024 * 1024;
+
+export interface PendingPeerMeshInvitation {
+ readonly invitationId: string;
+ readonly secretDigest: string;
+ readonly expiresAt: number;
+}
+
+export interface PeerMeshStateV1 {
+ readonly version: 1;
+ readonly role: 'authority' | 'member';
+ readonly localPeerId: string;
+ readonly authority: PeerMeshAuthorityTarget;
+ readonly roster: SignedPeerMeshRosterV1;
+ readonly authorityPrivateKey?: string;
+ readonly pendingInvitations: readonly PendingPeerMeshInvitation[];
+}
+
+export interface PeerMeshStateStore {
+ read(): readonly PeerMeshStateV1[];
+ mutate<T>(
+ operation: (
+ state: readonly PeerMeshStateV1[],
+ ) =>
+ | { readonly state: readonly PeerMeshStateV1[]; readonly result: T }
+ | Promise<{ readonly state: readonly PeerMeshStateV1[]; readonly result: T }>,
+ ): Promise<T>;
+ close(): Promise<void>;
+}
+
+export async function openPeerMeshStateStore(dataRoot: string): Promise<PeerMeshStateStore> {
+ await mkdir(dataRoot, { recursive: true, mode: 0o700 });
+ if (process.platform !== 'win32') await chmod(dataRoot, 0o700);
+ const owner = await acquireFileLifetimeOwner(join(dataRoot, LOCK_FILE));
+ try {
+ const state = await readState(join(dataRoot, STATE_FILE));
+ return new PeerMeshStateStoreImpl(dataRoot, owner, state);
+ } catch (error) {
+ await owner.close();
+ throw error;
+ }
+}
+
+class PeerMeshStateStoreImpl implements PeerMeshStateStore {
+ readonly #path: string;
+ #state: readonly PeerMeshStateV1[];
+ #tail = Promise.resolve();
+ #closed = false;
+
+ constructor(
+ dataRoot: string,
+ private readonly owner: FileLifetimeOwner,
+ state: readonly PeerMeshStateV1[],
+ ) {
+ this.#path = join(dataRoot, STATE_FILE);
+ this.#state = state;
+ }
+
+ read(): readonly PeerMeshStateV1[] {
+ this.#assertOpen();
+ return this.#state;
+ }
+
+ mutate<T>(
+ operation: (
+ state: readonly PeerMeshStateV1[],
+ ) =>
+ | { readonly state: readonly PeerMeshStateV1[]; readonly result: T }
+ | Promise<{ readonly state: readonly PeerMeshStateV1[]; readonly result: T }>,
+ ): Promise<T> {
+ this.#assertOpen();
+ const task = this.#tail.then(async () => {
+ const updated = await operation(this.#state);
+ if (updated.state === this.#state) return updated.result;
+ const canonical = decodePeerMeshStates(updated.state);
+ assertStateAdvance(this.#state, canonical);
+ await writeState(this.#path, canonical);
+ this.#state = canonical;
+ return updated.result;
+ });
+ this.#tail = task.then(
+ () => undefined,
+ () => undefined,
+ );
+ return task;
+ }
+
+ async close(): Promise<void> {
+ if (this.#closed) return;
+ this.#closed = true;
+ await this.#tail;
+ await this.owner.close();
+ }
+
+ #assertOpen(): void {
+ if (this.#closed) throw new Error('Peer Mesh state store is closed');
+ }
+}
+
+export function decodePeerMeshState(value: unknown): PeerMeshStateV1 {
+ if (!value || typeof value !== 'object' || Array.isArray(value)) {
+ throw new Error('Invalid Peer Mesh state');
+ }
+ const record = value as Record<string, unknown>;
+ const expectedKeys =
+ record.role === 'authority'
+ ? [
+ 'version',
+ 'role',
+ 'localPeerId',
+ 'authority',
+ 'roster',
+ 'authorityPrivateKey',
+ 'pendingInvitations',
+ ]
+ : ['version', 'role', 'localPeerId', 'authority', 'roster', 'pendingInvitations'];
+ if (
+ Object.keys(record).length !== expectedKeys.length ||
+ expectedKeys.some((key) => !Object.hasOwn(record, key))
+ ) {
+ throw new Error('Invalid Peer Mesh state');
+ }
+ if (record.version !== 1 || (record.role !== 'authority' && record.role !== 'member')) {
+ throw new Error('Unsupported Peer Mesh state');
+ }
+ const localPeerId = boundedString(record.localPeerId, 'localPeerId', 256);
+ const authority = decodeAuthorityTarget(record.authority);
+ const roster = decodeSignedPeerMeshRoster(record.roster);
+ if (!roster.roster.members.includes(localPeerId)) {
+ throw new Error('Local peer is not a member of its Peer Mesh roster');
+ }
+ const privateKey = record.authorityPrivateKey;
+ if (record.role === 'authority') {
+ if (authority.peerId !== localPeerId || typeof privateKey !== 'string') {
+ throw new Error('Invalid Peer Mesh authority state');
+ }
+ validatePeerMeshAuthorityKeyPair({
+ publicKey: roster.authorityPublicKey,
+ privateKey: boundedString(privateKey, 'authorityPrivateKey', 256),
+ });
+ } else if (privateKey !== undefined) {
+ throw new Error('Peer Mesh member state contains an authority private key');
+ }
+ if (!roster.roster.members.includes(authority.peerId)) {
+ throw new Error('Peer Mesh authority is not present in its roster');
+ }
+ const pending = decodePendingInvitations(record.pendingInvitations);
+ if (record.role !== 'authority' && pending.length !== 0) {
+ throw new Error('Peer Mesh member state contains authority invitations');
+ }
+ return Object.freeze({
+ version: 1,
+ role: record.role,
+ localPeerId,
+ authority,
+ roster,
+ ...(privateKey ? { authorityPrivateKey: privateKey } : {}),
+ pendingInvitations: Object.freeze(pending),
+ });
+}
+
+function decodePeerMeshStates(value: unknown): readonly PeerMeshStateV1[] {
+ if (!Array.isArray(value) || value.length > PEER_MESH_MAX_MESHES) {
+ throw new Error('Invalid Peer Mesh state collection');
+ }
+ const states = value.map(decodePeerMeshState);
+ const meshIds = states.map(({ roster }) => roster.roster.meshId);
+ if (new Set(meshIds).size !== meshIds.length) {
+ throw new Error('Duplicate Peer Mesh state');
+ }
+ return Object.freeze(states);
+}
+
+function assertStateAdvance(
+ current: readonly PeerMeshStateV1[],
+ next: readonly PeerMeshStateV1[],
+): void {
+ for (const previous of current) {
+ const updated = next.find(
+ ({ roster }) => roster.roster.meshId === previous.roster.roster.meshId,
+ );
+ if (!updated) throw new Error('Peer Mesh state cannot be removed implicitly');
+ if (updated.localPeerId !== previous.localPeerId || updated.role !== previous.role) {
+ throw new Error('Peer Mesh state identity cannot change');
+ }
+ if (updated.roster.roster.revision < previous.roster.roster.revision) {
+ throw new Error('Peer Mesh roster revision cannot roll back');
+ }
+ if (
+ updated.roster.roster.revision === previous.roster.roster.revision &&
+ JSON.stringify(updated.roster) !== JSON.stringify(previous.roster)
+ ) {
+ throw new Error('Peer Mesh roster revision cannot identify different facts');
+ }
+ if (previous.roster.roster.closed && !updated.roster.roster.closed) {
+ throw new Error('Closed Peer Mesh state is terminal');
+ }
+ }
+}
+
+export function authorityKeys(state: PeerMeshStateV1): PeerMeshAuthorityKeyPair {
+ if (state.role !== 'authority' || !state.authorityPrivateKey) {
+ throw new Error('Peer Mesh operation requires the authority');
+ }
+ return Object.freeze({
+ publicKey: state.roster.authorityPublicKey,
+ privateKey: state.authorityPrivateKey,
+ });
+}
+
+async function readState(path: string): Promise<readonly PeerMeshStateV1[]> {
+ try {
+ const info = await lstat(path);
+ if (!info.isFile() || info.size > MAX_STATE_BYTES)
+ throw new Error('Invalid Peer Mesh state file');
+ const document = JSON.parse(await readFile(path, 'utf8')) as unknown;
+ if (!document || typeof document !== 'object' || Array.isArray(document)) {
+ throw new Error('Invalid Peer Mesh state document');
+ }
+ const record = document as Record<string, unknown>;
+ if (record.version !== 1) throw new Error('Unsupported Peer Mesh state document');
+ return decodePeerMeshStates(record.meshes);
+ } catch (error) {
+ if (isNodeError(error, 'ENOENT')) return Object.freeze([]);
+ throw error;
+ }
+}
+
+async function writeState(path: string, state: readonly PeerMeshStateV1[]): Promise<void> {
+ if (state.length === 0) {
+ await unlink(path).catch((error: unknown) => {
+ if (!isNodeError(error, 'ENOENT')) throw error;
+ });
+ return;
+ }
+ const document = `${JSON.stringify({ version: 1, meshes: state }, null, 2)}\n`;
+ if (Buffer.byteLength(document) > MAX_STATE_BYTES)
+ throw new Error('Peer Mesh state is too large');
+ const temporary = `${path}.${process.pid}.tmp`;
+ let replaced = false;
+ try {
+ const handle = await open(temporary, 'wx', 0o600);
+ try {
+ await handle.writeFile(document, 'utf8');
+ await handle.sync();
+ } finally {
+ await handle.close();
+ }
+ if (process.platform !== 'win32') await chmod(temporary, 0o600);
+ await rename(temporary, path);
+ replaced = true;
+ await syncDirectory(dirname(path));
+ } finally {
+ if (!replaced) await unlink(temporary).catch(() => undefined);
+ }
+}
+
+function decodePendingInvitations(value: unknown): PendingPeerMeshInvitation[] {
+ if (!Array.isArray(value) || value.length > PEER_MESH_MAX_PENDING_INVITATIONS) {
+ throw new Error('Invalid pending Peer Mesh invitations');
+ }
+ const invitations = value.map((entry) => {
+ if (!entry || typeof entry !== 'object' || Array.isArray(entry)) {
+ throw new Error('Invalid pending Peer Mesh invitation');
+ }
+ const record = entry as Record<string, unknown>;
+ const expiresAt = record.expiresAt;
+ if (!Number.isSafeInteger(expiresAt) || (expiresAt as number) < 1) {
+ throw new Error('Invalid pending Peer Mesh invitation expiry');
+ }
+ return Object.freeze({
+ invitationId: boundedString(record.invitationId, 'invitationId', 64),
+ secretDigest: boundedString(record.secretDigest, 'secretDigest', 64),
+ expiresAt: expiresAt as number,
+ });
+ });
+ if (new Set(invitations.map(({ invitationId }) => invitationId)).size !== invitations.length) {
+ throw new Error('Duplicate pending Peer Mesh invitation');
+ }
+ return invitations;
+}
+
+function boundedString(value: unknown, label: string, max: number): string {
+ if (typeof value !== 'string' || value.length === 0 || value.length > max) {
+ throw new Error(`Invalid Peer Mesh ${label}`);
+ }
+ return value;
+}
+
+async function syncDirectory(path: string): Promise<void> {
+ if (process.platform === 'win32') return;
+ const handle = await open(path, 'r');
+ try {
+ await handle.sync();
+ } finally {
+ await handle.close();
+ }
+}
+
+function isNodeError(error: unknown, code: string): error is NodeJS.ErrnoException {
+ return error instanceof Error && 'code' in error && error.code === code;
+}
diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts
index a517b41ee..2ec15be4d 100644
--- a/packages/runtime-host/src/transport/peer-native.ts
+++ b/packages/runtime-host/src/transport/peer-native.ts
@@ -46,6 +46,7 @@ export class RuntimeHostPeerError extends Error {
}
export interface RuntimeHostPeerNativeStream {
+ readonly peerId: string;
read(): Promise<Buffer | null>;
write(bytes: Buffer): Promise<void>;
close(): Promise<void>;
@@ -62,8 +63,16 @@ export interface RuntimeHostPeerNativeEndpoint {
readonly coordinationRelays?: readonly string[];
readonly directDeadlineMs: number;
}): Promise<RuntimeHostPeerNativeStream>;
+ connectMeshControl(options: {
+ readonly requestId: number;
+ readonly peerId: string;
+ readonly routeHints: readonly string[];
+ readonly coordinationRelays?: readonly string[];
+ readonly directDeadlineMs: number;
+ }): Promise<RuntimeHostPeerNativeStream>;
cancelConnect(requestId: number): Promise<boolean>;
accept(): Promise<RuntimeHostPeerNativeStream | null>;
+ acceptMeshControl(): Promise<RuntimeHostPeerNativeStream | null>;
close(): Promise<void>;
}
diff --git a/packages/storage/package.json b/packages/storage/package.json
index 69f0eddf9..47ea5c420 100644
--- a/packages/storage/package.json
+++ b/packages/storage/package.json
@@ -17,6 +17,7 @@
"./encrypted-file-managed-secret-store": "./dist/encrypted-file-managed-secret-store.js",
"./execution-stores": "./dist/execution-stores.js",
"./external-sessions": "./dist/external-sessions.js",
+ "./file-lifetime-owner": "./dist/file-lifetime-owner.js",
"./file-update-lock": "./dist/file-update-lock.js",
"./foreign-session-store": "./dist/foreign-session-store.js",
"./git-worktree-child-executor": "./dist/git-worktree-child-executor.js",
diff --git a/packages/storage/src/file-lifetime-owner.ts b/packages/storage/src/file-lifetime-owner.ts
new file mode 100644
index 000000000..30cccf4eb
--- /dev/null
+++ b/packages/storage/src/file-lifetime-owner.ts
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { FileHandle } from 'node:fs/promises';
+import {
+ openStableNativeLockFile,
+ releaseNativeFileLock,
+ tryAcquireNativeFileLock,
+} from './native-file-lock.js';
+
+export interface FileLifetimeOwner {
+ close(): Promise<void>;
+}
+
+export async function acquireFileLifetimeOwner(path: string): Promise<FileLifetimeOwner> {
+ const handle = await openStableNativeLockFile(path);
+ if (!tryAcquireNativeFileLock(handle)) {
+ await handle.close();
+ throw new Error(`Another process owns ${path}`);
+ }
+ return new FileLifetimeOwnerImpl(handle);
+}
+
+class FileLifetimeOwnerImpl implements FileLifetimeOwner {
+ #closed = false;
+
+ constructor(private readonly handle: FileHandle) {}
+
+ async close(): Promise<void> {
+ if (this.#closed) return;
+ this.#closed = true;
+ releaseNativeFileLock(this.handle);
+ await this.handle.close();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org