feat: add createClient.rejectPending and createClient.resubscribe

README.md

@@ -106,6 +106,22 @@ The returned `clientApi` will be correctly typed, with synchronous functions con
 
 The static method `close()` will remove all event listeners from the `channel` used to create the client. It will not close or destroy the MessagePort used as the `channel`.
 
+### `createClient.rejectPending(clientApi, error)`
+
+Rejects every in-flight method call with `error` and returns the number of calls rejected. Use this when the transport to the server has dropped (e.g. the process hosting the server was killed) and pending calls can never be answered — without it they would hang until `options.timeout`. Unlike `close()`, the client remains fully usable afterwards: new calls can be made and event listeners stay registered. A response arriving later for a rejected call is ignored.
+
+Note that a rejected call may still have executed on the server if the request was delivered before the transport dropped — whether it is safe to retry is the caller's judgement (reads generally are; mutations need care).
+
+No-op returning `0` if nothing is pending or the client is closed.
+
+### `createClient.resubscribe(clientApi)`
+
+Re-sends a subscription message to the server for every event — including events on nested sub-objects — that currently has at least one listener, and returns the number of subscription messages sent. Use this after the server has restarted: a restarted server has lost its subscription state, so it will not emit events until the client re-subscribes. Safe to call repeatedly — the server ignores duplicate subscriptions, so events are not double-delivered.
+
+Only call this once the transport to the restarted server is connected again. Subscription messages written into a down transport are lost, and on some transports each write triggers a reconnect attempt, which can keep the transport busy while the server is still down.
+
+No-op returning `0` if the client is closed.
+
 ### Errors
 
 The client can reject a call with one of the following error classes. Each carries a stable `.code` property so consumers can identify it without matching against the error message. Both are exported from the package and can also be checked with `instanceof`.

client.js

@@ -45,6 +45,8 @@ const emitterSubscribeMethods = [
 ]
 const emitterUnsubscribeMethods = ['removeListener', 'off']
 const closeProp = Symbol('close')
+const rejectPendingProp = Symbol('rejectPending')
+const resubscribeProp = Symbol('resubscribe')
 
 // Per-call message ids are namespaced into a random band so that the id spaces
 // of two client instances sharing one transport (or one client re-created
@@ -243,6 +245,35 @@ export function createClient(
     log.info({ pendingCount }, 'RPC client closed')
   }
 
+  /** @param {Error} error */
+  function handleRejectPending(error) {
+    if (closed) return 0
+    const pendingCount = pending.size
+    // Rejecting the inner promise settles the p-timeout wrapper, which clears
+    // its timer, so no timers leak and the fallback cannot double-settle.
+    for (const [, [, reject]] of pending) {
+      reject(error)
+    }
+    pending.clear()
+    collector.clear()
+    log.info({ pendingCount }, 'Rejected pending RPC calls')
+    return pendingCount
+  }
+
+  function handleResubscribe() {
+    if (closed) return 0
+    let onCount = 0
+    for (const encodedEventName of emitter.eventNames()) {
+      if (typeof encodedEventName !== 'string') continue
+      if (emitter.listenerCount(encodedEventName) === 0) continue
+      const [eventPropArray, eventName] = parse(encodedEventName)
+      send([msgType.ON, eventName, eventPropArray])
+      onCount++
+    }
+    log.info({ onCount }, 'Re-sent event subscriptions')
+    return onCount
+  }
+
   const subClientCache = new Map()
 
   return createSubClient([], {})
