fix(core): bring main back under the line ceiling and the socket rule

packages/core/src/contacts/store.ts

@@ -32,6 +32,7 @@ import { joinContact, joinRequest } from './join.ts';
 import { evaluatePolicy, TailscaleClient } from './tailscale.ts';
 import { attestedTier, checkTier, modelTier, tierAllowed } from './tier.ts';
 import { fingerprintWords } from './words.ts';
+import { TextSocket } from '../ws-client.ts';
 
 export { CONTACT_JOIN_ROUTE } from './join.ts';
 export const CONTACT_LINK_ROUTE = '/coordination/contact/link';
@@ -51,6 +52,8 @@ export const REPLY_WINDOW_MS = 7 * 24 * 3_600_000;
 const SKEW_MS = 60_000;
 const HELLO_INPUT = 'boite-contact-hello\n';
 const TICK_MS = 1000;
+/** How long the upgrade to a host may take before the link counts as failed. */
+const LINK_DIAL_MS = 8000;
 /** Every this many ticks, what waits to be told to a contact is offered again. */
 const FLUSH_TICKS = 30;
 const TAILSCALE_CACHE_MS = 60_000;
@@ -82,7 +85,7 @@ export class ContactStore {
   /** The origin an invitation gives: this machine's tailnet name. Tests point it at loopback. */
   selfUrl: () => Promise<string> = () => this.tailnetUrl();
   /** Guest side: the link this core keeps to each host. */
-  private readonly dialed = new Map<string, { socket: WebSocket; state: 'connecting' | 'connected' }>();
+  private readonly dialed = new Map<string, { socket: TextSocket | null; state: 'connecting' | 'connected' }>();
   /** Host side: the link each guest keeps to this core, by the guest's id. */
   private readonly hosted = new Map<string, { routeId: string; socket: LinkSocket }>();
   /** Join and hello nonces already used, until they would be stale anyway. */
@@ -784,49 +787,44 @@ export class ContactStore {
       'x-boite-hello': Buffer.from(hello).toString('base64url'),
       'x-boite-signature': this.core.coordination.signature(Buffer.from(`${HELLO_INPUT}${hello}`)).toString('base64'),
     };
-    let socket: WebSocket;
-    try {
-      // Bun's client takes headers; the hello rides in them, before any frame.
-      socket = new WebSocket(`${contact.url.replace(/^http/, 'ws')}${CONTACT_LINK_ROUTE}`, { headers } as unknown as string[]);
-    } catch {
-      // A link that fails counts whatever was heard before: the mark is now.
-      this.backoff.failed(contact.coreId, 'link', this.backoff.mark());
-      return;
-    }
     const coreId = contact.coreId;
-    const link = { socket, state: 'connecting' as 'connecting' | 'connected' };
+    const link: { socket: TextSocket | null; state: 'connecting' | 'connected' } = { socket: null, state: 'connecting' };
     this.dialed.set(coreId, link);
-    socket.addEventListener('open', () => {
+    // A link that fails counts whatever was heard before: the mark is now.
+    const lost = () => {
+      if (this.dialed.get(coreId) !== link) return;
+      this.dialed.delete(coreId);
+      this.backoff.failed(coreId, 'link', this.backoff.mark());
+      if (!this.closed) this.changed();
+    };
+    // The hello rides in the upgrade's headers, before any frame.
+    TextSocket.connect(`${contact.url.replace(/^http/, 'ws')}${CONTACT_LINK_ROUTE}`, LINK_DIAL_MS, { headers }).then(socket => {
       if (this.closed || this.dialed.get(coreId) !== link) { socket.close(1000); return; }
+      link.socket = socket;
       link.state = 'connected';
       this.backoff.heard(coreId);
       if (this.has(coreId)) this.mutate(coreId, entry => { entry.lastSeenAt = Date.now(); });
       this.flush(coreId);
-    });
-    socket.addEventListener('message', event => { this.background(this.serve(coreId, socket, event.data)); });
-    socket.addEventListener('close', () => {
-      if (this.dialed.get(coreId) !== link) return;
-      this.dialed.delete(coreId);
-      this.backoff.failed(coreId, 'link', this.backoff.mark());
-      if (!this.closed) this.changed();
-    });
+      socket.onMessage(text => { this.background(this.serve(coreId, socket, text)); });
+      socket.onClose(lost);
+    }, lost);
   }
 
   /** A request of the host, carried down the link: answered by the same handler as one posted to this core. */
