Send realtime notifications one at a time on the bus connection (#456)

src/__tests__/lib/realtime/bus.test.ts

@@ -22,6 +22,9 @@ const pg = vi.hoisted(() => {
   /** Kept across tests: the bridge connects once for the life of a process. */
   const listening: string[] = []
   let failNotify = false
+  /** How many queries the connection has been handed and not yet answered. */
+  let inFlight = 0
+  let mostInFlight = 0
   class Client {
     on(event: string, handler: (arg: unknown) => void) {
       if (event === 'notification') {
@@ -33,6 +36,18 @@ const pg = vi.hoisted(() => {
     }
     async query(text: string, values?: unknown[]) {
       queries.push(text)
+      inFlight++
+      mostInFlight = Math.max(mostInFlight, inFlight)
+      try {
+        // A round trip takes time, and it is during that time that a second
+        // query can arrive on the same connection.
+        await new Promise((resolve) => setImmediate(resolve))
+        return this.answer(text, values)
+      } finally {
+        inFlight--
+      }
+    }
+    answer(text: string, values?: unknown[]) {
       if (text.startsWith('LISTEN ')) listening.push(text)
       if (text.startsWith('SELECT pg_notify')) {
         if (failNotify) throw new Error('connection lost')
@@ -57,6 +72,8 @@ const pg = vi.hoisted(() => {
     setFailNotify(value: boolean) {
       failNotify = value
     },
+    /** The most queries the connection ever held at one moment. */
+    mostInFlight: () => mostInFlight,
     /**
      * Between tests, not between processes: the connection and its LISTEN
      * belong to the instance and are made once, exactly as in production.
@@ -65,6 +82,7 @@ const pg = vi.hoisted(() => {
       notifications.length = 0
       queries.length = 0
       failNotify = false
+      mostInFlight = 0
     },
   }
 })
@@ -96,7 +114,7 @@ afterEach(() => {
 })
 
 describe('publishing', () => {
-  it('delivers here at once, without waiting for the database', () => {
+  it('delivers here at once, without waiting for the database', async () => {
     const heard: unknown[] = []
     const stop = onBusMessage((message) => heard.push(message))
 
@@ -106,6 +124,8 @@ describe('publishing', () => {
     // round trip to see a change made in this process.
     expect(heard).toEqual([{ t: 'record', change }])
     stop()
+    // Let the relay finish, so its notification is not counted by the next test.
+    await settle()
   })
 
   it('sends it on to the other instances, stamped with who sent it', async () => {
@@ -126,6 +146,40 @@ describe('publishing', () => {
     expect(pg.listening).toContain('LISTEN torqvoice_realtime')
   })
 
+  it('sends one at a time and in order, however many are published at once', async () => {
+    // A save announces several things in the same breath. The connection
+    // takes one query at a time: pg queues the rest for now, warns that it
+    // will stop, and from version 9 refuses them, which would lose the event
+    // for every other instance.
+    for (let id = 1; id <= 5; id++) {
+      publishToBus({ t: 'record', change: { ...change, id: `job-${id}` } })
+    }
+    await new Promise((resolve) => setTimeout(resolve, 50))
+
+    expect(pg.notifications.map((n) => JSON.parse(n.payload).change.id)).toEqual([
+      'job-1',
+      'job-2',
+      'job-3',
+      'job-4',
+      'job-5',
+    ])
+    expect(pg.mostInFlight()).toBe(1)
+  })
+
+  it('keeps sending after one that could not be sent', async () => {
+    const errors = vi.spyOn(console, 'error').mockImplementation(() => undefined)
+    pg.setFailNotify(true)
+    publishToBus({ t: 'record', change })
+    await settle()
+    pg.setFailNotify(false)
+
+    publishToBus({ t: 'record', change: { ...change, id: 'job-2' } })
+    await settle()
+
+    expect(errors).toHaveBeenCalled()
+    expect(pg.notifications.map((n) => JSON.parse(n.payload).change.id)).toEqual(['job-2'])
+  })
+
   it('does not fail the write when the relay cannot send', async () => {
     pg.setFailNotify(true)
     const errors = vi.spyOn(console, 'error').mockImplementation(() => undefined)

src/lib/realtime/bus.server.ts

@@ -194,17 +194,34 @@ function createBridge(connectionString: string): Bridge {
   // nobody on it writes anything, still has to hear the others.
   void ensure().catch(() => undefined)
 
+  const send = async (payload: string): Promise<void> => {
+    const connection = await ensure()
+    if (!connection) return
+    try {
+      await connection.query('SELECT pg_notify($1, $2)', [CHANNEL, payload])
+    } catch (error) {
+      // The write that caused this has already happened and must not fail
+      // because the other instances could not be told.
+      console.error('[realtime] could not relay an event:', error)
+    }
+  }
+
+  /**
+   * One at a time, in the order they were published.
+   *
+   * A connection runs one query at a time, and a save announces several
+   * things in the same breath. Handed all of them at once, pg queues them
+   * itself for now, warns that it will stop, and from version 9 refuses
+   * them, which would lose the event for every other instance. So the queue
+   * is kept here. `send` never rejects, so one that fails does not stop the
+   * ones behind it.
+   */
+  let sending: Promise<void> = Promise.resolve()
+
   return {
-    async notify(payload: string) {
-      const connection = await ensure()
-      if (!connection) return
-      try {
-        await connection.query('SELECT pg_notify($1, $2)', [CHANNEL, payload])
-      } catch (error) {
-        // The write that caused this has already happened and must not fail
-        // because the other instances could not be told.
-        console.error('[realtime] could not relay an event:', error)
-      }
+    notify(payload: string) {
+      sending = sending.then(() => send(payload))
+      return sending
     },
   }
 }