ui: reconnect call (#7711)

* Update Call.hs

* Update Internal.hs

* add reconnecting string

* Update WebRTC.kt

* desktop: show reconnecting state, keep call duration

* android keep connection time correct wiith adding reconnecting

* Update CallTypes.swift add reconnecting status on IOS

* Update WebRTC.swift add reconnecting

* add reconnecting and make sure call duration is correct

* Update call.ts

* core: regenerate call.js from call.ts

* Update Call.hs revert call status change

minimization

* Update Internal.hs rever to minimize changes

* Update CallTypes.swift minimize changes

* Update WebRTC.kt minimize changes

* Update CallView.desktop.kt desktop shows reconnecting without sending to core

* Update CallView.android.kt show reconnecting without sending it to the core

* Update ActiveCallView.swift show reconnecting to ui without sending to core

* Update CallView.desktop.kt add reconnecting sound on reconnect

* Update CallView.android.kt add reconnecting sound

* Update ActiveCallView.swift play reconnecting sound when reconnecting

* Update RecAndPlay.desktop.kt

* rebuild

* refactor

* build

* android, desktop: restart ICE to reconnect calls

* ios: restart ICE to reconnect calls

* webrtc: add call reconnection e2e tests

* android, desktop: fix lost call commands

---------

Co-authored-by: GuavaGoat <320862384+GuavaGoat@users.noreply.github.com>
Co-authored-by: Evgeny Poberezkin <evgeny@poberezkin.com>
Co-authored-by: Evgeny @ SimpleX Chat <259188159+evgeny-simplex@users.noreply.github.com>

packages/simplex-chat-webrtc/README.md

@@ -4,3 +4,20 @@
 npm i
 npm run build
 ```
+
+## Tests
+
+End-to-end tests run two headless Chromium peers with `call.js` and simulate network outages by relaying media via UDP sockets controlled by the test:
+
+```
+npx playwright-core install chromium
+npm test
+```
+
+Set `CHROME` to use another Chromium executable, and `VERBOSE=1` to log signaling and connection states.
+
+Full-stack tests in `test/fullstack.test.js` also pass call signaling via two chat cores and a local SMP server. They run when `SIMPLEX_CHAT` and `SMP_SERVER` are set to `simplex-chat` and `smp-server` executables:
+
+```
+SIMPLEX_CHAT=path/to/simplex-chat SMP_SERVER=path/to/smp-server npm test
+```

packages/simplex-chat-webrtc/package.json

@@ -8,7 +8,7 @@
     "dist"
   ],
   "scripts": {
-    "test": "echo \"Error: no test specified\" && exit 1",
+    "test": "tsc && node --test test/*.test.js",
     "build": "prettier --write --ignore-unknown . && tsc && ./copy"
   },
   "repository": {
@@ -31,6 +31,7 @@
     "husky": "^7.0.4",
     "isomorphic-webcrypto": "^2.3.8",
     "lint-staged": "^12.4.1",
+    "playwright-core": "^1.64.0",
     "prettier": "^2.6.2",
     "typescript": "^4.6.4"
   },

packages/simplex-chat-webrtc/src/call.ts

@@ -142,6 +142,7 @@ interface WCallAnswer extends IWCallCommand, IWCallResponse {
 interface WCallIceCandidates extends IWCallCommand, IWCallResponse {
   type: "ice"
   iceCandidates: string // JSON strings for RTCIceCandidateInit[]
+  session?: string // JSON string for RTCSessionDescriptionInit of ICE restart offer or answer
 }
 
 interface WCEnableMedia extends IWCallCommand {
@@ -281,6 +282,7 @@ interface Call {
   // controls whether transceiver's track was replaced already or set initially (in video call)
   cameraTrackWasSetBefore: boolean
   peerHasOldVersion: boolean
+  receiveRestartSession: (session: string, iceCandidates: RTCIceCandidateInit[]) => Promise<void>
 }
 
 interface NotConnectedCall {
@@ -306,6 +308,8 @@ let inactiveCallMediaSources: CallMediaSources = {
 let activeCall: Call | undefined
 let notConnectedCall: NotConnectedCall | undefined
 let answerTimeout = 30_000
+let reconnectTimeout = 30_000
+let iceRestartInterval = 10_000
 var useWorker = false
 var isDesktop = false
 var localizedState = ""
@@ -357,7 +361,8 @@ const processCommand = (function () {
     return {
       peerConnectionConfig: {
         iceServers: iceServers ?? defaultIceServers,
-        iceCandidatePoolSize: 10,
+        // Chrome gathers no candidates on ICE restart when candidates pool is used
+        iceCandidatePoolSize: 0,
         encodedInsertableStreams,
         iceTransportPolicy: relay ? "relay" : "all",
         // needed for Android WebView >= 69 && <= 72 where default was "plan-b" which is incompatible with transceivers
@@ -486,10 +491,14 @@ const processCommand = (function () {
       layout: notConnectedCall?.layout ?? LayoutType.Default,
       cameraTrackWasSetBefore: localStream.getVideoTracks().length > 0,
       peerHasOldVersion: false,
+      receiveRestartSession,
     }
     localOrPeerMediaSourcesChanged(call)
     await setupMediaStreams(call)
     let connectionTimeout: number | undefined = setTimeout(connectionHandler, answerTimeout)
+    let reconnectingTimeout: number | undefined
+    let iceRestartTimeout: number | undefined
+    let wasConnected = false
     if (pc.connectionState) {
       pc.addEventListener("connectionstatechange", connectionStateChange)
     } else {
@@ -500,33 +509,45 @@ const processCommand = (function () {
     async function connectionStateChange() {
       // "failed" means the second party did not answer in time (15 sec timeout in Chrome WebView)
       // See https://source.chromium.org/chromium/chromium/src/+/main:third_party/webrtc/p2p/base/p2p_constants.cc;l=70)
-      if (pc.connectionState !== "failed") connectionHandler()
+      if (pc.connectionState !== "failed" || wasConnected) connectionHandler()
     }
 
-    async function connectionHandler() {
+    async function connectionHandler(reconnectTimedOut = false) {
+      let state: string =
+        pc.connectionState ??
+        (pc.iceConnectionState != "completed" && pc.iceConnectionState != "checking"
+          ? pc.iceConnectionState
+          : pc.iceConnectionState == "completed"
+          ? "connected"
+          : "connecting") /* webView 69-70 doesn't have connectionState yet */
+      if (wasConnected) {
+        if (reconnectTimedOut) {
+          reconnectingTimeout = undefined
+          if (activeCall !== call || state == "connected") return
+          state = "disconnected"
+        } else if (state == "disconnected" || state == "failed") {
+          if (reconnectingTimeout !== undefined) return
+          reconnectingTimeout = setTimeout(() => connectionHandler(true), reconnectTimeout)
+          if (isIceRestarter()) restartIce()
+          state = "reconnecting"
+        } else if (state == "connecting" || state == "new") {
+          return
+        }
+      }
       sendMessageToNative({
         resp: {
           type: "connection",
           state: {
-            connectionState:
-              pc.connectionState ??
-              (pc.iceConnectionState != "completed" && pc.iceConnectionState != "checking"
-                ? pc.iceConnectionState
-                : pc.iceConnectionState == "completed"
-                ? "connected"
-                : "connecting") /* webView 69-70 doesn't have connectionState yet */,
+            connectionState: state,
             iceConnectionState: pc.iceConnectionState,
             iceGatheringState: pc.iceGatheringState,
             signalingState: pc.signalingState,
           },
         },
       })
-      if (
-        pc.connectionState == "disconnected" ||
-        pc.connectionState == "failed" ||
-        (!pc.connectionState && (pc.iceConnectionState == "disconnected" || pc.iceConnectionState == "failed"))
-      ) {
+      if (state == "disconnected" || state == "failed") {
         clearConnectionTimeout()
+        clearReconnecting()
         if (pc.connectionState) {
           pc.removeEventListener("connectionstatechange", connectionStateChange)
         } else {
@@ -536,8 +557,10 @@ const processCommand = (function () {
           setTimeout(() => sendMessageToNative({resp: {type: "ended"}}), 0)
         }
         endCall()
-      } else if (pc.connectionState == "connected" || (!pc.connectionState && pc.iceConnectionState == "connected")) {
+      } else if (state == "connected") {
         clearConnectionTimeout()
+        clearReconnecting()
+        wasConnected = true
         const stats = (await pc.getStats()) as Map<string, any>
         for (const stat of stats.values()) {
           const {type, state} = stat
@@ -564,6 +587,51 @@ const processCommand = (function () {
         connectionTimeout = undefined
       }
     }
+
+    function clearReconnecting() {
+      clearTimeout(reconnectingTimeout)
+      reconnectingTimeout = undefined
+      clearTimeout(iceRestartTimeout)
+      iceRestartTimeout = undefined
+    }
+
+    // Only the party that sent the initial offer restarts ICE, so that restart offers never collide
+    function isIceRestarter(): boolean {
+      return pc.localDescription?.type == "offer"
+    }
+
+    async function restartIce() {
+      if (activeCall !== call || reconnectingTimeout === undefined) return
+      iceRestartTimeout = setTimeout(restartIce, iceRestartInterval)
+      // the previous restart offer is not answered yet
+      if (pc.signalingState != "stable") return
+      try {
+        const iceCandidates = getIceCandidates(pc, config)
+        const offer = await pc.createOffer({iceRestart: true})
+        await pc.setLocalDescription(offer)
+        sendMessageToNative({resp: {type: "ice", session: serialize(offer), iceCandidates: await iceCandidates}})
+      } catch (e) {
+        console.log("ICE restart error", e)
+      }
+    }
+
+    async function receiveRestartSession(session: string, iceCandidates: RTCIceCandidateInit[]) {
+      const desc: RTCSessionDescriptionInit = parse(session)
+      const remoteDesc = new RTCSessionDescription(!webView69Or70() ? desc : adaptSdpToOldWebView(desc))
+      if (desc.type == "offer" && !isIceRestarter()) {
+        await pc.setRemoteDescription(remoteDesc)
+        const localIceCandidates = getIceCandidates(pc, config)
+        const answer = await pc.createAnswer()
+        await pc.setLocalDescription(answer)
+        addIceCandidates(pc, iceCandidates)
+        sendMessageToNative({resp: {type: "ice", session: serialize(answer), iceCandidates: await localIceCandidates}})
+      } else if (desc.type == "answer" && pc.signalingState == "have-local-offer") {
+        await pc.setRemoteDescription(remoteDesc)
+        addIceCandidates(pc, iceCandidates)
+      } else {
+        throw new Error(`unexpected ICE restart ${desc.type} in ${pc.signalingState} state`)
+      }
+    }
   }
 
   function serialize<T>(x: T): string {
@@ -729,7 +797,10 @@ const processCommand = (function () {
           break
         case "ice":
           const remoteIceCandidates: RTCIceCandidateInit[] = parse(command.iceCandidates)
-          if (pc) {
+          if (activeCall && command.session) {
+            await activeCall.receiveRestartSession(command.session, remoteIceCandidates)
+            resp = {type: "ok"}
+          } else if (pc) {
             addIceCandidates(pc, remoteIceCandidates)
             resp = {type: "ok"}
           } else {

packages/simplex-chat-webrtc/test/chat.js

@@ -0,0 +1,147 @@
+// Runs a local SMP server and chat cores with WebSocket API, so that call signaling goes via the chat protocol.
+const {spawn, execFileSync} = require("child_process")
+const fs = require("fs")
+const net = require("net")
+const path = require("path")
+const {sleep, log} = require("./harness")
+
+const START_TIMEOUT = 30000
+
+function freePort() {
+  return new Promise((resolve, reject) => {
+    const server = net.createServer()
+    server.on("error", reject)
+    server.listen(0, "127.0.0.1", () => {
+      const {port} = server.address()
+      server.close(() => resolve(port))
+    })
+  })
+}
+
+async function waitUntil(what, check) {
+  const deadline = Date.now() + START_TIMEOUT
+  while (!(await check())) {
+    if (Date.now() > deadline) throw new Error(`${what} did not start in ${START_TIMEOUT}ms`)
+    await sleep(200)
+  }
+}
+
+function canConnect(port) {
+  return new Promise((resolve) => {
+    const socket = net.connect(port, "127.0.0.1", () => {
+      socket.destroy()
+      resolve(true)
+    })
+    socket.on("error", () => resolve(false))
+  })
+}
+
+function spawnLogged(name, bin, args, env) {
+  const proc = spawn(bin, args, {env, stdio: ["ignore", "pipe", "pipe"]})
+  const output = (data) => log(`${name}:`, data.toString().trimEnd())
+  proc.stdout.on("data", output)
+  proc.stderr.on("data", output)
+  return proc
+}
+
+async function startSmpServer(bin, dir) {
+  const env = {...process.env, SMP_SERVER_CFG_PATH: path.join(dir, "cfg"), SMP_SERVER_LOG_PATH: path.join(dir, "log")}
+  const init = ["init", "--ip", "127.0.0.1", "--no-password", "--disable-web", "--disable-store-log", "-y"]
+  const fingerprint = execFileSync(bin, init, {env, stdio: ["ignore", "pipe", "pipe"]})
+    .toString()
+    .match(/Fingerprint: (\S+)/)[1]
+  const port = await freePort()
+  const iniFile = path.join(dir, "cfg", "smp-server.ini")
+  const ini = fs
+    .readFileSync(iniFile, "utf8")
+    .replace(/^port = .*$/m, `port = ${port}`)
+    .replace(/^host = .*$/m, "host = localhost")
+  fs.writeFileSync(iniFile, ini)
+  const proc = spawnLogged("smp-server", bin, ["start"], env)
+  await waitUntil("SMP server", () => canConnect(port))
+  return {address: `smp://${fingerprint}@localhost:${port}`, stop: () => proc.kill()}
+}
+
+class ChatClient {
+  constructor(name, proc, ws) {
+    this.name = name
+    this.proc = proc
+    this.ws = ws
+    this.corrId = 0
+    this.pending = new Map()
+    this.events = []
+    this.waiters = []
+    this.onEvent = async () => {}
+    ws.addEventListener("message", (e) => this.receive(JSON.parse(e.data)))
+  }
+
+  static async start(bin, dir, name, smpServer) {
+    const port = await freePort()
+    const args = ["-d", path.join(dir, name), "-s", smpServer, "-p", String(port), "--user-display-name", name]
+    const proc = spawnLogged(`chat ${name}`, bin, args, process.env)
+    await waitUntil(`chat ${name}`, () => canConnect(port))
+    const ws = new WebSocket(`ws://127.0.0.1:${port}`)
+    await new Promise((resolve, reject) => {
+      ws.addEventListener("open", resolve, {once: true})
+      ws.addEventListener("error", reject, {once: true})
+    })
+    return new ChatClient(name, proc, ws)
+  }
+
+  receive({corrId, resp}) {
+    if (corrId) {
+      this.pending.get(corrId)?.(resp)
+      this.pending.delete(corrId)
+      return
+    }
+    log(`${this.name} event:`, resp.type)
+    this.events.push(resp)
+    this.onEvent(resp).catch((e) => log(`${this.name} event error`, e))
+    this.waiters = this.waiters.filter((w) => !(w.pred(resp) && (w.resolve(resp), true)))
+  }
+
+  async cmd(cmd) {
+    const corrId = String(++this.corrId)
+    log(`${this.name} cmd:`, cmd.slice(0, 100))
+    const resp = await new Promise((resolve) => {
+      this.pending.set(corrId, resolve)
+      this.ws.send(JSON.stringify({corrId, cmd}))
+    })
+    if (resp.type === "chatCmdError") throw new Error(`${this.name} ${cmd.split(" ", 2).join(" ")}: ${JSON.stringify(resp)}`)
+    return resp
+  }
+
+  waitEvent(pred, timeout, what) {
+    const found = this.events.find(pred)
+    if (found) return Promise.resolve(found)
+    return new Promise((resolve, reject) => {
+      const waiter = {
+        pred,
+        resolve: (resp) => {
+          clearTimeout(timer)
+          resolve(resp)
+        },
+      }
+      this.waiters.push(waiter)
+      const timer = setTimeout(() => {
+        this.waiters = this.waiters.filter((w) => w !== waiter)
+        reject(new Error(`${this.name}: no ${what} event in ${timeout}ms`))
+      }, timeout)
+    })
+  }
+
+  stop() {
+    this.ws.close()
+    this.proc.kill()
+  }
+}
+
+async function connectContacts(alice, bob) {
+  const inv = await alice.cmd("/c")
+  await bob.cmd(`/c ${inv.connLinkInvitation.connFullLink}`)
+  const connected = (c) => c.waitEvent((e) => e.type === "contactConnected", START_TIMEOUT, "contactConnected")
+  const [a, b] = await Promise.all([connected(alice), connected(bob)])
+  return [a.contact.contactId, b.contact.contactId]
+}
+
+module.exports = {startSmpServer, ChatClient, connectContacts}

