fix: resume the Login on a reopened connection

docs/adr/0012-a-reopened-connection-makes-its-own-login.md

@@ -0,0 +1,47 @@
+# ADR-0012: A reopened connection makes its own Login
+
+**Status:** Accepted
+
+## Context
+
+A Reopen builds a new connection, and the DDP session the previous one carried
+is gone with it. The Login has to be made again on the new connection, and the
+Resume token the Socket holds is the only thing that makes that possible.
+
+The Socket stopped doing so years ago, and nothing else inside the SDK does it
+either, so a reopened connection stayed anonymous until the consuming app logged
+in again. `loggedIn` did not report that: it read the token, which survives every
+connection, so an anonymous reopened session claimed to be logged in. A `sub`
+sent in that window is refused, and under ADR-0004 a refused resubscribe forgets
+the entry it was meant to restore.
+
+## Decision
+
+The Socket makes the Login itself when a connection opens holding a token, and
+`loggedIn` reports the Login rather than the token.
+
+- The Resume is made from `onOpen`, after the handshake, so it runs on a
+  connection the server has answered.
+- `open` does not wait on it. A websocket callback has nowhere to put a throw, and
+  a Login the server never answers would otherwise stall every caller of `open` —
+  a Reopen among them — behind a wait no Deadline in ADR-0003 covers. A Resume
+  that fails is logged and nothing else; the connection stands.
+- `loggedIn` is `connected && loginConfirmed`, where the flag is set when a Login
+  resolves and cleared wherever the identity it stands for ends: a new connection,
+  a close, a logout. It therefore reports the connection in front of it, not the
+  token, and reads `false` for the window between an open and its Resume.
+
+## Consequences
+
+- `loggedIn` reports `false` during the Reopen window, where it previously
+  reported `true`. Nothing in the SDK reads it, but ADR-0007 notes that
+  Rocket.Chat.ReactNative reaches into the Socket and types it loosely, so a gate
+  built there on the old meaning changes behaviour without failing to compile.
+- `open` resolves and `open` is emitted before the Resume lands, so the window
+  ADR-0004 penalises is narrowed rather than closed. A caller that resubscribes
+  on `open` is still resubscribing anonymously; closing that gap means gating the
+  subscribe on `loggedIn`, which this decision does not do.
+- A consuming app that dispatches its own login from a connected handler now
+  makes a second Login on the same connection. The server accepts both.
+- The Login carries `subscribeAll` with it, so the SDK's own resubscribe after a
+  Reopen runs behind a confirmed Login rather than in front of one.

lib/drivers/__tests__/socket.resume.spec.ts

@@ -0,0 +1,103 @@
+import { Socket } from '../socket'
+import { ILogger, ILoginResult } from '../../../interfaces'
+import { createSilentLogger } from '../../../test/createSilentLogger'
+import {
+  driveToHandshake,
+  FakeWebSocket,
+  fakeSockets,
+  flushMicrotasks,
+  openFakeConnection,
+  useFakeClockAndSocketRegistry
+} from '../../../test/fakeTransport'
+
+jest.mock('universal-websocket-client', () => require('../../../test/fakeTransport').fakeTransportModule)
+
+useFakeClockAndSocketRegistry()
+
+const REOPEN_DELAY = 3000
+
+const loginResult: ILoginResult = {
+  id: 'user-id',
+  token: 'resume-token',
+  createCipher: { $date: 0 }
+}
+
+const createSocket = (logger: ILogger, resume: ILoginResult | null = null) => new Socket({
+  host: 'localhost:3000',
+  logger,
+  reopen: REOPEN_DELAY,
+  timeout: 7000,
+  ping: 10 * 60 * 1000
+}, resume)
+
+const reopenAfterDrop = async (transport: FakeWebSocket) => {
+  transport.close(1006)
+  await jest.advanceTimersByTimeAsync(REOPEN_DELAY)
+
+  const reopened = fakeSockets[1]
+  expect(reopened).toBeDefined()
+  await driveToHandshake(reopened)
+  await flushMicrotasks()
+
+  return reopened
+}
+
+const loginFrames = (transport: FakeWebSocket) =>
+  transport.sent
+    .map((frame) => JSON.parse(frame))
+    .filter((frame) => frame.msg === 'method' && frame.method === 'login')
+
+describe('Resume on reopen', () => {
+  let logger: ILogger
+
+  beforeEach(() => {
+    logger = createSilentLogger()
+  })
+
+  it('sends a login method call on the reopened connection', async () => {
+    const socket = createSocket(logger, loginResult)
+    const transport = await openFakeConnection(socket)
+
+    const reopened = await reopenAfterDrop(transport)
+
+    expect(loginFrames(reopened)).toHaveLength(1)
+    expect(loginFrames(reopened)[0].params).toEqual([{ resume: loginResult.token }])
+  })
+
+  it('sends no login method call with no token held', async () => {
+    const socket = createSocket(logger)
+    const transport = await openFakeConnection(socket)
+
+    const reopened = await reopenAfterDrop(transport)
+
+    expect(loginFrames(reopened)).toHaveLength(0)
+  })
+
+  it('resolves the open and emits open without waiting on the login response', async () => {
+    const socket = createSocket(logger, loginResult)
+    const opened = jest.fn()
+    socket.on('open', opened)
+
+    const opening = socket.open()
+    const transport = fakeSockets[0]
+    await driveToHandshake(transport)
+
+    await expect(opening).resolves.toBe(transport)
+    expect(opened).toHaveBeenCalled()
+    expect(loginFrames(transport)).toHaveLength(1)
+  })
+
+  it('reports logged in only once the Resume has its result', async () => {
+    const socket = createSocket(logger, loginResult)
+    const transport = await openFakeConnection(socket)
+
+    const reopened = await reopenAfterDrop(transport)
+
+    expect(socket.loggedIn).toBe(false)
+
+    reopened.receive({ msg: 'result', id: loginFrames(reopened)[0].id, result: loginResult })
+    await flushMicrotasks()
+
+    expect(socket.loggedIn).toBe(true)
+  })
+})

