feat: transport-reset recovery for server restarts

README.md

@@ -91,6 +91,15 @@ Project instance lifecycle is owned entirely by the server. The client cannot cl
 - `closeComapeoCoreClient(client)` tears down the manager, the project-routing channel, and every project reference. After this, all calls — including `getProject(id)` — reject with [`ClientClosedError`](#errors). (The services client is independent; close it separately with [`closeComapeoServicesClient`](#closecomapeoservicesclientservicesclient-clientapicomapeoservicesapi-void).)
 - Calls already in flight when the client closes reject with [`RpcChannelClosedError`](#errors); they are not re-routed.
 
+### Transport reset
+
+When the process hosting the server dies and restarts while the client stays alive (e.g. Android's foreground service being killed), the transport owner should drive a two-phase recovery:
+
+- At drop time, call `notifyCoreClientTransportReset(client)` (and `notifyServicesClientTransportReset(servicesClient)`): every in-flight call rejects immediately with [`TransportClosedError`](#errors) (`code: 'RPC_TRANSPORT_CLOSED'`) instead of waiting out its timeout. Reads are safe to retry once the transport reconnects; whether to replay a mutation is the caller's judgement — nothing is replayed automatically.
+- Once the transport is connected to the restarted server, call `resubscribeCoreClient(client)` (and `resubscribeServicesClient(servicesClient)`): every event subscription — manager and per-project — is re-sent, since the fresh server has no subscription state. Resubscription is deliberately not done at drop time: ON frames written into a down transport can keep nudging it into reconnect attempts while the server stays down.
+
+Project references need no recovery: their channels are keyed by project id, which a restarted server serves identically — the next call transparently re-opens the project. Both functions are safe to call repeatedly and are no-ops after the client is closed. Requires rpc-reflector >= 4.4.
+
 ### Events
 
 The client reflects the `EventEmitter` interface of the manager and of each project. `client.on(event, listener)` forwards events emitted on the server across the channel; `removeListener` / `off` stop the forwarding. After a reference is closed these emitter methods behave differently — see [Errors](#errors).

src/client.js

@@ -7,7 +7,7 @@ import {
   SERVICES_ID,
   SubChannel,
 } from './lib/sub-channel.js'
-import { ClientClosedError } from './errors.js'
+import { ClientClosedError, TransportClosedError } from './errors.js'
 
 /** @import { ClientApi, MessagePortLike } from 'rpc-reflector' */
 /** @import { MapeoProject, MapeoManager } from '@comapeo/core' */
@@ -86,6 +86,23 @@ function createClosedProxy(makeError) {
  * >} ComapeoCoreClientApi */
 
 const CLOSE = Symbol('close')
+const TRANSPORT_RESET = Symbol('transportReset')
+const RESUBSCRIBE = Symbol('resubscribe')
+
+/**
+ * `createClient.rejectPending` / `createClient.resubscribe` ship in
+ * rpc-reflector >= 4.4. Fail with a clear message rather than a bare
+ * TypeError when the installed version predates them.
+ *
+ * @param {'rejectPending' | 'resubscribe'} name
+ */
+function assertResetApi(name) {
+  if (typeof (/** @type {any} */ (createClient)[name]) !== 'function') {
+    throw new Error(
+      `createClient.${name} is unavailable — transport-reset support requires rpc-reflector >= 4.4`,
+    )
+  }
+}
 
 /**
  * Create the client side of `createComapeoCoreServer`.
@@ -145,8 +162,46 @@ export function createComapeoCoreClient(messagePort, opts = {}) {
   let clientClosed = false
   const clientClosedProxy = createClosedProxy(() => new ClientClosedError())
 
+  function handleTransportReset() {
+    if (clientClosed) return
+    assertResetApi('rejectPending')
+    // 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. Project
+    // references stay valid: their channels are keyed by project id, which
+    // a restarted server serves identically.
+    createClient.rejectPending(managerClient, new TransportClosedError())
+    createClient.rejectPending(projectRoutingClient, new TransportClosedError())
+    for (const entry of openProjectClients) {
+      createClient.rejectPending(entry.client, new TransportClosedError())
+    }
+  }
+
+  function handleResubscribe() {
+    if (clientClosed) return
+    assertResetApi('resubscribe')
+    // Safe to call repeatedly: the server ignores duplicate ON messages. A
+    // replayed project subscription also re-opens that project server-side —
+    // an active listener is an expression of interest.
+    createClient.resubscribe(managerClient)
+    for (const entry of openProjectClients) {
+      createClient.resubscribe(entry.client)
+    }
+  }
+
   const client = new Proxy(managerClient, {
     get(target, prop, receiver) {
+      if (prop === TRANSPORT_RESET) {
+        return handleTransportReset
+      }
+
+      if (prop === RESUBSCRIBE) {
+        return handleResubscribe
+      }
+
       if (prop === CLOSE) {
         return async () => {
           managerChannel.close()
@@ -272,6 +327,39 @@ export async function closeComapeoCoreClient(client) {
   return client[CLOSE]()
 }
 
+/**
+ * Notify the client that its transport to the server has dropped (e.g. the
+ * process hosting the server died): every in-flight call — manager, project
+ * routing, and per-project — rejects immediately with `TransportClosedError`
+ * instead of waiting out its timeout. The client remains fully usable;
+ * project references stay valid and serve the restarted server once the
+ * transport reconnects. Safe to call repeatedly; no-op after
+ * `closeComapeoCoreClient`.
+ *
+ * Deliberately does NOT replay event subscriptions — call
+ * {@link resubscribeCoreClient} once the transport is connected again.
+ *
+ * @param {ComapeoCoreClientApi} client client created with `createComapeoCoreClient`
+ */
+export function notifyCoreClientTransportReset(client) {
+  // @ts-expect-error
+  client[TRANSPORT_RESET]()
+}
+
+/**
+ * Re-send every event subscription (manager and per-project) to the server.
+ * Call after the transport to a restarted server is connected again — the
+ * fresh server has no subscription state until then. Safe to call
+ * repeatedly (the server ignores duplicate subscriptions); no-op after
+ * `closeComapeoCoreClient`.
+ *
+ * @param {ComapeoCoreClientApi} client client created with `createComapeoCoreClient`
+ */
+export function resubscribeCoreClient(client) {
+  // @ts-expect-error
+  client[RESUBSCRIBE]()
+}
+
 /**
  * @typedef {ClientApi<ComapeoServicesApi>} ComapeoServicesClientApi
  */
@@ -303,3 +391,27 @@ export function createComapeoServicesClient(messagePort, opts = {}) {
 export function closeComapeoServicesClient(servicesClient) {
   createClient.close(servicesClient)
 }
+
+/**
+ * Services-client counterpart of {@link notifyCoreClientTransportReset}:
+ * reject the services client's in-flight calls with `TransportClosedError`.
+ * Safe to call repeatedly and after `closeComapeoServicesClient` (no-op).
+ *
+ * @param {ComapeoServicesClientApi} servicesClient client created with `createComapeoServicesClient`
+ */
+export function notifyServicesClientTransportReset(servicesClient) {
+  assertResetApi('rejectPending')
+  createClient.rejectPending(servicesClient, new TransportClosedError())
+}
+
+/**
+ * Services-client counterpart of {@link resubscribeCoreClient}: re-send the
+ * services client's event subscriptions once the transport is back up. Safe
+ * to call repeatedly and after `closeComapeoServicesClient` (no-op).
+ *
+ * @param {ComapeoServicesClientApi} servicesClient client created with `createComapeoServicesClient`
+ */
+export function resubscribeServicesClient(servicesClient) {
+  assertResetApi('resubscribe')
+  createClient.resubscribe(servicesClient)
+}

src/errors.js

@@ -17,6 +17,19 @@ export const ProjectLeftError = createErrorClass({
   status: 410,
 })
 
+/**
+ * Rejection for calls that were in flight when the transport to the server
+ * dropped (e.g. the process hosting the server died). Distinguishable from
+ * ordinary failures so callers can decide whether the call 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,