packages/simplex-chat-webrtc/test/fullstack.test.js

@@ -0,0 +1,188 @@
+// Calls between two chat cores connected via a local SMP server, with the test playing the native app:
+// it passes messages between call.js and the core API as the Android and desktop apps do.
+const {test} = require("node:test")
+const assert = require("node:assert/strict")
+const fs = require("fs")
+const os = require("os")
+const path = require("path")
+const {openPeers, sleep} = require("./harness")
+const {startSmpServer, ChatClient, connectContacts} = require("./chat")
+
+const {SIMPLEX_CHAT, SMP_SERVER} = process.env
+const MEDIA = "audio"
+const DISCONNECT_DETECTION = 10000
+const RECONNECT = 20000
+const SIGNALING = 15000
+const CORE_CALL_STATUSES = ["connecting", "connected", "disconnected", "failed"]
+
+async function withChatCall({useWorker = false}, run) {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "simplex-call-test-"))
+  const stops = []
+  try {
+    const smp = await startSmpServer(SMP_SERVER, path.join(dir, "smp"))
+    stops.push(smp.stop)
+    const alice = await ChatClient.start(SIMPLEX_CHAT, dir, "alice", smp.address)
+    stops.push(() => alice.stop())
+    const bob = await ChatClient.start(SIMPLEX_CHAT, dir, "bob", smp.address)
+    stops.push(() => bob.stop())
+    const [aliceContact, bobContact] = await connectContacts(alice, bob)
+    const peers = await openPeers({useWorker})
+    stops.push(peers.close)
+    // alice calls bob: a is alice's call.js, b is bob's call.js
+    const call = {...peers, alice, bob, aliceContact, bobContact}
+    connectApp(call.a, alice, aliceContact, peers)
+    connectApp(call.b, bob, bobContact, peers)
+    await startCall(call)
+    await run(call)
+  } finally {
+    for (const stop of stops.reverse()) await stop()
+    fs.rmSync(dir, {recursive: true, force: true})
+  }
+}
+
+function connectApp(peer, chat, contactId, {network, signaling}) {
+  peer.onResponse = async (resp) => {
+    switch (resp.type) {
+      case "ice": {
+        const {session, iceCandidates} = await network.routeSignal(peer.name, resp.session, resp.iceCandidates)
+        const extraInfo = JSON.stringify({rtcSession: session, rtcIceCandidates: iceCandidates})
+        signaling.send(() => chat.cmd(`/_call extra @${contactId} ${extraInfo}`))
+        break
+      }
+      case "connection":
+        if (CORE_CALL_STATUSES.includes(resp.state.connectionState)) {
+          await chat.cmd(`/_call status @${contactId} ${resp.state.connectionState}`)
+        }
+        break
+      case "ended":
+        await chat.cmd(`/_call end @${contactId}`).catch(() => {})
+        break
+    }
+  }
+  chat.onEvent = async (event) => {
+    switch (event.type) {
+      case "callExtraInfo":
+        await peer.command({type: "ice", iceCandidates: event.extraInfo.rtcIceCandidates, session: event.extraInfo.rtcSession})
+        break
+      case "callEnded":
+        await peer.command({type: "end"})
+        break
+    }
+  }
+}
+
+async function startCall({a, b, network, alice, bob, aliceContact, bobContact}) {
+  const capabilities = await a.command({type: "capabilities", media: MEDIA})
+  assert.equal(capabilities.resp.type, "capabilities")
+  const callType = {media: MEDIA, capabilities: capabilities.resp.capabilities}
+  await alice.cmd(`/_call invite @${aliceContact} ${JSON.stringify(callType)}`)
+
+  const {callInvitation} = await bob.waitEvent((e) => e.type === "callInvitation", SIGNALING, "callInvitation")
+  const offer = await b.command({type: "start", media: MEDIA, aesKey: callInvitation.sharedKey, iceServers: []})
+  assert.equal(offer.resp.type, "offer", JSON.stringify(offer.resp))
+  const offerSignal = await network.routeSignal("B", offer.resp.offer, offer.resp.iceCandidates)
+  const callOffer = {
+    callType: {media: MEDIA, capabilities: offer.resp.capabilities},
+    rtcSession: {rtcSession: offerSignal.session, rtcIceCandidates: offerSignal.iceCandidates},
+  }
+  await bob.cmd(`/_call offer @${bobContact} ${JSON.stringify(callOffer)}`)
+
+  const offered = await alice.waitEvent((e) => e.type === "callOffer", SIGNALING, "callOffer")
+  const answer = await a.command({
+    type: "offer",
+    offer: offered.offer.rtcSession,
+    iceCandidates: offered.offer.rtcIceCandidates,
+    media: MEDIA,
+    aesKey: offered.sharedKey,
+    iceServers: [],
+  })
+  assert.equal(answer.resp.type, "answer", JSON.stringify(answer.resp))
+  const answerSignal = await network.routeSignal("A", answer.resp.answer, answer.resp.iceCandidates)
+  const callAnswer = {rtcSession: answerSignal.session, rtcIceCandidates: answerSignal.iceCandidates}
+  await alice.cmd(`/_call answer @${aliceContact} ${JSON.stringify(callAnswer)}`)
+
+  const answered = await bob.waitEvent((e) => e.type === "callAnswer", SIGNALING, "callAnswer")
+  const ok = await b.command({type: "answer", answer: answered.answer.rtcSession, iceCandidates: answered.answer.rtcIceCandidates})
+  assert.equal(ok.resp.type, "ok", JSON.stringify(ok.resp))
+  await Promise.all([a.waitForState("connected", RECONNECT), b.waitForState("connected", RECONNECT)])
+}
+
+async function assertAudioDecoded(a, b) {
+  const before = [await a.receivedAudioEnergy(), await b.receivedAudioEnergy()]
+  await sleep(2000)
+  const after = [await a.receivedAudioEnergy(), await b.receivedAudioEnergy()]
+  assert.ok(after[0] > before[0], `A decoded no audio: ${before[0]} -> ${after[0]}`)
+  assert.ok(after[1] > before[1], `B decoded no audio: ${before[1]} -> ${after[1]}`)
+}
+
+async function callItem(chat, contactId) {
+  const r = await chat.cmd(`/_get chat @${contactId} count=1`)
+  return r.chat.chatItems[0].content
+}
+
+function restartSessionsReceived(chat) {
+  return chat.events.filter((e) => e.type === "callExtraInfo" && e.extraInfo.rtcSession).length
+}
+
+const skip = !(SIMPLEX_CHAT && SMP_SERVER) && "set SIMPLEX_CHAT and SMP_SERVER to chat CLI and SMP server executables"
+
+test("calls via chat core", {skip}, async (t) => {
+  for (const [app, useWorker] of [
+    ["android", false],
+    ["desktop", true],
+  ]) {
+    await t.test(`${app}: encrypted call reconnects with ICE restart after network path change`, () =>
+      withChatCall({useWorker}, async ({a, b, network, alice, bob, aliceContact, bobContact}) => {
+        assert.ok(a.messages.find((m) => m.resp.type === "capabilities").resp.capabilities.encryption, "call is not encrypted")
+        assert.equal(await a.page.evaluate(() => !!activeCall.worker), useWorker, "encryption worker")
+        await assertAudioDecoded(a, b)
+        assert.deepEqual(await callItem(alice, aliceContact), {type: "sndCall", status: "progress", duration: 0})
+        const connectedAt = Date.now()
+
+        const from = [a.messages.length, b.messages.length]
+        network.down = true
+        network.changePath()
+        await Promise.all([
+          a.waitForState("reconnecting", DISCONNECT_DETECTION, from[0]),
+          b.waitForState("reconnecting", DISCONNECT_DETECTION, from[1]),
+        ])
+        await sleep(3000)
+        network.down = false
+        await Promise.all([a.waitForState("connected", RECONNECT, from[0]), b.waitForState("connected", RECONNECT, from[1])])
+        await assertAudioDecoded(a, b)
+        assert.ok(restartSessionsReceived(alice) > 0, "alice received no ICE restart offer")
+        assert.ok(restartSessionsReceived(bob) > 0, "bob received no ICE restart answer")
+        assert.equal((await callItem(alice, aliceContact)).status, "progress")
+        assert.equal((await callItem(bob, bobContact)).status, "progress")
+
+        await alice.cmd(`/_call end @${aliceContact}`)
+        await a.command({type: "end"})
+        await bob.waitEvent((e) => e.type === "callEnded", SIGNALING, "callEnded")
+        const minDuration = Math.floor((Date.now() - connectedAt) / 1000) - 1
+        const aliceItem = await callItem(alice, aliceContact)
+        assert.equal(aliceItem.status, "ended")
+        assert.ok(aliceItem.duration >= minDuration, `call duration ${aliceItem.duration} < ${minDuration}`)
+      })
+    )
+  }
+
+  await t.test("call ends in both chats when reconnection times out", () =>
+    withChatCall({}, async ({a, b, network, alice, bob, aliceContact, bobContact}) => {
+      await Promise.all([a.setTimeouts({reconnectTimeout: 5000}), b.setTimeouts({reconnectTimeout: 5000})])
+      network.down = true
+      // the first party to time out ends the call, and the other party ends it on "callEnded" event
+      const endedBy = (peer, other) => peer.waitForEnded(DISCONNECT_DETECTION + 10000).then(() => other)
+      const other = await Promise.any([endedBy(a, bob), endedBy(b, alice)])
+      await other.waitEvent((e) => e.type === "callEnded", SIGNALING, "callEnded")
+      network.down = false
+      for (const [chat, contactId] of [
+        [alice, aliceContact],
+        [bob, bobContact],
+      ]) {
+        const item = await callItem(chat, contactId)
+        assert.equal(item.status, "ended", `${chat.name} call item: ${JSON.stringify(item)}`)
+        assert.ok(item.duration >= 5, `${chat.name} call duration ${item.duration}`)
+      }
+    })
+  )
+})