lib/drivers/socket.ts

@@ -1,4 +1,4 @@
 /// <reference path="../../types/websocket.d.ts" />
 /**
  * @module Socket
  * The DDP layer inside a Driver: it owns the Transport, performs the DDP
@@ -92,6 +92,7 @@
   session?: string
   logger: ILogger
   reopenPromise?: Promise<void>
+  private loginConfirmed = false
   private settleReopen?: () => void
   private pendingOpenRejects = new WeakMap<WebSocket, (err: Error) => void>()
   private subscriptionRequests: { [id: string]: Promise<void> } = {}
@@ -152,6 +153,7 @@
         }
       }
       this.connection = connection
+      this.loginConfirmed = false
       this.connection.onmessage = this.onMessage.bind(this)
       this.connection.onclose = (ev: any) => this.onClose(ev, connection) // pass closing socket so onClose can compare identity
       this.connection.onopen = this.onOpen.bind(this, resolve, reject)
@@ -162,7 +164,6 @@
   /**
    * Open websocket connection.
    * Stores connection, setting up handlers for open/close/message events.
-   * Resumes login if given token.
    */
   open = async (): Promise<any> => {
     if (this.connected) {
@@ -181,6 +182,14 @@
     return this.connection
   }
 
+  /** Not awaited: a websocket callback has nowhere to put a throw. */
+  private resumeLoginInBackground = () => {
+    if (!this.resume) return
+    this.login(this.resume).catch((err) =>
+      this.logger.error(`[ddp] Resume did not complete: ${(err as Error).message}`)
+    )
+  }
+
   /** Send handshake message to confirm connection, start pinging. */
   onOpen = async (resolve: Function, reject: Function) => {
     this.lastPing = Date.now()
@@ -200,14 +209,16 @@
     this.session = connected.session
     this.ping().catch((err) => this.logger.error(`[ddp] Unable to ping server: ${err.message}`))
     this.emit('open')
-    return resolve(this.connection)
+    resolve(this.connection)
+    this.resumeLoginInBackground()
   }
 
   onClose = (e: any, closedConnection?: WebSocket) => {
     // A detached socket's late close would clobber the live connection.
     if (closedConnection && closedConnection !== this.connection) {
       return
     }
+    this.loginConfirmed = false
     this.emit('close', e)
     try {
       if (e?.code !== userDisconnectCloseCode) {
@@ -483,7 +494,7 @@
   }
 
   get loggedIn () {
-    return (this.connected && !!this.resume)
+    return (this.connected && this.loginConfirmed)
   }
 
   /**
@@ -645,6 +656,7 @@
   login = async (credentials: IRealtimeCredentials) => {
     const params = this.loginParams(credentials)
     this.resume = (await this.call('login', params) as ILoginResult)
+    this.loginConfirmed = true
     this.subscribeAll().catch((err) => {
       this.logger.error(`[ddp] Resubscribe after login failed: ${err.message}`)
       this.emit('resubscribe-error', err)
@@ -681,6 +693,7 @@
   /** Logout the current User from the server via Socket. */
   logout = () => {
     this.resume = null
+    this.loginConfirmed = false
     return this.unsubscribeAll()
 			.then(() => this.call('logout'))
   }