@@ -261,6 +292,12 @@ export function createClient(
         if (prop === closeProp && propArray.length === 0) {
           return () => handleClose()
         }
+        if (prop === rejectPendingProp && propArray.length === 0) {
+          return handleRejectPending
+        }
+        if (prop === resubscribeProp && propArray.length === 0) {
+          return handleResubscribe
+        }
         // if (prop === util.inspect.custom) {
         //   // Only Node < 12, not called in browsers
         //   return () => '[rpcProxyClient]'
@@ -375,6 +412,41 @@ createClient.close = function close(client) {
   return client[closeProp]()
 }
 
+/**
+ * Reject every in-flight method call on a client with the given error, e.g.
+ * when the transport has dropped and pending calls can never be answered.
+ * Unlike `close`, the client remains fully usable afterwards. A response
+ * arriving later for a rejected call is ignored. No-op (returning 0) if
+ * nothing is pending or the client is closed.
+ *
+ * Note this is a static method on `createClient` and it expects a client
+ * created with `createClient` as its argument.
+ *
+ * @param {any} client A client created with `createClient`
+ * @param {Error} error Error to reject each pending call with
+ * @returns {number} Number of calls rejected
+ */
+createClient.rejectPending = function rejectPending(client, error) {
+  return client[rejectPendingProp](error)
+}
+
+/**
+ * Re-send an `ON` subscription message to the server for every event
+ * (including events on nested sub-clients) that currently has at least one
+ * listener. Use this after the server has restarted and lost its
+ * subscription state: the reconnected server will start emitting the
+ * subscribed events again. No-op (returning 0) if the client is closed.
+ *
+ * Note this is a static method on `createClient` and it expects a client
+ * created with `createClient` as its argument.
+ *
+ * @param {any} client A client created with `createClient`
+ * @returns {number} Number of subscription (`ON`) messages sent
+ */
+createClient.resubscribe = function resubscribe(client) {
+  return client[resubscribeProp]()
+}
+
 /**
  * For non-objectMode streams we receive the response as either Buffer or
  * strings (Node also supports Uint8Arrays in streams, but message-stream will

test/e2e.test.js

@@ -691,6 +691,198 @@ function runTests(setup) {
     t.end()
   })
 
+  test('rejectPending() rejects in-flight calls and the client stays usable', async (t) => {
+    const api = {
+      neverResolves() {
+        return new Promise(() => {})
+      },
+      /**
+       * @param {number} a
+       * @param {number} b
+       */
+      add(a, b) {
+        return a + b
+      },
+    }
+    const { client } = setup(t, api, { timeout: 200 })
+    const transportError = new Error('transport dropped')
+
+    const inFlight = [client.neverResolves(), client.neverResolves()]
+
+    t.equal(
+      createClient.rejectPending(client, transportError),
+      2,
+      'Returns the number of calls rejected',
+    )
+    for (const call of inFlight) {
+      try {
+        await call
+        t.fail('Expected rejection')
+      } catch (err) {
+        t.equal(err, transportError, 'Rejects with the caller-supplied error')
+      }
+    }
+
+    t.equal(
+      createClient.rejectPending(client, transportError),
+      0,
+      'Returns 0 when nothing is pending',
+    )
+
+    // Wait past the call timeout so any timer left behind by the rejected calls
+    // has had a chance to fire before we check the client is still healthy.
+    await delay(300)
+    t.equal(await client.add(1, 2), 3, 'Client still works after rejectPending')
+    t.end()
+  })
+
+  test('A late response for a call rejected by rejectPending() is ignored', async (t) => {
+    /** @type {(value: string) => void} */
+    let respond = () => {}
+    /** @type {() => void} */
+    let onServerCalled = () => {}
+    const serverCalled = new Promise((resolve) => {
+      onServerCalled = () => resolve(undefined)
+    })
+    const api = {
+      slowMethod() {
+        onServerCalled()
+        return new Promise((resolve) => {
+          respond = resolve
+        })
+      },
+      /**
+       * @param {number} a
+       * @param {number} b
+       */
+      add(a, b) {
+        return a + b
+      },
+    }
+    /** @type {unknown[]} */
+    const warnings = []
+    const logger = createTestLogger({
+      warn: (...args) => warnings.push(args),
+    })
+    const { client } = setup(t, api, { timeout: 5000, logger })
+
+    const inFlight = client.slowMethod()
+    // The server must be handling the call before we give up on it, otherwise
+    // there is nothing in flight to answer late.
+    await serverCalled
+    createClient.rejectPending(client, new Error('transport dropped'))
+    await inFlight.catch(() => {})
+
+    // The server only now answers the call the client has already given up on.
+    respond('late result')
+    await delay(200)
+
+    t.equal(warnings.length, 1, 'Late response is logged as ignored')
+    t.equal(
+      await client.add(1, 2),
+      3,
+      'Client still works after a late response',
+    )
+    t.end()
+  })
+
+  test('resubscribe() restores event subscriptions on a restarted server', async (t) => {
+    const rootEmitter = new EventEmitter()
+    const nestedEmitter = new EventEmitter()
+    const api = Object.assign(rootEmitter, { nested: nestedEmitter })
+    const { client, server, serverMPort } = setup(t, api)
+
+    /** @type {string[]} */
+    const received = []
+    const unsubscribed = () => received.push('unsubscribedEvent')
+    client.on('rootEvent', () => received.push('rootEvent'))
+    client.nested.on('nestedEvent', () => received.push('nestedEvent'))
+    client.on('unsubscribedEvent', unsubscribed)
+    client.off('unsubscribedEvent', unsubscribed)
+    await delay(200)
+
+    // Restart the server: closing it drops its listeners and the replacement
+    // starts with no subscription state, as a real server restart would.
+    server.close()
+    const restartedServer = createServer(api, serverMPort, {
+      logger: makeLogger(),
+    })
+    t.teardown(() => restartedServer.close())
+
+    rootEmitter.emit('rootEvent')
+    nestedEmitter.emit('nestedEvent')
+    await delay(200)
+    t.deepEqual(received, [], 'Restarted server emits nothing to the client')
+
+    t.equal(
+      createClient.resubscribe(client),
+      2,
+      'Returns the number of subscriptions re-sent',
+    )
+    await delay(200)
+    t.deepEqual(
+      rootEmitter.eventNames(),
+      ['rootEvent'],
+      'Only the root event that still has a listener is re-subscribed',
+    )
+    t.deepEqual(
+      nestedEmitter.eventNames(),
+      ['nestedEvent'],
+      'Nested event is re-subscribed',
+    )
+
+    rootEmitter.emit('rootEvent')
+    nestedEmitter.emit('nestedEvent')
+    await delay(200)
+    t.deepEqual(
+      received.sort(),
+      ['nestedEvent', 'rootEvent'],
+      'Events reach the client again after resubscribe',
+    )
+    t.end()
+  })
+
+  test('rejectPending() and resubscribe() are no-ops after close()', async (t) => {
+    const emitter = new EventEmitter()
+    const api = Object.assign(emitter, {
+      neverResolves() {
+        return new Promise(() => {})
+      },
+    })
+    const { client, server, serverMPort } = setup(t, api, { timeout: 5000 })
+
+    client.on('someEvent', () => {})
+    const inFlight = client.neverResolves()
+    await delay(200)
+
+    createClient.close(client)
+    await inFlight.catch(() => {})
+
+    server.close()
+    const restartedServer = createServer(api, serverMPort, {
+      logger: makeLogger(),
+    })
+    t.teardown(() => restartedServer.close())
+
+    t.equal(
+      createClient.rejectPending(client, new Error('nope')),
+      0,
+      'rejectPending() returns 0 on a closed client',
+    )
+    t.equal(
+      createClient.resubscribe(client),
+      0,
+      'resubscribe() returns 0 on a closed client',
+    )
+    await delay(200)
+    t.equal(
+      emitter.eventNames().length,
+      0,
+      'Nothing is re-subscribed on the server',
+    )
+    t.end()
+  })
+
   test('Non-string methods / props are not supported', (t) => {
     const { client } = setup(t, myApi)
     // @ts-expect-error
@@ -948,3 +1140,8 @@ function whenServerSubscribed(emitter, eventName, fn) {
     process.nextTick(fn)
   })
 }
+
+/** @param {number} ms */
+function delay(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms))
+}