packages/simplex-chat-webrtc/test/harness.js

@@ -0,0 +1,317 @@
+// Runs two headless Chromium peers with call.js; the test plays the native app and the chat core.
+// Media can only flow through UDP relays controlled by the test, so outages and path changes are simulated
+// by dropping packets in the relays.
+const http = require("http")
+const fs = require("fs")
+const path = require("path")
+const dgram = require("dgram")
+const {chromium} = require("playwright-core")
+const LZString = require("lz-string")
+
+const ROOT = path.join(__dirname, "..")
+const PAGE_FILES = {
+  "/android/call.html": path.join(ROOT, "src/android/call.html"),
+  "/android/style.css": path.join(ROOT, "src/android/style.css"),
+  "/call.js": path.join(ROOT, "dist/call.js"),
+  "/lz-string.min.js": path.join(ROOT, "node_modules/lz-string/libs/lz-string.min.js"),
+}
+const CONTENT_TYPES = {".html": "text/html", ".js": "text/javascript", ".css": "text/css"}
+
+const startTime = Date.now()
+function log(...args) {
+  if (process.env.VERBOSE) console.log(((Date.now() - startTime) / 1000).toFixed(1).padStart(6), ...args)
+}
+
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
+const decode = (s) => JSON.parse(LZString.decompressFromBase64(s))
+const encode = (x) => LZString.compressToBase64(JSON.stringify(x))
+
+function servePage() {
+  const server = http.createServer((req, res) => {
+    const file = PAGE_FILES[new URL(req.url, "http://localhost").pathname]
+    if (!file) return res.writeHead(404).end()
+    res.writeHead(200, {"Content-Type": CONTENT_TYPES[path.extname(file)]}).end(fs.readFileSync(file))
+  })
+  return new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve(server)))
+}
+
+function bindUdp(ip) {
+  return new Promise((resolve) => {
+    const socket = dgram.createSocket("udp4")
+    socket.bind(0, ip, () => resolve(socket))
+  })
+}
+
+// Relays one ICE generation: each peer is told that the other peer is at the relay socket.
+class Relay {
+  constructor(network, ip) {
+    this.network = network
+    this.ip = ip
+    this.ufrag = {A: undefined, B: undefined}
+    this.real = {A: undefined, B: undefined}
+    this.dead = false
+    this.forwarded = 0
+  }
+
+  async open() {
+    this.sockets = {A: await bindUdp(this.ip), B: await bindUdp(this.ip)}
+    this.sockets.A.on("message", (msg, from) => this.forward(msg, from, "B", "A"))
+    this.sockets.B.on("message", (msg, from) => this.forward(msg, from, "A", "B"))
+  }
+
+  // socket A is advertised to B as A's address, so packets on it come from B and go to A via socket B
+  forward(msg, from, sender, receiver) {
+    if (this.dead || this.network.down) return
+    this.real[sender] = from
+    const to = this.real[receiver]
+    if (!to) return
+    this.forwarded++
+    this.sockets[sender].send(msg, to.port, to.address)
+  }
+
+  port(side) {
+    return this.sockets[side].address().port
+  }
+
+  close() {
+    this.sockets.A.close()
+    this.sockets.B.close()
+  }
+}
+
+class Network {
+  constructor() {
+    this.down = false
+    this.relays = []
+  }
+
+  // Returns the rewritten candidate, or undefined if the candidate cannot be relayed.
+  async route(side, candidate) {
+    const [, component, protocol, , ip, port, , type] = candidate.candidate.split(" ")
+    if (protocol.toLowerCase() !== "udp" || type !== "host" || ip.includes(":") || component !== "1") return
+    const ufrag = candidate.usernameFragment
+    if (this.relays.some((r) => r.ufrag[side] === ufrag)) return
+    let relay = this.relays.find((r) => !r.dead && r.ufrag[side] === undefined)
+    if (!relay) {
+      relay = new Relay(this, ip)
+      await relay.open()
+      this.relays.push(relay)
+    }
+    relay.ufrag[side] = ufrag
+    relay.real[side] = {address: ip, port: +port}
+    log(`relay ${this.relays.indexOf(relay)}: ${side} ${ip}:${port} -> ${relay.ip}:${relay.port(side)}`)
+    const parts = candidate.candidate.split(" ")
+    parts[4] = relay.ip
+    parts[5] = String(relay.port(side))
+    return {...candidate, candidate: parts.join(" ")}
+  }
+
+  // Takes and returns compressed session description and candidates, as in signaling messages
+  async routeSignal(side, session, iceCandidates) {
+    // candidates in SDP would connect the peers directly, bypassing the relays
+    if (session && decode(session).sdp.includes("a=candidate:")) throw new Error(`${side}: SDP has ICE candidates`)
+    const routed = []
+    for (const c of decode(iceCandidates)) {
+      const r = c?.candidate && (await this.route(side, c))
+      if (r) routed.push(r)
+    }
+    return {session, iceCandidates: encode(routed)}
+  }
+
+  // Addresses of all current ICE generations become unreachable, as when a device switches networks.
+  changePath() {
+    for (const relay of this.relays) relay.dead = true
+  }
+
+  close() {
+    for (const relay of this.relays) relay.close()
+  }
+}
+
+class Peer {
+  constructor(name, page, {ignoresIceRestart = false} = {}) {
+    this.name = name
+    this.page = page
+    this.ignoresIceRestart = ignoresIceRestart
+    this.messages = []
+    this.waiters = []
+    this.onResponse = async () => {}
+  }
+
+  async open(url) {
+    await this.page.exposeFunction("sendToTest", (json) => this.receive(JSON.parse(json)))
+    this.page.on("console", (m) => log(`${this.name} console:`, m.text().slice(0, 200)))
+    this.page.on("pageerror", (e) => log(`${this.name} page error:`, e.message))
+    await this.page.goto(url)
+    await this.page.evaluate(() => {
+      sendMessageToNative = (msg) => window.sendToTest(JSON.stringify(msg))
+    })
+  }
+
+  receive(msg) {
+    const resp = msg.resp
+    log(`${this.name} <-`, resp.type, resp.state?.connectionState ?? "", resp.session ? "with session" : "")
+    this.messages.push(msg)
+    if (!msg.command) this.onResponse(resp).catch((e) => log(`${this.name} response error`, e))
+    this.waiters = this.waiters.filter((w) => !(w.pred(msg) && (w.resolve(msg), true)))
+  }
+
+  connectionStates() {
+    return this.messages.filter((m) => m.resp.type === "connection").map((m) => m.resp.state.connectionState)
+  }
+
+  sessionsSent() {
+    return this.messages.filter((m) => m.resp.type === "ice" && !m.command && m.resp.session).length
+  }
+
+  ended() {
+    return this.messages.some((m) => m.resp.type === "ended")
+  }
+
+  waitFor(pred, timeout, what) {
+    const found = this.messages.find(pred)
+    if (found) return Promise.resolve(found)
+    return new Promise((resolve, reject) => {
+      const waiter = {
+        pred,
+        resolve: (msg) => {
+          clearTimeout(timer)
+          resolve(msg)
+        },
+      }
+      this.waiters.push(waiter)
+      const timer = setTimeout(() => {
+        this.waiters = this.waiters.filter((w) => w !== waiter)
+        reject(new Error(`${this.name}: no ${what} in ${timeout}ms, connection states: ${this.connectionStates()}`))
+      }, timeout)
+    })
+  }
+
+  waitForState(state, timeout, after = 0) {
+    const pred = (m) => m.resp.type === "connection" && m.resp.state.connectionState === state && this.messages.indexOf(m) >= after
+    return this.waitFor(pred, timeout, `"${state}" state`)
+  }
+
+  waitForEnded(timeout) {
+    return this.waitFor((m) => m.resp.type === "ended", timeout, "ended")
+  }
+
+  command(command) {
+    log(`${this.name} ->`, command.type, command.session ? "with session" : "")
+    return this.page.evaluate((c) => processCommand({corrId: 0, command: c}), command)
+  }
+
+  async setTimeouts(timeouts) {
+    await this.page.evaluate((t) => {
+      if (t.reconnectTimeout) reconnectTimeout = t.reconnectTimeout
+      if (t.iceRestartInterval) iceRestartInterval = t.iceRestartInterval
+    }, timeouts)
+  }
+
+  async receivedBytes() {
+    return (await this.inboundAudio()).bytesReceived
+  }
+
+  // energy grows only when received audio is decoded, i.e. it is decrypted with the right key
+  async receivedAudioEnergy() {
+    return (await this.inboundAudio()).totalAudioEnergy
+  }
+
+  inboundAudio() {
+    return this.page.evaluate(async () => {
+      let audio
+      ;(await activeCall.connection.getStats()).forEach((s) => {
+        if (s.type === "inbound-rtp" && s.kind === "audio" && (!audio || s.bytesReceived > audio.bytesReceived)) audio = s
+      })
+      return {bytesReceived: audio?.bytesReceived ?? 0, totalAudioEnergy: audio?.totalAudioEnergy ?? 0}
+    })
+  }
+}
+
+// Signaling stops while the network is down and resumes in order, as messages queued in SMP servers.
+class Signaling {
+  constructor(network) {
+    this.network = network
+    this.queue = Promise.resolve()
+    this.closed = false
+  }
+
+  send(deliver) {
+    this.queue = this.queue.then(async () => {
+      while (this.network.down && !this.closed) await sleep(100)
+      if (!this.closed) await deliver().catch((e) => log("signaling error", e.message))
+    })
+  }
+}
+
+// useWorker encrypts media in a worker with RTCRtpScriptTransform, as the desktop app does
+async function openPeers({answerer = {}, useWorker = false} = {}) {
+  const server = await servePage()
+  const url = `http://localhost:${server.address().port}/android/call.html`
+  const browser = await chromium.launch({
+    executablePath: process.env.CHROME,
+    args: [
+      "--use-fake-ui-for-media-stream",
+      "--use-fake-device-for-media-stream",
+      "--disable-features=WebRtcHideLocalIpsWithMdns",
+      "--autoplay-policy=no-user-gesture-required",
+    ],
+  })
+  const newPeer = async (name, opts) => {
+    const context = await browser.newContext({permissions: ["camera", "microphone"]})
+    const peer = new Peer(name, await context.newPage(), opts)
+    await peer.open(url)
+    if (useWorker) await peer.page.evaluate(() => (useWorker = true))
+    return peer
+  }
+  // A sends the call invitation and B accepts it, so B sends the WebRTC offer, as "start" command in the apps
+  const a = await newPeer("A", answerer)
+  const b = await newPeer("B")
+  const network = new Network()
+  const signaling = new Signaling(network)
+  const close = async () => {
+    signaling.closed = true
+    await browser.close()
+    network.close()
+    server.close()
+  }
+  return {a, b, network, signaling, close}
+}
+
+// The test relays signaling between the peers directly, as the apps and the chat core would do.
+async function startCall({media = "audio", answerer = {}} = {}) {
+  const peers = await openPeers({answerer})
+  const {a, b, network, signaling, close} = peers
+  const relayIce = (from, to) => async (resp) => {
+    if (resp.type !== "ice") return
+    const {session, iceCandidates} = await network.routeSignal(from.name, resp.session, resp.iceCandidates)
+    signaling.send(() => to.command({type: "ice", iceCandidates, session: to.ignoresIceRestart ? undefined : session}))
+  }
+  a.onResponse = relayIce(a, b)
+  b.onResponse = relayIce(b, a)
+
+  try {
+    await a.command({type: "capabilities", media})
+    const offer = await b.command({type: "start", media, iceServers: []})
+    if (offer.resp.type !== "offer") throw new Error(`B start: ${JSON.stringify(offer.resp)}`)
+    const offerSignal = await network.routeSignal("B", offer.resp.offer, offer.resp.iceCandidates)
+    const answer = await a.command({
+      type: "offer",
+      offer: offerSignal.session,
+      iceCandidates: offerSignal.iceCandidates,
+      media,
+      iceServers: [],
+    })
+    if (answer.resp.type !== "answer") throw new Error(`A offer: ${JSON.stringify(answer.resp)}`)
+    const answerSignal = await network.routeSignal("A", answer.resp.answer, answer.resp.iceCandidates)
+    const ok = await b.command({type: "answer", answer: answerSignal.session, iceCandidates: answerSignal.iceCandidates})
+    if (ok.resp.type !== "ok") throw new Error(`B answer: ${JSON.stringify(ok.resp)}`)
+    await Promise.all([a.waitForState("connected", 20000), b.waitForState("connected", 20000)])
+  } catch (e) {
+    await close()
+    throw e
+  }
+  return peers
+}
+
+module.exports = {openPeers, startCall, sleep, log}

