feat(desktop): open bots on other computers in the account

apps/Codync.xcodeproj/project.pbxproj

@@ -862,7 +862,7 @@
 				GCC_WARN_UNUSED_VARIABLE = YES;
 				IPHONEOS_DEPLOYMENT_TARGET = 26.0;
 				MACOSX_DEPLOYMENT_TARGET = 14.0;
-				MARKETING_VERSION = 2.45.1;
+				MARKETING_VERSION = 2.46.0;
 				MTL_ENABLE_DEBUG_INFO = INCLUDE_SOURCE;
 				MTL_FAST_MATH = YES;
 				ONLY_ACTIVE_ARCH = YES;
@@ -960,7 +960,7 @@
 				GCC_WARN_UNUSED_VARIABLE = YES;
 				IPHONEOS_DEPLOYMENT_TARGET = 26.0;
 				MACOSX_DEPLOYMENT_TARGET = 14.0;
-				MARKETING_VERSION = 2.45.1;
+				MARKETING_VERSION = 2.46.0;
 				MTL_ENABLE_DEBUG_INFO = NO;
 				MTL_FAST_MATH = YES;
 				PRODUCT_NAME = "$(TARGET_NAME)";
@@ -1219,7 +1219,7 @@
 				GCC_WARN_UNUSED_VARIABLE = YES;
 				IPHONEOS_DEPLOYMENT_TARGET = 26.0;
 				MACOSX_DEPLOYMENT_TARGET = 14.0;
-				MARKETING_VERSION = 2.45.1;
+				MARKETING_VERSION = 2.46.0;
 				MTL_ENABLE_DEBUG_INFO = INCLUDE_SOURCE;
 				MTL_FAST_MATH = YES;
 				ONLY_ACTIVE_ARCH = YES;
@@ -1369,7 +1369,7 @@
 				GCC_WARN_UNUSED_VARIABLE = YES;
 				IPHONEOS_DEPLOYMENT_TARGET = 26.0;
 				MACOSX_DEPLOYMENT_TARGET = 14.0;
-				MARKETING_VERSION = 2.45.1;
+				MARKETING_VERSION = 2.46.0;
 				MTL_ENABLE_DEBUG_INFO = NO;
 				MTL_FAST_MATH = YES;
 				PRODUCT_NAME = "$(TARGET_NAME)";

apps/desktop/package.json

