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/09/05 13:39:50 UTC
(maka) 01/01: fix(peer): stabilize shared Session admission and path recovery
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/session-peer-collaboration
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 649894e664ed1a1fd63f6b2c481855f42062fe8f
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Sat Sep 5 21:39:37 2026 +0800
fix(peer): stabilize shared Session admission and path recovery
Align peer stream quotas with distinct Guest mounts, report admission capacity, retire intentional closes with logical FIN, and avoid Guest capability-provider ownership. Hedge stalled stream establishment without replaying application requests. Cover collaboration authority, exact retries, subscriptions and restart with the production Host stack.
Generated-by: OpenAI Codex
---
.../__tests__/runtime-host-desktop-manager.test.ts | 27 ++
.../src/main/runtime-host-desktop-manager.ts | 5 +-
native/runtime-host-peer/src/engine.rs | 226 ++++++++++----
.../src/__tests__/peer-listener.test.ts | 74 ++++-
.../runtime-host/src/__tests__/peer-native.test.ts | 19 ++
.../__tests__/peer-session-collaboration.test.ts | 323 +++++++++++++++++++++
.../src/__tests__/reconnecting-connection.test.ts | 64 ++++
packages/runtime-host/src/client/connection.ts | 26 +-
packages/runtime-host/src/client/index.ts | 2 +-
.../runtime-host/src/client/reconnect-lifecycle.ts | 20 +-
.../runtime-host/src/server/connection-session.ts | 5 +
packages/runtime-host/src/server/peer-listener.ts | 39 ++-
packages/runtime-host/src/transport/peer-native.ts | 34 ++-
13 files changed, 789 insertions(+), 75 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts
index b2392b6b5b..c7c127fc44 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts
@@ -22,6 +22,7 @@ import test from 'node:test';
import type { BotIncomingMessage } from '@maka/runtime/bots';
import {
RuntimeHostOperationError,
+ RuntimeHostPeerError,
RuntimeHostPermanentReconnectError,
RuntimeHostRequestInterruptedError,
type RuntimeHostSpawnedProcess,
@@ -822,6 +823,32 @@ test('reconnects after a pairing candidate becomes bound to this Client', async
await manager.close();
});
+test('reports Guest admission capacity instead of retrying the initial mount', async () => {
+ const local = candidateHarness();
+ const capacity = new RuntimeHostPeerError('peer_capacity_exceeded', 'Host connection capacity is full');
+ let starts = 0;
+ const manager = await startRuntimeHostDesktopManager(
+ {} as DesktopRuntimeHostCandidateStartInput,
+ {
+ startCandidate: async () => {
+ starts += 1;
+ if (starts === 1) return ready(local.candidate);
+ throw capacity;
+ },
+ },
+ );
+ try {
+ await assert.rejects(
+ manager.mountGuest(peerGuestTarget('shared-full'), () => undefined),
+ (error: unknown) => error === capacity,
+ );
+ assert.equal(starts, 2);
+ assert.equal(manager.current('shared-full'), undefined);
+ } finally {
+ await manager.close();
+ }
+});
+
test('completes Guest import at credential activation while reconnect continues', async () => {
const local = candidateHarness({ hostId: 'host-a' });
const remoteHostId = 'a'.repeat(64);
diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts
index 36d8eca62d..0d8e505e9b 100644
--- a/apps/desktop/src/main/runtime-host-desktop-manager.ts
+++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts
@@ -25,6 +25,7 @@ import {
forceTerminateRegisteredRuntimeHost,
RuntimeHostOperationError,
RuntimeHostPermanentReconnectError,
+ RuntimeHostPeerError,
RuntimeHostRequestInterruptedError,
runtimeHostStartupError,
LOCAL_RUNTIME_HOST_PROFILE,
@@ -1009,7 +1010,9 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
first ? target.input.onConnectionPhase : undefined,
);
},
- retryInitialFailure,
+ retryInitialFailure: retryInitialFailure
+ ? (error) => !(error instanceof RuntimeHostPeerError && error.code === 'peer_capacity_exceeded')
+ : false,
...(initialSignal ? { initialSignal } : {}),
onReconnectError: (error) => {
console.warn('[runtime-host] reconnect attempt failed:', error);
diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs
index 618bcbb128..3cae471721 100644
--- a/native/runtime-host-peer/src/engine.rs
+++ b/native/runtime-host-peer/src/engine.rs
@@ -92,6 +92,8 @@ const MAX_REMEMBERED_RELAY_FAILURES: u8 = 3;
const MAX_RELAY_ADDRESSES_PER_PEER: usize = 4;
const MAX_PUBLISHED_COORDINATION_RELAY_ADDRESSES: usize = 16;
const TRANSIT_FALLBACK_DELAY: Duration = Duration::from_millis(250);
+const STREAM_OPEN_HEDGE_DELAY: Duration = Duration::from_millis(250);
+const MAX_PARALLEL_STREAM_OPENS: usize = 2;
const MAX_TRANSIT_RESERVATIONS: usize = 32;
const MAX_TRANSIT_CIRCUITS: usize = 8;
const MAX_TRANSIT_CIRCUITS_PER_PEER: usize = 2;
@@ -268,7 +270,7 @@ struct PendingConnect {
result: Option<oneshot::Sender<Result<PeerStream, PeerError>>>,
stream_kind: StreamKind,
deadline: Instant,
- opening: Option<PendingStreamOpen>,
+ openings: Vec<PendingStreamOpen>,
rejected_connections: HashSet<ConnectionId>,
dials: HashMap<ConnectionId, DialOrigin>,
direct_routes: Vec<Multiaddr>,
@@ -298,6 +300,7 @@ struct WebRtcRelayAttempt {
struct PendingStreamOpen {
connection_id: ConnectionId,
+ started: Instant,
task: tokio::task::JoinHandle<()>,
}
@@ -917,7 +920,7 @@ async fn run_endpoint_async(
result: Some(result),
stream_kind,
deadline: Instant::now() + options.deadline,
- opening: None,
+ openings: Vec::new(),
rejected_connections: HashSet::new(),
dials: HashMap::new(),
direct_routes: started.direct_routes,
@@ -1270,15 +1273,12 @@ async fn run_endpoint_async(
continue;
}
};
- if waiter
- .opening
- .as_ref()
- .is_none_or(|opening| opening.connection_id != connection_id)
- {
+ let Some(opening_index) = waiter.openings.iter()
+ .position(|opening| opening.connection_id == connection_id) else {
direct.pending.insert(request_id, waiter);
continue;
- }
- waiter.opening.take();
+ };
+ waiter.openings.swap_remove(opening_index);
if !waiter.webrtc_attempted_relays.is_empty() {
webrtc_debug(format_args!(
"application stream result peer={} request={} success={}",
@@ -1537,6 +1537,14 @@ async fn run_endpoint_async(
let expired = direct.pending.iter()
.filter_map(|(request_id, item)| (item.result.is_some() && item.deadline <= now).then_some(*request_id))
.collect::<Vec<_>>();
+ // An established spare may precede the stalled open. Do not
+ // depend on another Swarm event to start its hedged attempt.
+ for request_id in direct.pending.keys().copied().collect::<Vec<_>>() {
+ maybe_open_peer_stream(
+ request_id, &mut direct.pending, &direct.retiring_connections,
+ &direct.health, stream_control.clone(), mesh_control.clone(), opened_tx.clone(),
+ );
+ }
for request_id in expired {
if let Some(waiter) = direct.pending.remove(&request_id) {
let error = pending_connect_deadline_error(&waiter);
@@ -2065,14 +2073,14 @@ fn retain_current_webrtc_relay_attempts(
_ => None,
})
.collect::<Vec<_>>();
- if waiter
- .opening
- .as_ref()
- .is_some_and(|opening| stale_dials.contains(&opening.connection_id))
- && let Some(opening) = waiter.opening.take()
- {
- opening.task.abort();
- }
+ waiter.openings.retain(|opening| {
+ if stale_dials.contains(&opening.connection_id) {
+ opening.task.abort();
+ false
+ } else {
+ true
+ }
+ });
for connection_id in &stale_dials {
remove_pending_dial(waiter, *connection_id);
retiring_connections.insert(*connection_id);
@@ -2229,13 +2237,22 @@ fn maybe_open_peer_stream(
.chain(waiter.transit_relay_peers.iter().copied())
.collect(),
};
- let excluded = retiring_connections
+ let mut excluded = retiring_connections
.union(&waiter.rejected_connections)
.copied()
.collect::<HashSet<_>>();
- if waiter.result.is_none() || waiter.opening.is_some() {
+ // Hedge only stream establishment, never application requests. Keep the
+ // original candidate alive: a slow but healthy path retains its full budget.
+ if waiter.result.is_none()
+ || waiter.openings.len() >= MAX_PARALLEL_STREAM_OPENS
+ || waiter
+ .openings
+ .iter()
+ .any(|opening| opening.started.elapsed() < STREAM_OPEN_HEDGE_DELAY)
+ {
return;
}
+ excluded.extend(waiter.openings.iter().map(|opening| opening.connection_id));
let stream_kind = waiter.stream_kind;
let candidates = match stream_kind {
StreamKind::Application => {
@@ -2261,8 +2278,9 @@ fn maybe_open_peer_stream(
StreamKind::Application => application_control,
StreamKind::MeshControl => mesh_control,
};
- waiter.opening = Some(PendingStreamOpen {
+ waiter.openings.push(PendingStreamOpen {
connection_id,
+ started: Instant::now(),
task: tokio::spawn(async move {
let result = tokio::select! {
_ = cancellation.cancelled() => return,
@@ -2410,14 +2428,14 @@ fn handle_swarm_event(
for connect in direct.pending.values_mut() {
remove_pending_dial(connect, connection_id);
connect.rejected_connections.remove(&connection_id);
- if connect
- .opening
- .as_ref()
- .is_some_and(|opening| opening.connection_id == connection_id)
- && let Some(opening) = connect.opening.take()
- {
- opening.task.abort();
- }
+ connect.openings.retain(|opening| {
+ if opening.connection_id == connection_id {
+ opening.task.abort();
+ false
+ } else {
+ true
+ }
+ });
}
direct.active.remove(&connection_id);
if let Some(relay) = coordination_relays.get_mut(&peer_id) {
@@ -2476,14 +2494,14 @@ fn handle_swarm_event(
for connect in direct.pending.values_mut() {
remove_pending_dial(connect, connection_id);
connect.rejected_connections.remove(&connection_id);
- if connect
- .opening
- .as_ref()
- .is_some_and(|opening| opening.connection_id == connection_id)
- && let Some(opening) = connect.opening.take()
- {
- opening.task.abort();
- }
+ connect.openings.retain(|opening| {
+ if opening.connection_id == connection_id {
+ opening.task.abort();
+ false
+ } else {
+ true
+ }
+ });
}
let mut failed_automatic = None;
for (peer_id, relay) in coordination_relays.iter_mut() {
@@ -2809,9 +2827,7 @@ fn reconcile_pending_transit_connects(
for connection_id in retired_webrtc_dials {
let _ = swarm.close_connection(connection_id);
}
- if let Some(opening) = waiter.opening.take() {
- opening.task.abort();
- }
+ abort_pending_openings(waiter);
retired_dials.extend(take_pending_dials_by_origin(waiter, DialOrigin::Transit));
waiter.next_route_attempt = now;
waiter.transit_after = now;
@@ -2890,7 +2906,7 @@ fn fail_pending_connect(
}
fn abort_pending_openings(waiter: &mut PendingConnect) {
- if let Some(opening) = waiter.opening.take() {
+ for opening in waiter.openings.drain(..) {
opening.task.abort();
}
}
@@ -3823,11 +3839,20 @@ fn retry_connect_routes(
if connect.next_route_attempt > now {
continue;
}
- let excluded = direct
+ let mut excluded = direct
.retiring_connections
.union(&connect.rejected_connections)
.copied()
.collect::<HashSet<_>>();
+ // A transport in the table is not proof that a new stream can make
+ // progress. A stalled open must not suppress dialing its alternatives.
+ excluded.extend(
+ connect
+ .openings
+ .iter()
+ .filter(|opening| now.duration_since(opening.started) >= STREAM_OPEN_HEDGE_DELAY)
+ .map(|opening| opening.connection_id),
+ );
let allowed_relays = if connect.result.is_some() {
connect.transit_relay_peers.clone()
} else {
@@ -3955,9 +3980,9 @@ fn retire_direct_dials(
.is_some_and(|health| health.rtt.is_some() && health.failures == 0)
|| direct.pending.values().any(|pending| {
pending
- .opening
- .as_ref()
- .is_some_and(|opening| opening.connection_id == connection_id)
+ .openings
+ .iter()
+ .any(|opening| opening.connection_id == connection_id)
})
{
continue;
@@ -4214,6 +4239,108 @@ mod tests {
std::fs::remove_dir_all(root).unwrap();
}
+ #[tokio::test]
+ async fn stalled_stream_open_hedges_without_cancelling_the_slow_path_or_using_unapproved_transit()
+ {
+ let peer_id = PeerId::random();
+ let relay = PeerId::random();
+ let (mut behaviour, control, _) = application_stream::Behaviour::new(
+ StreamProtocol::new(APPLICATION_PROTOCOL),
+ 8,
+ Some(Arc::new(RwLock::new(HashSet::new()))),
+ );
+ let mut handlers = Vec::new();
+ for (id, address) in [
+ (1, "/ip4/127.0.0.1/udp/1/quic-v1".to_owned()),
+ (2, "/ip4/127.0.0.1/tcp/1".to_owned()),
+ (3, format!("/ip4/127.0.0.1/tcp/2/p2p/{relay}/p2p-circuit")),
+ ] {
+ let address: Multiaddr = address.parse().unwrap();
+ handlers.push(
+ behaviour
+ .handle_established_inbound_connection(
+ ConnectionId::new_unchecked(id),
+ peer_id,
+ &address,
+ &address,
+ )
+ .unwrap(),
+ );
+ }
+ let now = Instant::now();
+ let (result, _response) = oneshot::channel();
+ let waiter = PendingConnect {
+ attempt_id: ConnectAttemptId(1),
+ peer_id,
+ result: Some(result),
+ stream_kind: StreamKind::Application,
+ deadline: now + Duration::from_secs(30),
+ openings: Vec::new(),
+ rejected_connections: HashSet::new(),
+ dials: HashMap::new(),
+ direct_routes: Vec::new(),
+ identified_routes: Vec::new(),
+ coordination_relays: Vec::new(),
+ coordination_relay_peers: Vec::new(),
+ transit_relay_peers: HashSet::new(),
+ transit_after: now,
+ next_route_attempt: now,
+ retry_coordination: false,
+ webrtc_attempted_relays: HashSet::new(),
+ next_webrtc_attempt_id: 0,
+ webrtc_active_attempt: None,
+ cancellation: CancellationToken::new(),
+ };
+ let mut pending = HashMap::from([(1, waiter)]);
+ let (opened, _results) = mpsc::channel(8);
+ let attempt = |pending: &mut HashMap<u32, PendingConnect>| {
+ maybe_open_peer_stream(
+ 1,
+ pending,
+ &HashSet::new(),
+ &HashMap::new(),
+ control.clone(),
+ control.clone(),
+ opened.clone(),
+ );
+ };
+ attempt(&mut pending);
+ assert_eq!(pending[&1].openings.len(), 1);
+ assert_eq!(
+ pending[&1].openings[0].connection_id,
+ ConnectionId::new_unchecked(1)
+ );
+ attempt(&mut pending);
+ assert_eq!(pending[&1].openings.len(), 1);
+ pending.get_mut(&1).unwrap().openings[0].started = now - STREAM_OPEN_HEDGE_DELAY;
+ attempt(&mut pending);
+ assert_eq!(pending[&1].openings.len(), 2);
+ assert_eq!(
+ pending[&1].openings[1].connection_id,
+ ConnectionId::new_unchecked(2)
+ );
+ for opening in &mut pending.get_mut(&1).unwrap().openings {
+ opening.started = now - STREAM_OPEN_HEDGE_DELAY;
+ }
+ attempt(&mut pending);
+ assert_eq!(pending[&1].openings.len(), 2);
+ assert!(
+ pending[&1]
+ .openings
+ .iter()
+ .all(|opening| !opening.task.is_finished())
+ );
+ // Once both direct paths are rejected, the unapproved relay cannot win.
+ abort_pending_openings(pending.get_mut(&1).unwrap());
+ pending.get_mut(&1).unwrap().rejected_connections.extend([
+ ConnectionId::new_unchecked(1),
+ ConnectionId::new_unchecked(2),
+ ]);
+ attempt(&mut pending);
+ assert!(pending[&1].openings.is_empty());
+ drop(handlers);
+ }
+
#[test]
fn application_streams_commit_on_the_best_established_path() {
let relay_peer_id = PeerId::random();
@@ -4271,7 +4398,7 @@ mod tests {
result: Some(result),
stream_kind: StreamKind::Application,
deadline: now,
- opening: None,
+ openings: Vec::new(),
rejected_connections: HashSet::new(),
dials: HashMap::new(),
direct_routes: Vec::new(),
@@ -4309,7 +4436,7 @@ mod tests {
result: Some(result),
stream_kind: StreamKind::Application,
deadline: now + Duration::from_secs(30),
- opening: None,
+ openings: Vec::new(),
rejected_connections: HashSet::new(),
dials: HashMap::new(),
direct_routes: Vec::new(),
@@ -4364,7 +4491,7 @@ mod tests {
result: Some(result),
stream_kind: StreamKind::Application,
deadline: now + Duration::from_secs(30),
- opening: None,
+ openings: Vec::new(),
rejected_connections: HashSet::new(),
dials: HashMap::new(),
direct_routes: Vec::new(),
@@ -4390,8 +4517,9 @@ mod tests {
waiter
.dials
.insert(removed_connection_id, DialOrigin::WebRtc(removed_attempt));
- waiter.opening = Some(PendingStreamOpen {
+ waiter.openings.push(PendingStreamOpen {
connection_id: removed_connection_id,
+ started: Instant::now(),
task: tokio::spawn(std::future::pending()),
});
waiter.coordination_relay_peers = vec![replacement_relay];
@@ -4403,7 +4531,7 @@ mod tests {
assert!(removed_cancellation.is_cancelled());
assert!(!is_current_webrtc_relay_attempt(&waiter, removed_attempt));
- assert!(waiter.opening.is_none());
+ assert!(waiter.openings.is_empty());
assert!(!waiter.dials.contains_key(&removed_connection_id));
assert!(retiring_connections.contains(&removed_connection_id));
let replacement = next_webrtc_relay_peer(
diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts
index 8c99248d68..5c3fd49970 100644
--- a/packages/runtime-host/src/__tests__/peer-listener.test.ts
+++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts
@@ -183,7 +183,7 @@ test('rechecks peer authority at admission after the authentication response is
await listener.cleanup();
});
-test('bounds active application streams from one authenticated peer', async () => {
+test('bounds active application streams from one authenticated principal', async () => {
const streams = Array.from({ length: 5 }, () => authenticatedPendingStream('remote-peer'));
let accepted = 0;
const listener = createRuntimeHostPeerListener(
@@ -205,6 +205,72 @@ test('bounds active application streams from one authenticated peer', async () =
await listener.cleanup();
});
+test('admits distinct Guest mounts on one Desktop while bounding principals, devices and the Host; resume spends no slot', async (t) => {
+ let admit!: (stream: RuntimeHostPeerNativeStream) => void;
+ const accepted: import('../server/listener-set.js').RuntimeHostListenerConnection[] = [];
+ const client = peerWith([]);
+ const listener = createRuntimeHostPeerListener(
+ {
+ ...client,
+ serveApplication: async (onStream, signal) => {
+ admit = onStream;
+ return client.serveApplication(onStream, signal);
+ },
+ },
+ UNUSED_REACHABILITY,
+ {
+ authenticate: (credential: string) => ({
+ principalKind: 'session_guest',
+ principalId: credential,
+ credentialId: credential,
+ operationGrants: [],
+ }),
+ subscribeRevocations: () => () => {},
+ } as never,
+ (connection) => {
+ accepted.push(connection);
+ },
+ );
+ t.after(() => listener.cleanup());
+ let sequence = 0;
+ const open = async (
+ principal: string,
+ peerId = 'desktop',
+ session = ++sequence,
+ generation = 1,
+ ) => {
+ const stream = authenticatedPendingStream(peerId, principal, {
+ sessionId: session.toString(16).padStart(64, '0'),
+ generation,
+ received: 0,
+ });
+ const writes: Buffer[] = [];
+ admit({
+ ...stream,
+ write: async (bytes) => {
+ writes.push(Buffer.from(bytes));
+ },
+ });
+ await waitForImmediate();
+ return JSON.parse(writes[0]!.toString()) as { accepted: boolean; reason?: string };
+ };
+ for (let i = 0; i < 128; i++) assert.equal((await open(`guest-${i}`)).accepted, true);
+ // Additional connections for one grant cannot bypass its quota by changing PeerId.
+ for (let i = 0; i < 3; i++) assert.equal((await open('guest-0', `device-${i}`)).accepted, true);
+ assert.equal((await open('guest-0', 'another-device')).reason, 'capacity_exceeded');
+ for (let i = 0; i < 32; i++) assert.equal((await open(`profile-${i}`)).accepted, true);
+ assert.equal((await open('over-device')).reason, 'capacity_exceeded');
+ // The 160 established Desktop connections remain authorized during reattach.
+ assert.equal((await open('guest-0', 'desktop', 1, 2)).accepted, true);
+ assert.equal(accepted.length, 163);
+ for (let i = 0; i < 93; i++)
+ assert.equal((await open(`other-${i}`, 'other-desktop')).accepted, true);
+ assert.equal((await open('over-host', 'third-desktop')).reason, 'capacity_exceeded');
+ accepted[0]!.transport.abort(new Error('mount removed'));
+ await waitForImmediate();
+ assert.equal((await open('replacement')).accepted, true);
+});
+
test('projects newly accepted coordination relays from the running peer endpoint', async () => {
let coordinationRelays: readonly string[] = [];
const peer = {
@@ -330,6 +396,8 @@ function pendingStream(
function authenticatedPendingStream(
peerId: string,
+ credential = 'accepted',
+ resume?: { sessionId: string; generation: number; received: number },
): RuntimeHostPeerNativeStream & { readonly aborted: boolean } {
let finish!: (value: null) => void;
const pending = new Promise<null>((resolve) => {
@@ -345,7 +413,9 @@ function authenticatedPendingStream(
read: async () => {
if (first) {
first = false;
- return Buffer.from('{"v":1,"credential":"accepted"}\n');
+ return Buffer.from(
+ `${JSON.stringify(resume ? { v: 2, credential, resume } : { v: 1, credential })}\n`,
+ );
}
return pending;
},
diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts
index f8c05f33de..844bb801a2 100644
--- a/packages/runtime-host/src/__tests__/peer-native.test.ts
+++ b/packages/runtime-host/src/__tests__/peer-native.test.ts
@@ -444,6 +444,25 @@ test('bounds and separates the peer credential preface from Runtime Host frames'
streamWith(Buffer.from('{"v":2,"accepted":true,"resume":{"received":65536}}\n')),
);
assert.equal(resumedResult.resume?.received, 65_536);
+ await assert.rejects(
+ readRuntimeHostPeerAuthenticationResult(
+ streamWith(Buffer.from('{"v":2,"accepted":false,"reason":"capacity_exceeded"}\n')),
+ ),
+ (error: unknown) =>
+ error instanceof RuntimeHostPeerError && error.code === 'peer_capacity_exceeded',
+ );
+ for (const invalid of [
+ { v: 2, accepted: true, reason: 'capacity_exceeded' },
+ { v: 2, accepted: false, reason: 'unknown' },
+ { v: 2, accepted: false, reason: 'capacity_exceeded', resume: { received: 0 } },
+ ]) {
+ await assert.rejects(
+ readRuntimeHostPeerAuthenticationResult(
+ streamWith(Buffer.from(`${JSON.stringify(invalid)}\n`)),
+ ),
+ /result is invalid/u,
+ );
+ }
for (const invalid of [
{ ...resume, generation: 0 },
{ ...resume, received: -1 },
diff --git a/packages/runtime-host/src/__tests__/peer-session-collaboration.test.ts b/packages/runtime-host/src/__tests__/peer-session-collaboration.test.ts
new file mode 100644
index 0000000000..8ab1409613
--- /dev/null
+++ b/packages/runtime-host/src/__tests__/peer-session-collaboration.test.ts
@@ -0,0 +1,323 @@
+/*
+ * 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 { connect as tcpConnect, createServer, type Socket } from 'node:net';
+import { once } from 'node:events';
+import { setTimeout as delay } from 'node:timers/promises';
+import test from 'node:test';
+import { FakeBackend } from '@maka/runtime/test-only/fake-backend';
+import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores';
+import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority';
+import { RuntimeHostKernel } from '../server/host-kernel.js';
+import { defineInteractiveRuntimeHostComposition } from '../server/host-composition.js';
+import { createExecutionRuntimeHostComposition } from '../server/execution-composition.js';
+import { startRuntimeHostAuthenticatedListenerSet } from '../server/listener-set.js';
+import { openRuntimeHostAccessAuthority } from '../server/access-authority.js';
+import { connectRuntimeHost, type RuntimeHostConnection } from '../client/connection.js';
+import { connectPeerRuntimeHost } from '../client/host-profile.js';
+import type { RuntimeHostPeerClient } from '../client/peer-client.js';
+import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js';
+import {
+ RUNTIME_HOST_PROTOCOL_VERSION,
+ decodeCollaborationInvitationCode,
+ type SubscriptionFrame,
+} from '../protocol/index.js';
+
+test('production collaboration retains distinct Guest mounts, exact requests and subscriptions across lost replies and Host restart', {
+ timeout: 60_000,
+}, async (t) => {
+ const base = await mkdtemp(join(tmpdir(), 'maka-peer-collaboration-'));
+ const root = join(base, 'root');
+ const capability = await resolveStorageRoot({ path: root, kind: 'interactive' });
+ const sockets = new Set<Socket>();
+ const executions = new Map<string, number>();
+ const guests: RuntimeHostConnection[] = [];
+ const connections = new Set<RuntimeHostConnection>();
+ let acceptStream: (stream: RuntimeHostPeerNativeStream) => void = (stream) => stream.abort();
+ let host: RuntimeHostKernel | undefined;
+ let local: RuntimeHostConnection | undefined;
+ let dropCommittedReply = false;
+ const disconnectPaths = () => {
+ for (const socket of sockets) socket.destroy();
+ };
+ const wire = (socket: Socket, peerId: string): RuntimeHostPeerNativeStream => {
+ sockets.add(socket);
+ socket.once('close', () => sockets.delete(socket));
+ socket.on('error', () => {});
+ const reader = socket[Symbol.asyncIterator]();
+ return {
+ peerId,
+ path: { kind: 'direct', transport: 'tcp' },
+ read: async () => {
+ const next = await reader.next();
+ return next.done ? null : Buffer.from(next.value);
+ },
+ write: (bytes) =>
+ new Promise<void>((resolve, reject) => {
+ socket.write(bytes, (error) => (error ? reject(error) : resolve()));
+ }),
+ close: async () => {
+ socket.destroy();
+ },
+ abort: () => {
+ socket.destroy();
+ },
+ };
+ };
+ const server = createServer((socket) => acceptStream(wire(socket, 'desktop')));
+ server.listen(0, '127.0.0.1');
+ await once(server, 'listening');
+ const address = server.address();
+ assert.ok(address && typeof address !== 'string');
+ const reachability = {
+ lease: {
+ version: 1 as const,
+ peerId: 'host',
+ revision: 1,
+ issuedAt: Date.now(),
+ expiresAt: Date.now() + 300_000,
+ directRoutes: [],
+ coordinationRoutes: [],
+ },
+ publicKey: 'AA',
+ signature: 'AA',
+ };
+ const peer = {
+ identity: () => ({ peerId: 'host' }),
+ observeAuthenticatedReachability: () => reachability,
+ serveApplication: async (accept: typeof acceptStream, signal: AbortSignal) => {
+ acceptStream = accept;
+ await new Promise<void>((resolve) =>
+ signal.addEventListener('abort', () => resolve(), { once: true }),
+ );
+ },
+ connect: async () => {
+ const socket = tcpConnect(address.port, '127.0.0.1');
+ await once(socket, 'connect');
+ return wire(socket, 'host');
+ },
+ } as unknown as RuntimeHostPeerClient;
+ const connectGuest = async (credential: string) => {
+ const connection = await connectPeerRuntimeHost({
+ profileId: 'shared',
+ transport: { kind: 'libp2p-direct', reachability },
+ credential,
+ expectedRootId: capability.rootId,
+ clientInstanceId: 'desktop',
+ peerClient: peer,
+ });
+ connections.add(connection);
+ return connection;
+ };
+ const startHost = async (seed: boolean) => {
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ const sessionIds: string[] = [];
+ if (seed) {
+ const stores = await openInteractiveExecutionStoresForWrite(owner.lease);
+ for (let i = 0; i < 6; i++) {
+ const session = await stores.sessionStore.create({
+ cwd: root,
+ llmConnectionId: 'cccccccc-cccc-4ccc-8ccc-cccccccccccc',
+ llmConnectionSlug: 'test-provider',
+ model: 'test-model',
+ permissionMode: 'explore',
+ });
+ sessionIds.push(session.id);
+ }
+ }
+ const authority = await openRuntimeHostAccessAuthority(owner.controlDirectory);
+ const createRequest = authority.createTurnAccessRequest.bind(authority);
+ authority.createTurnAccessRequest = async (...args) => {
+ const committed = await createRequest(...args);
+ if (dropCommittedReply) {
+ dropCommittedReply = false;
+ disconnectPaths();
+ }
+ return committed;
+ };
+ host = await RuntimeHostKernel.start({
+ owner,
+ lifecycleMode: 'service',
+ accessAuthority: authority,
+ composition: defineInteractiveRuntimeHostComposition((context) =>
+ createExecutionRuntimeHostComposition(
+ context,
+ {},
+ {
+ primaryBackendFactory: (backendContext) => {
+ const backend = new FakeBackend(backendContext);
+ const send = backend.send.bind(backend);
+ backend.send = async function* (input) {
+ executions.set(input.turnId, (executions.get(input.turnId) ?? 0) + 1);
+ yield* send(input);
+ };
+ return backend;
+ },
+ },
+ ),
+ ),
+ listenerSetFactory: (input) =>
+ startRuntimeHostAuthenticatedListenerSet(input, {
+ peer: {
+ client: peer,
+ reachability: { current: () => reachability } as never,
+ accessAuthority: authority,
+ },
+ }),
+ });
+ const connected = await connectRuntimeHost({
+ rootPath: root,
+ protocol: { min: RUNTIME_HOST_PROTOCOL_VERSION, max: RUNTIME_HOST_PROTOCOL_VERSION },
+ });
+ assert.equal(connected.kind, 'connected');
+ if (connected.kind !== 'connected') throw new Error('Owner did not connect');
+ local = connected.connection;
+ return sessionIds;
+ };
+ t.after(async () => {
+ await Promise.allSettled([...connections].map((guest) => guest.close()));
+ await local?.close();
+ await host?.close();
+ disconnectPaths();
+ await new Promise<void>((resolve) => server.close(() => resolve()));
+ await rm(base, { recursive: true, force: true });
+ });
+ const sessionIds = await startHost(true);
+ const invitations = [];
+ for (const sessionId of sessionIds) {
+ const invitation = await local!.request('collaboration.invitation.prepare', {
+ sessionId,
+ grantKinds: ['session_observation', 'session_turn_request'],
+ });
+ invitations.push(invitation);
+ const { credential } = decodeCollaborationInvitationCode(invitation.invitationCode);
+ const pending = await connectGuest(credential);
+ await pending.request('access.credential.finalize', {});
+ // A dropped pending path can still be retained when the finalized Client
+ // reconnects. Exercise that overlap, not only a fully retired predecessor.
+ const guest = await connectGuest(credential);
+ guests.push(guest);
+ assert.equal((await guest.request('session.shared.query', {})).session?.id, sessionId);
+ await pending.close();
+ }
+ assert.equal(host!.connectionCount, 7);
+ const guest = guests[0]!;
+ const connectionId = guest.connectionId;
+ const subscription = await guest.openSessionSubscription({
+ sessionId: sessionIds[0]!,
+ transcript: { kind: 'none' },
+ });
+ const frames: SubscriptionFrame[] = [];
+ const events = (async () => {
+ for await (const frame of subscription) frames.push(frame);
+ })();
+ const intent = {
+ sessionId: sessionIds[0]!,
+ turnId: 'shared-turn',
+ content: { text: 'hello collaboration' },
+ };
+ dropCommittedReply = true;
+ const request = await guest.request('collaboration.turn-request.create', { intent }, 10_000);
+ assert.equal(guest.connectionId, connectionId);
+ assert.equal(
+ (await guest.request('collaboration.turn-request.create', { intent })).requestId,
+ request.requestId,
+ );
+ assert.equal((await guest.request('collaboration.turn-request.query', {})).requests.length, 1);
+ assert.equal(
+ (await guests[1]!.request('collaboration.turn-request.query', {})).requests.length,
+ 0,
+ );
+ await local!.request('collaboration.turn-request.decide', {
+ requestId: request.requestId,
+ decision: 'approve',
+ });
+ disconnectPaths();
+ await until(async () => {
+ const turn = await local!.request('turn.query', {
+ sessionId: intent.sessionId,
+ turnId: intent.turnId,
+ });
+ return turn.status === 'completed';
+ });
+ assert.equal(executions.get(intent.turnId), 1);
+ await until(async () => frames.some((frame) => frame.kind === 'subscription.session_delta'));
+ // Revoke while the observation's raw connection is gone: replay cannot
+ // reopen the subscription or cross the current Session authority.
+ disconnectPaths();
+ await local!.request('collaboration.grant.revoke', {
+ grantId: invitations[0]!.grants.find((grant) => grant.kind === 'session_observation')!.grantId,
+ });
+ await events;
+ assert.ok(
+ frames.some(
+ (frame) => frame.kind === 'subscription.closed' && frame.reason === 'access_revoked',
+ ),
+ );
+ await assert.rejects(
+ guest.openSessionSubscription({ sessionId: intent.sessionId, transcript: { kind: 'none' } }),
+ );
+ await Promise.all(guests.map((connection) => connection.close()));
+ guests.length = 0;
+ await local!.close();
+ local = undefined;
+ const epoch = host!.hostEpoch;
+ await host!.close();
+ host = undefined;
+ await startHost(false);
+ assert.notEqual(host!.hostEpoch, epoch);
+ const restored = await connectGuest(
+ decodeCollaborationInvitationCode(invitations[0]!.invitationCode).credential,
+ );
+ guests.push(restored);
+ const recovered = await restored.request('collaboration.turn-request.query', {});
+ assert.equal(recovered.requests.length, 1);
+ assert.equal(recovered.requests[0]!.requestId, request.requestId);
+ assert.deepEqual(
+ (
+ await local!.request('collaboration.turn-request.decide', {
+ requestId: request.requestId,
+ decision: 'approve',
+ })
+ ).kind,
+ 'already_decided',
+ );
+ assert.equal(executions.get(intent.turnId), 1);
+ const closed = restored.closed;
+ await local!.request('collaboration.principal.revoke', {
+ principalId: invitations[0]!.principalId,
+ });
+ await closed;
+ await assert.rejects(
+ connectGuest(decodeCollaborationInvitationCode(invitations[0]!.invitationCode).credential),
+ );
+});
+
+async function until(check: () => Promise<boolean>): Promise<void> {
+ for (let i = 0; i < 300; i++) {
+ if (await check()) return;
+ await delay(20);
+ }
+ throw new Error('Collaboration projection did not converge');
+}
diff --git a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
index 29857ebee7..7aed4c5582 100644
--- a/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
+++ b/packages/runtime-host/src/__tests__/reconnecting-connection.test.ts
@@ -547,6 +547,70 @@ test('an initial transient failure keeps a wakeable reconnect lifecycle', async
}
});
+test('initial retry policy reports a rejected admission without retrying', async () => {
+ const capacity = new Error('capacity exceeded');
+ let attempts = 0;
+ await assert.rejects(
+ startRuntimeHostReconnectLifecycle({
+ retryInitialFailure: (error) => error !== capacity,
+ connect: async (): Promise<RuntimeHostConnection> => {
+ attempts += 1;
+ throw capacity;
+ },
+ }),
+ (error: unknown) => error === capacity,
+ );
+ assert.equal(attempts, 1);
+});
+
+test('initial retry policy also stops a later initial admission attempt', async () => {
+ const capacity = new Error('capacity exceeded');
+ let attempts = 0;
+ const lifecycle = await startRuntimeHostReconnectLifecycle({
+ retryInitialFailure: (error) => error !== capacity,
+ connect: async (): Promise<RuntimeHostConnection> => {
+ attempts += 1;
+ if (attempts === 1) throw new Error('route unavailable');
+ throw capacity;
+ },
+ backoff: { minMs: 0, maxMs: 0 },
+ });
+ try {
+ await assert.rejects(lifecycle.waitForCurrent(), (error: unknown) => error === capacity);
+ assert.equal(attempts, 2);
+ } finally {
+ await lifecycle.close();
+ }
+});
+
+test('initial retry policy does not reject recovery of an established connection', async () => {
+ const first = connectionHarness('first', () => undefined);
+ const replacement = connectionHarness('replacement', () => undefined);
+ let attempts = 0;
+ let policyCalls = 0;
+ const lifecycle = await startRuntimeHostReconnectLifecycle({
+ retryInitialFailure: () => {
+ policyCalls += 1;
+ return false;
+ },
+ connect: async () => {
+ attempts += 1;
+ if (attempts === 1) return first.connection;
+ if (attempts === 2) throw new Error('capacity exceeded');
+ return replacement.connection;
+ },
+ backoff: { minMs: 0, maxMs: 0 },
+ });
+ try {
+ first.disconnect();
+ assert.equal(await lifecycle.waitForCurrent(first.connection), replacement.connection);
+ assert.equal(attempts, 3);
+ assert.equal(policyCalls, 0);
+ } finally {
+ await lifecycle.close();
+ }
+});
+
test('initial retry mode does not outlive caller cancellation', async () => {
const controller = new AbortController();
const connecting = deferred();
diff --git a/packages/runtime-host/src/client/connection.ts b/packages/runtime-host/src/client/connection.ts
index 86bbd01d38..4bd55a03e0 100644
--- a/packages/runtime-host/src/client/connection.ts
+++ b/packages/runtime-host/src/client/connection.ts
@@ -638,9 +638,24 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
}
async close(): Promise<void> {
- this.#clientCapabilities.close(new Error('Runtime Host connection closed by Client'));
- this.#transport.abort();
- await this.closed;
+ if (!this.peerPath) {
+ this.#clientCapabilities.close(new Error('Runtime Host connection closed by Client'));
+ this.#transport.abort();
+ await this.closed;
+ return;
+ }
+ // An intentional peer close must send logical FIN. Aborting its raw path
+ // instead leaves the Host retaining a recoverable session and its quota.
+ this.#fail(
+ new RuntimeHostTransportError('closed', 'Runtime Host connection closed by Client'),
+ true,
+ );
+ const deadline = setTimeout(() => this.#transport.abort(), 5_000);
+ try {
+ await this.closed;
+ } finally {
+ clearTimeout(deadline);
+ }
}
async replaceClientCapabilities(
@@ -946,7 +961,7 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
).catch((failure: unknown) => this.#fail(asError(failure)));
}
- #fail(error: Error): void {
+ #fail(error: Error, gracefulPeerClose = false): void {
if (this.#terminalError) return;
this.#terminalError = error;
if (this.#livenessTimer) clearTimeout(this.#livenessTimer);
@@ -983,7 +998,8 @@ class RuntimeHostConnectionImpl implements RuntimeHostConnection {
this.#configurationChangeListeners.clear();
this.#sessionCatalogChangeListeners.clear();
this.#scheduledTaskChangeListeners.clear();
- this.#transport.abort();
+ if (gracefulPeerClose) this.#transport.closeAfterFlush();
+ else this.#transport.abort();
}
}
diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts
index 35a6c05746..c5e727f15d 100644
--- a/packages/runtime-host/src/client/index.ts
+++ b/packages/runtime-host/src/client/index.ts
@@ -187,7 +187,7 @@ export {
type IssueRuntimeHostOwnerConnectionCodeInput,
type RuntimeHostOwnerConnectionCode,
} from './owner-connection-code.js';
-export { ensureRuntimeHostPeerIdentity } from '../transport/peer-native.js';
+export { ensureRuntimeHostPeerIdentity, RuntimeHostPeerError } from '../transport/peer-native.js';
export {
createRuntimeHostPeerClient,
createRuntimeHostPeerClientFromEnvironment,
diff --git a/packages/runtime-host/src/client/reconnect-lifecycle.ts b/packages/runtime-host/src/client/reconnect-lifecycle.ts
index c6cab25a49..006f41a7ec 100644
--- a/packages/runtime-host/src/client/reconnect-lifecycle.ts
+++ b/packages/runtime-host/src/client/reconnect-lifecycle.ts
@@ -77,7 +77,8 @@ export async function startRuntimeHostReconnectLifecycle<
>(input: {
readonly initial?: T;
readonly connect: (signal: AbortSignal) => Promise<T>;
- readonly retryInitialFailure?: boolean;
+ /** A predicate applies until the first successful connection, not to later reconnects. */
+ readonly retryInitialFailure?: boolean | ((error: Error) => boolean);
readonly initialSignal?: AbortSignal;
readonly onReconnectError?: (error: Error) => void;
readonly onFatalError?: (error: Error) => void;
@@ -101,7 +102,8 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
readonly closed: Promise<void>;
readonly #initial: T | undefined;
readonly #connect: (signal: AbortSignal) => Promise<T>;
- readonly #retryInitialFailure: boolean;
+ readonly #retryInitialFailure: boolean | ((error: Error) => boolean);
+ #connectedOnce = false;
readonly #initialSignal: AbortSignal | undefined;
readonly #onReconnectError: ((error: Error) => void) | undefined;
readonly #onFatalError: ((error: Error) => void) | undefined;
@@ -132,7 +134,7 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
constructor(input: {
readonly initial?: T;
readonly connect: (signal: AbortSignal) => Promise<T>;
- readonly retryInitialFailure?: boolean;
+ readonly retryInitialFailure?: boolean | ((error: Error) => boolean);
readonly initialSignal?: AbortSignal;
readonly onReconnectError?: (error: Error) => void;
readonly onFatalError?: (error: Error) => void;
@@ -185,7 +187,9 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
} catch (error) {
const failure = asError(error);
if (
- this.#retryInitialFailure &&
+ (typeof this.#retryInitialFailure === 'function'
+ ? this.#retryInitialFailure(failure)
+ : this.#retryInitialFailure) &&
!this.#closed &&
!this.#abort.signal.aborted &&
!this.#initialSignal?.aborted &&
@@ -299,6 +303,7 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
return;
}
this.#installedAt = this.#now();
+ this.#connectedOnce = true;
this.#setCurrent(resource);
void resource.closed.then(
() => this.#disconnected(resource),
@@ -362,7 +367,12 @@ class RuntimeHostReconnectLifecycleImpl<T extends RuntimeHostReconnectResource>
} catch (error) {
if (this.#closed || this.#quiesced || signal.aborted) return;
const failure = asError(error);
- if (failure instanceof RuntimeHostPermanentReconnectError) {
+ if (
+ failure instanceof RuntimeHostPermanentReconnectError ||
+ (!this.#connectedOnce &&
+ typeof this.#retryInitialFailure === 'function' &&
+ !this.#retryInitialFailure(failure))
+ ) {
this.#failPermanently(failure);
return;
}
diff --git a/packages/runtime-host/src/server/connection-session.ts b/packages/runtime-host/src/server/connection-session.ts
index 9d5141a7ed..2c7190c021 100644
--- a/packages/runtime-host/src/server/connection-session.ts
+++ b/packages/runtime-host/src/server/connection-session.ts
@@ -287,6 +287,11 @@ export class RuntimeHostConnectionSession {
#ensureClientCapabilities(): ClientCapabilityConnection | undefined {
if (this.#closed || this.#inputClosed) return;
+ // Guests observe and submit approval requests; they cannot provide Client
+ // Capabilities or directly admit a Turn. In particular, their pending and
+ // finalized connections may overlap while the credential becomes bound to
+ // the Client. Neither connection owns a capability-provider registration.
+ if (this.#options.connection.authority.principalKind === 'session_guest') return;
const service = this.#options.resolveClientCapabilities?.();
if (!service) return;
if (this.#clientCapabilityService && this.#clientCapabilityService !== service) {
diff --git a/packages/runtime-host/src/server/peer-listener.ts b/packages/runtime-host/src/server/peer-listener.ts
index c8b6e8bcd5..7985ad2811 100644
--- a/packages/runtime-host/src/server/peer-listener.ts
+++ b/packages/runtime-host/src/server/peer-listener.ts
@@ -39,8 +39,11 @@ import type {
} from './listener-set.js';
const MAX_PENDING_AUTHENTICATIONS = 16;
-const MAX_ACTIVE_STREAMS = 64;
-const MAX_ACTIVE_STREAMS_PER_PEER = 4;
+// A Desktop shares one PeerId across 128 Guest mounts and 32 Host profiles.
+// Authentication, not a transport identity, determines the small abuse quota.
+const MAX_ACTIVE_STREAMS = 256;
+const MAX_ACTIVE_STREAMS_PER_PEER = 160;
+const MAX_ACTIVE_STREAMS_PER_PRINCIPAL = 4;
export interface RuntimeHostPeerListenerConfiguration {
readonly nativePath: string;
@@ -89,7 +92,7 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract {
readonly #accessAuthority: RuntimeHostAccessAuthority;
readonly #accept: (connection: RuntimeHostListenerConnection) => void;
readonly #transports = new Set<FramedByteStreamTransport>();
- readonly #streams = new Set<RuntimeHostPeerNativeStream>();
+ readonly #streams = new Map<RuntimeHostPeerNativeStream, RuntimeHostConnectionAuthority>();
readonly #sessions = new Map<
string,
{ stream: ResumablePeerStream; authority: RuntimeHostConnectionAuthority }
@@ -217,11 +220,29 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract {
stream.abort();
return;
}
- const peerStreams = [...this.#streams].filter(
- (admitted) => admitted.peerId === stream.peerId,
- ).length;
- if (this.#streams.size >= MAX_ACTIVE_STREAMS || peerStreams >= MAX_ACTIVE_STREAMS_PER_PEER) {
- stream.abort();
+ let peerStreams = 0;
+ let principalStreams = 0;
+ for (const [admitted, owner] of this.#streams) {
+ if (admitted.peerId === stream.peerId) peerStreams++;
+ if (
+ owner.principalKind === authority.principalKind &&
+ owner.principalId === authority.principalId
+ )
+ principalStreams++;
+ }
+ if (
+ this.#streams.size >= MAX_ACTIVE_STREAMS ||
+ peerStreams >= MAX_ACTIVE_STREAMS_PER_PEER ||
+ principalStreams >= MAX_ACTIVE_STREAMS_PER_PRINCIPAL
+ ) {
+ // Legacy peers cannot distinguish capacity from credential rejection.
+ // Keep their previous EOF behavior, but give v2 clients a typed result.
+ if (resume) {
+ await writeRuntimeHostPeerAuthenticationResult(stream, false, {
+ reason: 'capacity_exceeded',
+ });
+ await stream.close();
+ } else stream.abort();
return;
}
const logical = resume
@@ -236,7 +257,7 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract {
});
}
const admittedStream = logical ?? stream;
- this.#streams.add(admittedStream);
+ this.#streams.set(admittedStream, authority);
let accepted = false;
try {
await writeRuntimeHostPeerAuthenticationResult(
diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts
index c631517d75..e8313d104b 100644
--- a/packages/runtime-host/src/transport/peer-native.ts
+++ b/packages/runtime-host/src/transport/peer-native.ts
@@ -35,6 +35,7 @@ export type RuntimeHostPeerErrorCode =
| 'transit_unavailable'
| 'peer_native_unavailable'
| 'peer_native_failed'
+ | 'peer_capacity_exceeded'
| 'peer_connect_in_progress';
export class RuntimeHostPeerError extends Error {
@@ -321,12 +322,18 @@ export async function writeRuntimeHostPeerAuthentication(
export async function writeRuntimeHostPeerAuthenticationResult(
stream: RuntimeHostPeerNativeStream,
accepted: boolean,
- resume?: { readonly received: number },
+ detail?: { readonly received: number } | { readonly reason: 'capacity_exceeded' },
): Promise<void> {
await withStreamDeadline(
stream.write(
Buffer.from(
- `${JSON.stringify(resume ? { v: 2, accepted, resume } : { v: 1, accepted })}\n`,
+ `${JSON.stringify(
+ detail && 'reason' in detail
+ ? { v: 2, accepted: false, reason: detail.reason }
+ : detail
+ ? { v: 2, accepted, resume: detail }
+ : { v: 1, accepted },
+ )}\n`,
'utf8',
),
),
@@ -375,6 +382,12 @@ export async function readRuntimeHostPeerAuthenticationResult(
if (!isAuthenticationResult(decoded.value)) {
throw new RuntimeHostPeerError('peer_native_failed', 'Peer authentication result is invalid');
}
+ if ('reason' in decoded.value) {
+ throw new RuntimeHostPeerError(
+ 'peer_capacity_exceeded',
+ 'Runtime Host peer connection capacity is full; close unused Host or shared Session connections and retry',
+ );
+ }
return {
accepted: decoded.value.accepted,
remainder: decoded.remainder,
@@ -687,7 +700,21 @@ function isAuthenticationResult(
value: unknown,
): value is
| { v: 1; accepted: boolean }
- | { v: 2; accepted: boolean; resume: { readonly received: number } } {
+ | { v: 2; accepted: boolean; resume: { readonly received: number } }
+ | { v: 2; accepted: false; reason: 'capacity_exceeded' } {
+ if (
+ typeof value === 'object' &&
+ value !== null &&
+ !Array.isArray(value) &&
+ Object.keys(value).length === 3 &&
+ 'v' in value &&
+ value.v === 2 &&
+ 'accepted' in value &&
+ value.accepted === false &&
+ 'reason' in value &&
+ value.reason === 'capacity_exceeded'
+ )
+ return true;
return (
typeof value === 'object' &&
value !== null &&
@@ -738,6 +765,7 @@ function isPeerErrorCode(value: string | undefined): value is RuntimeHostPeerErr
value === 'transit_unavailable' ||
value === 'peer_native_unavailable' ||
value === 'peer_native_failed' ||
+ value === 'peer_capacity_exceeded' ||
value === 'peer_connect_in_progress'
);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org