packages/simplex-chat-webrtc/test/reconnect.test.js

@@ -0,0 +1,95 @@
+const {test} = require("node:test")
+const assert = require("node:assert/strict")
+const {startCall, sleep} = require("./harness")
+
+// Chrome reports "disconnected" about 6 seconds after packets stop
+const DISCONNECT_DETECTION = 10000
+const RECONNECT = 20000
+
+async function assertMediaFlows(a, b) {
+  const before = [await a.receivedBytes(), await b.receivedBytes()]
+  await sleep(2000)
+  const after = [await a.receivedBytes(), await b.receivedBytes()]
+  assert.ok(after[0] > before[0], `A received no media: ${before[0]} -> ${after[0]}`)
+  assert.ok(after[1] > before[1], `B received no media: ${before[1]} -> ${after[1]}`)
+}
+
+async function assertReconnects(a, b, outage) {
+  const from = [a.messages.length, b.messages.length]
+  await Promise.all([
+    a.waitForState("reconnecting", DISCONNECT_DETECTION, from[0]),
+    b.waitForState("reconnecting", DISCONNECT_DETECTION, from[1]),
+  ])
+  await outage()
+  await Promise.all([a.waitForState("connected", RECONNECT, from[0]), b.waitForState("connected", RECONNECT, from[1])])
+  await assertMediaFlows(a, b)
+  assert.ok(!a.ended() && !b.ended(), "call ended")
+  assert.ok(!a.connectionStates().includes("disconnected"), `A states: ${a.connectionStates()}`)
+  assert.ok(!b.connectionStates().includes("disconnected"), `B states: ${b.connectionStates()}`)
+}
+
+async function withCall(opts, run) {
+  const call = await startCall(opts)
+  try {
+    await run(call)
+  } finally {
+    await call.close()
+  }
+}
+
+test("call reconnection", {concurrency: true}, async (t) => {
+  await t.test("recovers after a brief outage", () =>
+    withCall({}, async ({a, b, network}) => {
+      await assertMediaFlows(a, b)
+      network.down = true
+      await assertReconnects(a, b, async () => {
+        network.down = false
+      })
+    })
+  )
+
+  await t.test("recovers after an outage longer than ICE consent timeout", () =>
+    withCall({}, async ({a, b, network}) => {
+      network.down = true
+      await assertReconnects(a, b, async () => {
+        await sleep(15000)
+        network.down = false
+      })
+      assert.ok(b.sessionsSent() > 0, "offerer did not restart ICE")
+      assert.ok(a.sessionsSent() > 0, "answerer did not answer ICE restart")
+    })
+  )
+
+  await t.test("recovers when the network path changes", () =>
+    withCall({}, async ({a, b, network}) => {
+      network.down = true
+      network.changePath()
+      await assertReconnects(a, b, async () => {
+        network.down = false
+      })
+      const relay = network.relays[network.relays.length - 1]
+      assert.ok(!relay.dead && relay.forwarded > 0, "media does not use the new path")
+    })
+  )
+
+  await t.test("ends the call when reconnection times out", () =>
+    withCall({}, async ({a, b, network}) => {
+      await Promise.all([a.setTimeouts({reconnectTimeout: 5000}), b.setTimeouts({reconnectTimeout: 5000})])
+      network.down = true
+      await Promise.all([a.waitForState("reconnecting", DISCONNECT_DETECTION), b.waitForState("reconnecting", DISCONNECT_DETECTION)])
+      await Promise.all([a.waitForEnded(10000), b.waitForEnded(10000)])
+      assert.deepEqual(a.connectionStates().slice(-2), ["reconnecting", "disconnected"])
+      assert.deepEqual(b.connectionStates().slice(-2), ["reconnecting", "disconnected"])
+    })
+  )
+
+  await t.test("ends the call when the peer does not support ICE restart", () =>
+    withCall({answerer: {ignoresIceRestart: true}}, async ({a, b, network}) => {
+      await Promise.all([a.setTimeouts({reconnectTimeout: 10000}), b.setTimeouts({reconnectTimeout: 10000})])
+      network.changePath()
+      await Promise.all([a.waitForEnded(DISCONNECT_DETECTION + 15000), b.waitForEnded(DISCONNECT_DETECTION + 15000)])
+      assert.ok(b.sessionsSent() > 0, "offerer did not restart ICE")
+      assert.equal(a.sessionsSent(), 0)
+    })
+  )
+})