@@ -1,7 +1,7 @@
 {
   "name": "codync-desktop",
   "productName": "Codync",
-  "version": "2.45.1",
+  "version": "2.46.0",
   "description": "Codync for macOS, Linux and Windows: your coding agents as teammates.",
   "author": {
     "name": "leepokai",

apps/desktop/src/main/channel/crypto.test.ts

@@ -1,7 +1,7 @@
 import assert from 'node:assert/strict'
 import { readFileSync } from 'node:fs'
 import { test } from 'node:test'
-import { b64url, deviceHello, FrameCodec, openFrame, privateKey, rawPublic, sasCommit, sealFrame, unb64url } from './crypto.ts'
+import { b64url, deviceHello, FrameCodec, openFrame, privateKey, rawPublic, sasCode, sasCommit, sealFrame, unb64url } from './crypto.ts'
 
 const vectors = JSON.parse(readFileSync(new URL('../../../../../docs/reference/fixtures/remote-relay-vectors.json', import.meta.url), 'utf8'))
 const { keys, handshake: hs, frames, sas } = vectors
@@ -50,6 +50,7 @@ test('a long message goes in 256 KiB chunks and comes back whole', () => {
   assert.deepEqual(opened[2], long)
 })
 
-test('the access request commit matches the protocol vectors', () => {
+test('the access request commit and code match the protocol vectors', () => {
   assert.equal(b64url(sasCommit(device.pub, unb64url(sas.deviceNonce))), sas.commit)
+  assert.equal(sasCode(unb64url(keys.hostSignPub), device.pub, unb64url(sas.deviceNonce), unb64url(sas.hostNonce)), sas.code)
 })

apps/desktop/src/main/channel/crypto.ts

@@ -57,6 +57,12 @@ export function deviceHello(device: DeviceKey, hostSignPub: Uint8Array, eph: Key
 /** §4.2 B: what the device commits to before the computer shows its half of the code. */
 export const sasCommit = (dk: Uint8Array, nD: Uint8Array) => createHash('sha256').update(Buffer.concat([text('codync/sascommit/v1'), dk, nD])).digest()
 
+/** §4.2 B: the 6-digit code both sides show once the device revealed its nonce. */
+export function sasCode(hk: Uint8Array, dk: Uint8Array, nD: Uint8Array, nH: Uint8Array) {
+  const hash = createHash('sha256').update(Buffer.concat([text('codync/sas/v2'), hk, dk, nD, nH])).digest()
+  return String(hash.readUInt32BE(0) % 1_000_000).padStart(6, '0')
+}
+
 // ---- frames (§6.3) ----
 
 export const CHUNK = 256 * 1024

apps/desktop/src/main/channel/index.ts

@@ -10,8 +10,15 @@ import { RelayChannel, unreachable, type ChannelError, type Socket } from './rel
 // (`host-proxy.ts`), carried by one `RelayChannel` per computer.
 
 const channels = new Map<string, RelayChannel>()
+/** Open streams by `<window>:<id>`, so a crashed page's can be found. */
 const streams = new Map<string, () => void>()
 
+function closeStream(key: string) {
+  const stop = streams.get(key)
+  streams.delete(key)
+  stop?.()
+}
+
 function closeAll() {
   for (const stop of streams.values()) stop()
   streams.clear()
@@ -57,19 +64,20 @@ function subscription(path: string): { sub: 'events' | 'term'; body: unknown } |
 }
 
 function openStream(account: AccountService, sender: WebContents, id: string, computerId: string, signKey: string, path: string) {
+  const key = `${sender.id}:${id}`
   const send = (channel: string, payload: unknown) => {
     if (!sender.isDestroyed()) sender.send(channel, id, payload)
   }
   const end = (error: ChannelError | null) => {
-    if (!streams.delete(id)) return
+    if (!streams.delete(key)) return
     send('host:stream:end', error)
   }
   const sub = subscription(path)
   if (!sub) return send('host:stream:end', { status: 404, message: 'Not found' })
   let stop: (() => void) | null = null
-  streams.set(id, () => stop?.())
+  streams.set(key, () => stop?.())
   channelFor(account, computerId, signKey).then((channel) => {
-    if (!streams.has(id)) return
+    if (!streams.has(key)) return
     stop = channel.subscribe(sub.sub, sub.body, (event) => send('host:stream:data', JSON.stringify(event)), end)
   }, (error: ChannelError) => end(error))
 }
@@ -85,10 +93,10 @@ export function registerChannels(account: AccountService) {
     }
   })
   ipcMain.on('channel:stream:open', (e, id: string, computerId: string, signKey: string, path: string) => openStream(account, e.sender, id, computerId, signKey, path))
-  ipcMain.on('channel:stream:close', (_e, id: string) => {
-    const stop = streams.get(id)
-    streams.delete(id)
-    stop?.()
+  ipcMain.on('channel:stream:close', (e, id: string) => closeStream(`${e.sender.id}:${id}`))
+  // A crashed page never says its streams went (a page that unloads closes its own, preload/index.ts).
+  app.on('render-process-gone', (_e, contents) => {
+    for (const key of [...streams.keys()]) if (key.startsWith(`${contents.id}:`)) closeStream(key)
   })
   onAttention((on) => {
     for (const channel of channels.values()) channel.setAttended(on)

apps/desktop/src/main/channel/relay-channel.test.ts

@@ -158,3 +158,49 @@ test('a socket that goes quiet or fails is closed and dialed again', async () =>
     mock.timers.reset()
   }
 })
+
+test("a computer that rejoins while the relay's window is full gets a new handshake, then the calls that never left", async () => {
+  mock.timers.enable({ apis: ['setInterval', 'setTimeout', 'Date'] })
+  try {
+    const { ch, relays } = channel()
+    const calls = Array.from({ length: 250 }, () => ch.call('hello', {}, 600_000).catch(() => 'dropped'))
+    const relay = relays[0]!
+    relay.push({ t: 'presence', online: true })
+    assert.equal(relay.sent.length, 199)
+    mock.timers.tick(500)
+    relay.push({ t: 'presence', online: true })
+    // `hello` waits out the window, longer than the handshake's own deadline, which runs from when it leaves.
+    mock.timers.tick(10_500)
+    assert.equal(relay.closedWith, null)
+    assert.equal(ch.state, 'ready')
+    const resent = relay.sent.slice(199)
+    assert.equal(resent.length, 51)
+    for (const call of resent) relay.answer({ id: call.id, ok: 'answered' })
+    const ends = await Promise.all(calls)
+    assert.equal(ends.filter((end) => end === 'dropped').length, 199)
+    assert.equal(ends.filter((end) => end === 'answered').length, 51)
+    ch.close()
+  } finally {
+    mock.timers.reset()
+  }
+})
+
+test("a call that times out while it waits on the relay's window never reaches the computer", async () => {
+  mock.timers.enable({ apis: ['setInterval', 'setTimeout', 'Date'] })
+  try {
+    const { ch, relays } = channel()
+    const earlier = Array.from({ length: 199 }, () => ch.call('hello', {}, 600_000).catch(() => {}))
+    const relay = relays[0]!
+    relay.push({ t: 'presence', online: true })
+    const late = ch.call('setBot', {}, 5_000)
+    mock.timers.tick(5_000)
+    await assert.rejects(late, { status: 0 })
+    mock.timers.tick(6_000)
+    assert.equal(relay.sent.length, 199)
+    assert.ok(relay.sent.every((message) => message.m === 'hello'))
+    ch.close()
+    await Promise.all(earlier)
+  } finally {
+    mock.timers.reset()
+  }
+})

apps/desktop/src/main/channel/relay-channel.ts

@@ -1,4 +1,5 @@
 import { deviceHello, FrameCodec, unb64url, type DeviceKey } from './crypto.ts'
+import { RelayPacer } from './relay-pacer.ts'
 
 // One end-to-end encrypted channel to an account computer over the Codync relay (docs/reference/remote-relay.md
 // §6–§7), as the iPhone's `ChannelTransport` does it, relay only: the computers the desktop app reaches this way
@@ -66,6 +67,8 @@ const asleep = (): ChannelError => ({ status: 0, message: 'Your computer is offl
 export class RelayChannel {
   state: 'idle' | 'connecting' | 'asleep' | 'ready' | 'refused' = 'idle'
   private socket: Socket | null = null
+  /** The socket's outgoing messages, held to the relay's per-link budget. */
+  private out: RelayPacer | null = null
   private codec: FrameCodec | null = null
   private hello: ReturnType<typeof deviceHello> | null = null
   private nextId = 0
@@ -101,7 +104,7 @@ export class RelayChannel {
         reject,
         timer: setTimeout(() => {
           if (!this.calls.delete(call)) return
-          if (call.id !== null) this.sendInner({ id: call.id, cancel: true })
+          if (call.id !== null) this.sendInner(() => ({ id: call.id, cancel: true }))
           reject(this.state === 'asleep' ? asleep() : unreachable())
           this.settle()
         }, timeoutMs),
@@ -124,7 +127,7 @@ export class RelayChannel {
     if (this.codec) this.sendSub(s)
     return () => {
       if (!this.subs.delete(s)) return
-      if (s.id !== null) this.sendInner({ id: s.id, cancel: true })
+      if (s.id !== null) this.sendInner(() => ({ id: s.id, cancel: true }))
       this.settle()
     }
   }
@@ -178,6 +181,7 @@ export class RelayChannel {
       return this.again()
     }
     this.socket = socket
+    this.out = new RelayPacer((text) => socket.send(text))
     this.state = 'connecting'
     this.heard = Date.now()
     socket.addEventListener('message', (e) => {
@@ -194,7 +198,7 @@ export class RelayChannel {
     socket.addEventListener('error', () => this.socket === socket && socket.close(BROKEN))
     this.heartbeat = setInterval(() => {
       if (Date.now() - this.heard > SILENCE) socket.close(BROKEN)
-      else socket.send(PING)
+      else this.out?.send(() => [PING])
     }, HEARTBEAT)
   }
 
@@ -223,12 +227,16 @@ export class RelayChannel {
   }
 
   private startHandshake() {
-    this.hello = deviceHello(this.setup.device, unb64url(this.setup.signKey))
+    const hello = deviceHello(this.setup.device, unb64url(this.setup.signKey))
+    this.hello = hello
     this.state = 'connecting'
-    this.socket?.send(JSON.stringify(this.hello.hello))
     const socket = this.socket
     if (this.handshakeTimer) clearTimeout(this.handshakeTimer)
-    this.handshakeTimer = setTimeout(() => this.hello && this.socket === socket && socket?.close(1000), HANDSHAKE)
+    // The deadline runs from when `hello` leaves: it may wait out the relay's window first.
+    this.out?.send(() => {
+      this.handshakeTimer = setTimeout(() => this.hello === hello && this.socket === socket && socket?.close(1000), HANDSHAKE)
+      return [JSON.stringify(hello.hello)]
+    })
   }
 
   private welcome(wire: { ek: string; sig: string }) {
@@ -294,6 +302,8 @@ export class RelayChannel {
   }
 
   private stopTimers() {
+    this.out?.clear()
+    this.out = null
     if (this.heartbeat) clearInterval(this.heartbeat)
     if (this.handshakeTimer) clearTimeout(this.handshakeTimer)
     if (this.idle) clearTimeout(this.idle)
@@ -310,9 +320,13 @@ export class RelayChannel {
     }, delay)
   }
 
-  /** The channel is gone: what was sent on it can't be answered any more. What still waits to be sent waits on. */
+  /**
+   * The channel is gone: what was sent on it can't be answered any more, and frames sealed for it but still held
+   * back by the pacer can't be opened by the next one. Calls and subscriptions not sent yet wait on.
+   */
   private drop(error: ChannelError) {
     this.codec = null
+    this.out?.clear()
     for (const call of this.calls) {
       if (call.id === null) continue
       this.calls.delete(call)
@@ -348,23 +362,38 @@ export class RelayChannel {
 
   // MARK: inner RPC
 
+  /**
+   * A call or subscription gets its id when it leaves, so one given up on while it waited for the relay's window
+   * (`call`'s timeout, an unsubscribe) never goes out, and `drop` keeps it waiting for the next channel.
+   */
   private sendCall(call: Call) {
-    call.id = this.id()
-    this.sendInner({ id: call.id, m: call.method, b: call.body ?? null })
+    this.sendInner(() => {
+      if (!this.calls.has(call)) return null
+      call.id = this.id()
+      return { id: call.id, m: call.method, b: call.body ?? null }
+    })
   }
 
   private sendSub(sub: Sub) {
-    sub.id = this.id()
-    this.sendInner({ id: sub.id, sub: sub.sub, b: sub.body })
+    this.sendInner(() => {
+      if (!this.subs.has(sub)) return null
+      sub.id = this.id()
+      return { id: sub.id, sub: sub.sub, b: sub.body }
+    })
   }
 
   private id() {
     this.nextId = this.nextId >= 0xffff_ffff ? 1 : this.nextId + 1
     return this.nextId
   }
 
-  private sendInner(message: unknown) {
-    if (!this.codec || !this.socket) return
-    for (const frame of this.codec.seal(message)) this.socket.send(JSON.stringify(frame))
+  /** Seals what `make` returns once the pacer lets it out, on the channel it was queued for. */
+  private sendInner(make: () => unknown) {
+    const codec = this.codec
+    if (!codec || !this.out) return
+    this.out.send(() => {
+      const message = make()
+      return message === null ? [] : codec.seal(message).map((frame) => JSON.stringify(frame))
+    })
   }
 }

apps/desktop/src/main/channel/relay-pacer.test.ts

@@ -0,0 +1,61 @@
+import assert from 'node:assert/strict'
+import { mock, test } from 'node:test'
+import { RelayPacer } from './relay-pacer.ts'
+
+test("a large upload waits out the relay's 12 MiB window instead of overrunning it, in order", () => {
+  mock.timers.enable({ apis: ['setTimeout', 'Date'] })
+  const written: string[] = []
+  const pacer = new RelayPacer((text) => written.push(text))
+  const frames = Array.from({ length: 20 }, (_, i) => String(i).padEnd(700 * 1024))
+  pacer.send(() => [...frames])
+  assert.equal(written.length, 17)
+  mock.timers.tick(10_999)
+  assert.equal(written.length, 17)
+  mock.timers.tick(1)
+  assert.deepEqual(written, frames)
+  mock.timers.reset()
+})
+
+test("many small messages wait out the relay's 200-message window", () => {
+  mock.timers.enable({ apis: ['setTimeout', 'Date'] })
+  const written: string[] = []
+  const pacer = new RelayPacer((text) => written.push(text))
+  for (let i = 0; i < 250; i++) pacer.send(() => ['{"t":"ping"}'])
+  assert.equal(written.length, 200)
+  mock.timers.tick(11_000)
+  assert.equal(written.length, 250)
+  mock.timers.reset()
+})
+
+test('a message is made only once the window has room for it, so it can still be given up on while it waits', () => {
+  mock.timers.enable({ apis: ['setTimeout', 'Date'] })
+  const written: string[] = []
+  const pacer = new RelayPacer((text) => written.push(text))
+  for (let i = 0; i < 200; i++) pacer.send(() => ['old'])
+  let madeAt: number | null = null
+  let wanted = true
+  pacer.send(() => {
+    madeAt = Date.now()
+    return wanted ? ['given up'] : []
+  })
+  pacer.send(() => ['next'])
+  assert.equal(madeAt, null)
+  wanted = false
+  mock.timers.tick(11_000)
+  assert.equal(madeAt, 11_000)
+  assert.deepEqual(written.slice(200), ['next'])
+  mock.timers.reset()
+})
+
+test('dropping what waits keeps what already left counting against the window', () => {
+  mock.timers.enable({ apis: ['setTimeout', 'Date'] })
+  const written: string[] = []
+  const pacer = new RelayPacer((text) => written.push(text))
+  for (let i = 0; i < 201; i++) pacer.send(() => ['old'])
+  pacer.clear()
+  pacer.send(() => ['new'])
+  assert.equal(written.length, 200)
+  mock.timers.tick(11_000)
+  assert.deepEqual(written.slice(200), ['new'])
+  mock.timers.reset()
+})