-  private async serve(hostId: string, socket: WebSocket, data: unknown): Promise<void> {
+  private async serve(hostId: string, socket: TextSocket, data: string): Promise<void> {
     let frame: Partial<Extract<Frame, { t: 'req' }>>;
-    try { frame = JSON.parse(typeof data === 'string' ? data : Buffer.from(data as ArrayBuffer).toString('utf8')) as typeof frame; } catch { return; }
+    try { frame = JSON.parse(data) as typeof frame; } catch { return; }
     if (frame?.t !== 'req' || typeof frame.id !== 'string' || typeof frame.body !== 'string' || typeof frame.signature !== 'string') return;
     let response: CoordinationBridgeResponse;
     try { response = await this.core.coordination.forward({ coreId: hostId, body: frame.body, signature: frame.signature }); }
     catch { response = { status: 400, body: '', signature: '' }; }
-    if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ t: 'res', id: frame.id, ...response } satisfies Frame));
+    if (socket.open) socket.send(JSON.stringify({ t: 'res', id: frame.id, ...response } satisfies Frame));
   }
 
   /** Closes this core's side of a contact's link. */
   unlink(coreId: string): void {
     const dialed = this.dialed.get(coreId);
-    if (dialed) { this.dialed.delete(coreId); dialed.socket.close(1000, 'contact removed'); }
+    if (dialed) { this.dialed.delete(coreId); dialed.socket?.close(1000, 'contact removed'); }
     const hosted = this.hosted.get(coreId);
     if (hosted) {
       this.hosted.delete(coreId);
@@ -881,7 +879,7 @@ export class ContactStore {
     this.closed = true;
     if (this.timer !== null) clearInterval(this.timer);
     this.timer = null;
-    for (const { socket } of this.dialed.values()) socket.close(1001, 'core stopping');
+    for (const { socket } of this.dialed.values()) socket?.close(1001, 'core stopping');
     this.dialed.clear();
     for (const { socket } of this.hosted.values()) socket.close(1001, 'core stopping');
     this.hosted.clear();

packages/core/src/group.ts

@@ -29,18 +29,19 @@ import { hostname } from 'node:os';
 import { join as joinPath } from 'node:path';
 import { GRANT_TTL_MS, GROUP_MAX_CORES, GROUP_TICKET_TTL_MS, PAIRING_ROLES } from '@boite/contracts';
 import { deviceAdmitInput, deviceHelloInput, type CoordinationPeer, type DeviceHello, type DeviceInvite, type Group, type GroupDevice, type GroupInvite, type GroupTicket, type PairingRole, type Principal } from '@boite/contracts';
-import { boundedBody, MAX_BODY, PeerGone } from './coordination-wire.ts';
+import { boundedBody, PeerGone } from './coordination-wire.ts';
 import type { Core } from './core.ts';
 import { invalidParams, messageOf, refused, RpcFailure, unauthorized, unavailable } from './errors.ts';
 import { advertisedAddresses, probeTailnet, tailnetAddress, ticketAddresses, type Tailnet } from './group/addresses.ts';
-import { admissionInput, checkAddress, checkCard, checkRoster, clockOf, CORE_ENTRIES_MAX, DEVICE_ENTRIES_MAX, digestOf, fingerprint, homeOf, liveCores, liveDeviceKeys, liveDevices, memberOrigins, mergeRosters, groupName, peerOf, reachOf } from './group/roster.ts';
+import { admissionInput, checkAddress, checkCard, checkRoster, clockOf, CORE_ENTRIES_MAX, DEVICE_ENTRIES_MAX, digestOf, homeOf, liveCores, liveDeviceKeys, liveDevices, memberOrigins, mergeRosters, groupName, peerOf, reachOf } from './group/roster.ts';
 import { deviceAdmissionInput, deviceId, deviceKey, deviceSigned, SERVED_ORIGINS_MAX, type DeviceKeyEntry } from './group/device-keys.ts';
 import type { CoreCard, CoreEntry, DeviceEntry, Roster } from './group/roster.ts';
-import { boxPublic, boxSigningInput, newBoxKey, open, openResponse, pack, readBoxKey, seal, SEALED, sealedName, sealResponse, unpack } from './group/seal.ts';
+import { boxPublic, boxSigningInput, newBoxKey, open, pack, readBoxKey, seal, SEALED, sealedName, sealResponse, unpack } from './group/seal.ts';
 import type { Sealing } from './group/seal.ts';
 import { encodeInvite, encodeTicket, inviteOf, parseInvite, parseTicket, ticketSigningInput, inviteId, invitePsk } from './group/ticket.ts';
 import type { Invite } from './group/ticket.ts';
 import { DEVICES_MAX, GroupClients } from './group/clients.ts';
+import { askToJoin, JOIN_SIGNING_PREFIX, type JoinReply } from './group/join.ts';
 import { SyncBackoff } from './group/backoff.ts';
 import { newId, newToken } from './ids.ts';
 import { currentOs } from './paths.ts';
@@ -54,11 +55,7 @@ const SPENT_SETTING = 'group:spent';
 const JOINS_PER_MINUTE = 30;
 /** Requests one invitation is asked with before it is taken away: a machine joins once, and asks again a few times at most. */
 const JOIN_REQUESTS_MAX = 64;
-export const JOIN_ROUTE = '/group/join';
-/** What a join request and its answer are signed with, so neither passes for a coordination message. */
-const JOIN_SIGNING_PREFIX = 'boite-group-join\n';
 const JOIN_BODY_MAX = 16_384;
-const JOIN_TIMEOUT_MS = 8000;
 /** A member that missed a change is offered it again this often. */
 const TICK_MS = 15_000;
 /** Every member is asked for its roster this many ticks apart, whatever it acknowledged: five minutes. */
@@ -68,8 +65,7 @@ const NAME_REFRESH_MS = 5 * 60_000;
 const FAREWELL_MS = 6000;
 
 export type { NetworkSink } from './group/addresses.ts';
-
-interface JoinReply { nonce: string; publicKey: string; roster?: unknown; error?: string }
+export { JOIN_ROUTE } from './group/join.ts';
 
 export class GroupStore {
   roster: Roster | null = null;
@@ -707,7 +703,7 @@ export class GroupStore {
       const sealed = { ...seal(pack(body, signature), invite.x, context, invitePsk(invite.t)), context };
       let reply: JoinReply;
       try {
-        reply = await Promise.any(invite.a.map((url) => this.askToJoin(url, invite, sealed, nonce)));
+        reply = await Promise.any(invite.a.map((url) => askToJoin(url, invite, sealed, nonce)));
       } catch {
         throw refused(`no machine accepted the invitation at ${invite.a.join(', ')}: it is off or unreachable from here, or the invitation expired, was already used or was made by another machine`, { field: 'invite' });
       }
@@ -740,27 +736,6 @@ export class GroupStore {
     return this.view('owner')!;
   }
 
-  private async askToJoin(url: string, invite: Invite, sealed: { body: string } & Sealing, nonce: string): Promise<JoinReply> {
-    const response = await fetch(`${url}${JOIN_ROUTE}`, {
-      method: 'POST',
-      redirect: 'error',
-      signal: AbortSignal.timeout(JOIN_TIMEOUT_MS),
-      headers: { 'content-type': 'application/json', 'x-boite-peer': sealed.context.from, 'x-boite-invite': inviteId(invite.t), 'x-boite-signature': SEALED },
-      body: sealed.body,
-    });
-    const raw = await boundedBody(response.body, MAX_BODY);
-    // Only the machine that minted the invitation can answer under this key.
-    if (response.headers.get('x-boite-signature') !== SEALED) throw new Error('the answer is not sealed');
-    const { body, signature } = unpack(openResponse(raw, sealed.responseKey, sealed.context));
-    const reply = JSON.parse(body) as JoinReply;
-    // The invitation names who must answer: a machine with another key is not it.
-    const key = fingerprint(reply.publicKey);
-    if (key.coreId !== invite.c) throw new Error('another machine answered');
-    if (!verify(null, Buffer.from(JOIN_SIGNING_PREFIX + body), key.publicKey, Buffer.from(signature, 'base64'))) throw new Error('invalid signature');
-    if (reply.nonce !== nonce) throw new Error('nonce mismatch');
-    return reply;
-  }
-
   /**
    * The other end of `join`, on the member that minted the invitation. The
    * caller is in no roster yet. Its request names an invitation and is sealed

packages/core/src/group/join.ts

@@ -0,0 +1,38 @@
+/*
+ * The joining side of a group invitation: one sealed request to the member that
+ * minted it, and its answer checked against the key the invitation names.
+ * The member's side is `GroupStore.http` (docs/groups.md).
+ */
+import { verify } from 'node:crypto';
+import { boundedBody, MAX_BODY } from '../coordination-wire.ts';
+import { fingerprint } from './roster.ts';
+import { openResponse, SEALED, unpack, type Sealing } from './seal.ts';
+import { inviteId, type Invite } from './ticket.ts';
+
+export const JOIN_ROUTE = '/group/join';
+/** What a join request and its answer are signed with, so neither passes for a coordination message. */
+export const JOIN_SIGNING_PREFIX = 'boite-group-join\n';
+const JOIN_TIMEOUT_MS = 8000;
+
+export interface JoinReply { nonce: string; publicKey: string; roster?: unknown; error?: string }
+
+export async function askToJoin(url: string, invite: Invite, sealed: { body: string } & Sealing, nonce: string): Promise<JoinReply> {
+  const response = await fetch(`${url}${JOIN_ROUTE}`, {
+    method: 'POST',
+    redirect: 'error',
+    signal: AbortSignal.timeout(JOIN_TIMEOUT_MS),
+    headers: { 'content-type': 'application/json', 'x-boite-peer': sealed.context.from, 'x-boite-invite': inviteId(invite.t), 'x-boite-signature': SEALED },
+    body: sealed.body,
+  });
+  const raw = await boundedBody(response.body, MAX_BODY);
+  // Only the machine that minted the invitation can answer under this key.
+  if (response.headers.get('x-boite-signature') !== SEALED) throw new Error('the answer is not sealed');
+  const { body, signature } = unpack(openResponse(raw, sealed.responseKey, sealed.context));
+  const reply = JSON.parse(body) as JoinReply;
+  // The invitation names who must answer: a machine with another key is not it.
+  const key = fingerprint(reply.publicKey);
+  if (key.coreId !== invite.c) throw new Error('another machine answered');
+  if (!verify(null, Buffer.from(JOIN_SIGNING_PREFIX + body), key.publicKey, Buffer.from(signature, 'base64'))) throw new Error('invalid signature');
+  if (reply.nonce !== nonce) throw new Error('nonce mismatch');
+  return reply;
+}

packages/core/test/group-devices.test.ts

@@ -293,9 +293,9 @@ test('a machine joining a group tells a device of it to come back, not that it i
   let asked!: () => void;
   const held = new Promise<void>((resolve) => { release = resolve; });
   const asking = new Promise<void>((resolve) => { asked = resolve; });
-  const group = b.core.group as unknown as { askToJoin: (...args: unknown[]) => Promise<unknown> };
-  const ask = group.askToJoin.bind(b.core.group);
-  group.askToJoin = async (...args) => { asked(); await held; return ask(...args); };
+  const group = a.core.group as { http: (request: Request) => Promise<Response> };
+  const answer = group.http.bind(a.core.group);
+  group.http = async (request) => { asked(); await held; return answer(request); };
   const joining = b.core.group.join(invite);
   await asking;
   let refusal: { code?: number; data?: unknown } | undefined;