feat: add transport-reset handling for server restarts

src/client.js

@@ -6,7 +6,11 @@ import {
   SERVICES_ID,
   SubChannel,
 } from './lib/sub-channel.js'
-import { ClientClosedError, ProjectClosedError } from './errors.js'
+import {
+  ClientClosedError,
+  ProjectClosedError,
+  TransportClosedError,
+} from './errors.js'
 
 /** @import { ClientApi, MessagePortLike } from 'rpc-reflector' */
 /** @import { MapeoProject, MapeoManager } from '@comapeo/core' */
@@ -29,13 +33,24 @@ const EMITTER_METHODS = new Set([
   'listenerCount',
 ])
 
+// Removing a listener from a client that is already dead is correct teardown
+// behaviour (e.g. React effect cleanup running against a stale reference), so
+// unlike the other emitter methods these must not throw on a closed proxy.
+const EMITTER_UNSUBSCRIBE_METHODS = new Set([
+  'removeListener',
+  'off',
+  'removeAllListeners',
+])
+
 /**
  * Build the Proxy returned for a closed client/project reference. Method calls
  * (including nested namespaces such as `project.observation.*`) reject with
  * `makeError()`, keeping the `Promise`-returning contract callers expect.
  * EventEmitter methods are the exception: callers don't await them, so a
  * rejected promise would surface as an unhandled rejection — they throw
- * synchronously instead, at the call site.
+ * synchronously instead, at the call site. Unsubscribe methods are a further
+ * exception: they are no-ops that return the proxy for chaining, because
+ * removing a listener from a dead client is valid teardown, not a bug.
  *
  * @param {() => Error} makeError
  */
@@ -44,6 +59,9 @@ function createClosedProxy(makeError) {
   const handler = {
     get(_target, prop) {
       if (typeof prop === 'string' && EMITTER_METHODS.has(prop)) {
+        if (EMITTER_UNSUBSCRIBE_METHODS.has(prop)) {
+          return () => proxy
+        }
         return () => {
           throw makeError()
         }
@@ -57,7 +75,8 @@ function createClosedProxy(makeError) {
       return Promise.reject(makeError())
     },
   }
-  return new Proxy({}, handler)
+  const proxy = new Proxy({}, handler)
+  return proxy
 }
 
 /**
@@ -75,6 +94,8 @@ function createClosedProxy(makeError) {
  * >} ComapeoCoreClientApi */
 
 const CLOSE = Symbol('close')
+const TRANSPORT_RESET = Symbol('transportReset')
+const RESUBSCRIBE = Symbol('resubscribe')
 
 /**
  * @param {MessagePortLike} messagePort
@@ -108,6 +129,7 @@ export function createComapeoCoreClient(messagePort, opts = {}) {
    * @type {Set<{
    *   client: ClientApi<MapeoProject>,
    *   channel: SubChannel,
+   *   hardClose: (error: Error) => void,
    * }>}
    */
   const openProjectClients = new Set()
@@ -129,6 +151,46 @@ export function createComapeoCoreClient(messagePort, opts = {}) {
   let clientClosed = false
   const managerClosedProxy = createClosedProxy(() => new ClientClosedError())
 
+  // Bumped on every transport reset so a `getProject` whose routing response
+  // arrived just before the reset cannot cache a wrapper bound to the dead
+  // server (see `resolveProjectClient`).
+  let resetGeneration = 0
+
+  function handleTransportReset() {
+    if (clientClosed) return
+    resetGeneration++
+
+    // Fail in-flight calls fast with a distinguishable, retryable error
+    // instead of leaving them to hit the per-call timeout. Resubscription is
+    // deliberately NOT done here: at drop time the transport is down, and
+    // each ON frame written into it can nudge the native transport into
+    // retrying forever while the server stays down. The consumer calls
+    // `resubscribeCoreClient` once the transport is back up.
+    createClient.rejectPending(managerClient, new TransportClosedError())
+    createClient.rejectPending(projectRoutingClient, new TransportClosedError())
+
+    // Project instance ids are minted by a counter that restarts with the
+    // server, so a restarted server can mint an id equal to the one a cached
+    // wrapper is bound to — the instance-id currency check in
+    // `resolveProjectClient` could then falsely pass. Hard-close every
+    // wrapper and drop the cache so `getProject` always builds a fresh
+    // wrapper against the new server.
+    for (const entry of openProjectClients) {
+      entry.hardClose(new TransportClosedError())
+    }
+    openProjectClients.clear()
+    currentProjectClients.clear()
+  }
+
+  function handleResubscribe() {
+    // rpc-reflector's resubscribe is a no-op on a closed client, and the
+    // server ignores duplicate ON messages, so this is safe to call
+    // repeatedly and after close.
+    if (clientClosed) return
+    createClient.resubscribe(managerClient)
+    createClient.resubscribe(projectRoutingClient)
+  }
+
   const client = new Proxy(managerClient, {
     get(target, prop, receiver) {
       if (prop === CLOSE) {
@@ -157,6 +219,14 @@ export function createComapeoCoreClient(messagePort, opts = {}) {
         }
       }
 
+      if (prop === TRANSPORT_RESET) {
+        return handleTransportReset
+      }
+
+      if (prop === RESUBSCRIBE) {
+        return handleResubscribe
+      }
+
       if (prop === 'getProject') {
         return createProjectClient
       }
@@ -205,9 +275,15 @@ export function createComapeoCoreClient(messagePort, opts = {}) {
    * @returns {Promise<ClientApi<MapeoProject>>}
    */
   async function resolveProjectClient(projectPublicId) {
+    const generation = resetGeneration
     const instanceId =
       await projectRoutingClient.assertProjectExists(projectPublicId)
 
+    // A reset can land between the routing response arriving and this
+    // continuation running; a wrapper minted now would be bound to the dead
+    // server, so reject like any other call in flight during the reset.
+    if (generation !== resetGeneration) throw new TransportClosedError()
+
     const current = currentProjectClients.get(projectPublicId)
     if (current && current.instanceId === instanceId) return current.wrapper
 
@@ -232,9 +308,6 @@ export function createComapeoCoreClient(messagePort, opts = {}) {
     const projectClient = createClient(projectChannel, opts)
     projectChannel.start()
 
-    const registryEntry = { client: projectClient, channel: projectChannel }
-    openProjectClients.add(registryEntry)
-
     // Wrap projectClient to intercept `close`: after the wire close settles,
     // tear down the local client + channel — rejecting any in-flight calls.
     // Cache eviction is in the 'close' listener below, which also covers
@@ -254,6 +327,31 @@ export function createComapeoCoreClient(messagePort, opts = {}) {
     const closedProxy = createClosedProxy(() =>
       closed ? new ProjectClosedError() : new ClientClosedError(),
     )
+
+    const registryEntry = {
+      client: projectClient,
+      channel: projectChannel,
+      // Local-only teardown for a transport reset: the server this instance
+      // belonged to is gone, so there is no wire close to await. Stale
+      // references then behave like a closed project (`ProjectClosedError`),
+      // and `close()` on them resolves like an already-closed project.
+      hardClose: (/** @type {Error} */ error) => {
+        createClient.rejectPending(projectClient, error)
+        // Fire 'close' listeners (the app's teardown listeners and the
+        // cache-eviction listener below) before closing the client, matching
+        // what a server-initiated close delivers. Emitted before close so a
+        // `.off` called from inside a close listener hits a still-open
+        // client (a harmless OFF frame into a dead socket) rather than
+        // throwing.
+        createClient.emitLocal(projectClient, 'close')
+        createClient.close(projectClient)
+        projectChannel.close()
+        closed = true
+        closePromise = Promise.resolve()
+      },
+    }
+    openProjectClients.add(registryEntry)
+
     const wrappedProjectClient = new Proxy(projectClient, {
       get(target, prop, receiver) {
         if (prop === 'close') {
@@ -298,6 +396,53 @@ export async function closeComapeoCoreClient(client) {
   return client[CLOSE]()
 }
 
+/**
+ * Notify the core client that the underlying transport has dropped (e.g.
+ * Android killed the foreground service hosting the server). Call at drop
+ * time — the transport does not need to be back up. This:
+ *
+ * - rejects every call that was in flight with `TransportClosedError`
+ *   (`code: 'RPC_TRANSPORT_CLOSED'`), so callers can fail fast and retry
+ *   instead of waiting for the per-call timeout;
+ * - hard-closes every open project client (each fires its `'close'` event
+ *   locally, so app-held `once('close')` teardown listeners run) and drops
+ *   the project cache, so a later `getProject` builds a fresh wrapper
+ *   against the new server. Stale project references held by the app behave
+ *   like closed projects (`ProjectClosedError`), except that removing
+ *   listeners from them is a harmless no-op. A `getProject` in flight during
+ *   the reset rejects with `TransportClosedError`.
+ *
+ * This deliberately does NOT replay event subscriptions: writing into a
+ * still-down transport can keep nudging it into a hot retry loop. Once the
+ * transport has reconnected to the restarted server, call
+ * {@link resubscribeCoreClient} to restore subscriptions.
+ *
+ * No-op after `closeComapeoCoreClient`.
+ *
+ * @param {ComapeoCoreClientApi} client client created with `createComapeoCoreClient`
+ * @returns {void}
+ */
+export function notifyCoreClientTransportReset(client) {
+  // @ts-expect-error
+  return client[TRANSPORT_RESET]()
+}
+
+/**
+ * Re-send the core client's event subscriptions (manager events and project
+ * routing) after the transport has reconnected to a restarted server, which
+ * lost all subscription state. Call once the transport is back up, after
+ * having called `notifyCoreClientTransportReset` at drop time. Safe to call
+ * repeatedly (the server ignores duplicate subscriptions); no-op after
+ * `closeComapeoCoreClient`.
+ *
+ * @param {ComapeoCoreClientApi} client client created with `createComapeoCoreClient`
+ * @returns {void}
+ */
+export function resubscribeCoreClient(client) {
+  // @ts-expect-error
+  return client[RESUBSCRIBE]()
+}
+
 /**
  * @typedef {ClientApi<ComapeoServicesApi>} ComapeoServicesClientApi
  */
@@ -329,3 +474,33 @@ export function createComapeoServicesClient(messagePort, opts = {}) {
 export function closeComapeoServicesClient(servicesClient) {
   createClient.close(servicesClient)
 }
+
+/**
+ * Notify the services client that the underlying transport has dropped:
+ * rejects every call that was in flight with `TransportClosedError`
+ * (`code: 'RPC_TRANSPORT_CLOSED'`). Call at drop time. Like
+ * {@link notifyCoreClientTransportReset} this does not replay event
+ * subscriptions — call {@link resubscribeServicesClient} once the transport
+ * has reconnected. No-op after `closeComapeoServicesClient`.
+ *
+ * @param {ComapeoServicesClientApi} servicesClient client created with `createComapeoServicesClient`
+ * @returns {void}
+ */
+export function notifyServicesClientTransportReset(servicesClient) {
+  createClient.rejectPending(servicesClient, new TransportClosedError())
+}
+
+/**
+ * Re-send the services client's event subscriptions after the transport has
+ * reconnected to a restarted server, which lost all subscription state. Call
+ * once the transport is back up, after having called
+ * `notifyServicesClientTransportReset` at drop time. Safe to call repeatedly
+ * (the server ignores duplicate subscriptions); no-op after
+ * `closeComapeoServicesClient`.
+ *
+ * @param {ComapeoServicesClientApi} servicesClient client created with `createComapeoServicesClient`
+ * @returns {void}
+ */
+export function resubscribeServicesClient(servicesClient) {
+  createClient.resubscribe(servicesClient)
+}

src/errors.js

@@ -16,6 +16,20 @@ export const ProjectClosedError = createErrorClass({
   status: 410,
 })
 
+/**
+ * Rejected client-side into calls that were in flight when the underlying
+ * transport dropped (e.g. the process hosting the server was killed and
+ * restarted). Distinguishable from a real failure or a timeout: the call
+ * never completed on the server's side of the connection, so a read is safe
+ * to retry once the transport has reconnected.
+ */
+export const TransportClosedError = createErrorClass({
+  code: 'RPC_TRANSPORT_CLOSED',
+  message:
+    'Transport closed: the connection to the server dropped while the call was in flight',
+  status: 503,
+})
+
 /**
  * Thrown client-side when a method is called after the CoMapeo core client
  * (the whole IPC client) has been closed via `closeComapeoCoreClient`.

src/index.js

@@ -1,8 +1,12 @@
 export {
   createComapeoCoreClient,
   closeComapeoCoreClient,
+  notifyCoreClientTransportReset,
+  resubscribeCoreClient,
   createComapeoServicesClient,
   closeComapeoServicesClient,
+  notifyServicesClientTransportReset,
+  resubscribeServicesClient,
 } from './client.js'
 export {
   createComapeoCoreServer,

tests/events.js

@@ -52,21 +52,22 @@ test('Client listeners stop receiving events after removeListener', async (t) =>
   assert.equal(count, 1, 'no further events after removeListener')
 })
 
-test('EventEmitter methods throw synchronously after the project is closed', async (t) => {
+test('EventEmitter subscribe methods throw synchronously after the project is closed; unsubscribe methods are no-ops', async (t) => {
   const { client } = setup(t)
   const projectId = await client.createProject({ name: 'mapeo' })
   const project = await client.getProject(projectId)
 
   await project.close()
 
   // Emitter methods are not awaited by callers, so a rejected promise would
-  // surface as an unhandled rejection — they throw at the call site instead.
+  // surface as an unhandled rejection — subscribe methods throw at the call
+  // site instead. Unsubscribe methods are valid teardown on a closed
+  // reference (e.g. React effect cleanup), so they are no-ops.
   assert.throws(() => project.on('some-event', () => {}), {
     code: ProjectClosedError.code,
   })
-  assert.throws(() => project.removeListener('some-event', () => {}), {
-    code: ProjectClosedError.code,
-  })
+  assert.doesNotThrow(() => project.removeListener('some-event', () => {}))
+  assert.doesNotThrow(() => project.off('some-event', () => {}))
 })
 
 test('EventEmitter methods throw synchronously after the client is closed', async (t) => {