Merge pull request #2284 from ably/liveobjects/rto23c1-get-sync-wait-failure

[AIT-1274] LiveObjects: fail `get()`'s sync wait when the channel enters DETACHED/SUSPENDED/FAILED (RTO23c1)

Author: sacOO7

src/common/lib/client/realtimechannel.ts

@@ -928,7 +928,9 @@ class RealtimeChannel extends EventEmitter {
       this._presence.actOnChannelState(state, hasPresence, reason);
     }
     if (this._object) {
-      this._object.actOnChannelState(state, hasObjects);
+      // RTO23c1/RTO20e1 - pass `reason` explicitly: this runs before `this.errorReason` is assigned
+      // below, so the plugin cannot read it off the channel to set the sync-wait failure cause.
+      this._object.actOnChannelState(state, hasObjects, reason);
     }
     if (state === 'suspended' && this.connectionManager.state.sendEvents) {
       this.startRetryTimer();

src/plugins/liveobjects/realtimeobject.ts

@@ -1,5 +1,6 @@
 import type BaseClient from 'common/lib/client/baseclient';
 import type RealtimeChannel from 'common/lib/client/realtimechannel';
+import type ErrorInfo from 'common/lib/types/errorinfo';
 import type EventEmitter from 'common/lib/util/eventemitter';
 import type * as API from '../../../ably';
 import type { ChannelState, StatusSubscription } from '../../../ably';
@@ -19,6 +20,17 @@ export enum ObjectsEvent {
   synced = 'synced',
 }
 
+/**
+ * Internal-only signals emitted on `_eventEmitterInternal` (never on `_eventEmitterPublic`), so they
+ * are not observable through the public `RealtimeObject#on` API.
+ */
+enum ObjectsInternalEvent {
+  // RTO23c1 / RTO20e1 - emitted when the channel transitions into DETACHED/SUSPENDED/FAILED, so that
+  // parked objects-sync waiters (get()/publishAndApply) can fail. Carries the channel state and its
+  // errorReason as emit arguments.
+  syncWaitFailed = 'syncWaitFailed',
+}
+
 /** @spec RTO22 */
 export enum ObjectsOperationSource {
   local = 'local',
@@ -39,6 +51,35 @@ const StateToEventsMap: Record<ObjectsState, ObjectsEvent | undefined> = {
 
 export type ObjectsEventCallback = () => void;
 
+/**
+ * Remediation for a `get()` sync wait failing (RTO23c1). The rejection is recoverable, but the
+ * recovery differs per state: `ensureAttached` at `get()` entry re-attaches a DETACHED channel
+ * itself and proceeds through SUSPENDED (the SDK re-attaches when the connection recovers), so a
+ * plain retry suffices for those two states, whereas from FAILED `get()` rejects at entry (90001)
+ * until the channel is explicitly re-attached.
+ */
+function getSyncWaitFailureRemediation(state: ChannelState): string {
+  switch (state) {
+    case 'detached':
+      return 'Retry channel.object.get(). The retried call re-attaches the channel and waits for a fresh objects sync.';
+    case 'suspended':
+      return 'Retry channel.object.get() once the channel re-attaches. The SDK re-attaches suspended channels automatically when the connection recovers, or call channel.attach() to retry now.';
+    default:
+      return 'Inspect the cause for the underlying failure. Call channel.attach() to recover the channel, then retry channel.object.get(). Calling channel.object.get() on a failed channel without re-attaching first rejects immediately.';
+  }
+}
+
+/**
+ * Remediation for a `publishAndApply` sync wait failing (RTO20e1), reached via the public mutation
+ * APIs (`LiveMap.set`/`remove`, `LiveCounter.increment`/`decrement`, batch). The operation was
+ * published and ACKed before the wait started, so it is persisted server-side and must not be
+ * retried; only the local optimistic apply failed, and the local object converges on the next
+ * successful attach and objects sync. State-independent, unlike the `get()` remediation.
+ */
+function publishSyncWaitFailureRemediation(): string {
+  return 'Do not retry the operation. It was already published and acknowledged by Ably, so retrying would apply it twice. The local object converges automatically on the next successful attach and objects sync. Inspect the cause and channel.errorReason for why the channel left the attached state.';
+}
+
 export class RealtimeObject {
   gcGracePeriod: number;
 
@@ -91,9 +132,9 @@ export class RealtimeObject {
     // implicit attach before proceeding
     await this._channel.ensureAttached();
 
-    // if we're not synced yet, wait for sync sequence to finish before returning root
+    // RTO23c - if we're not synced yet, wait for sync sequence to finish before returning root
     if (this._state !== ObjectsState.synced) {
-      await this._eventEmitterInternal.once(ObjectsEvent.synced); // RTO1c
+      await this._waitForSyncedOrChannelFailure('the object could not be retrieved', getSyncWaitFailureRemediation); // RTO23c1
     }
 
     const pathObject = new DefaultPathObject(this, this._objectsPool.getRoot(), []);
@@ -234,24 +275,32 @@ export class RealtimeObject {
    * @spec RTO4 - handling of the `ATTACHED` transition
    * @spec RTO27 - manage the stored objects data across non-`ATTACHED` transitions
    */
-  actOnChannelState(state: ChannelState, hasObjects?: boolean): void {
+  actOnChannelState(state: ChannelState, hasObjects?: boolean, reason?: ErrorInfo | null): void {
     switch (state) {
       case 'attached':
         // RTO4 - ATTACHED is handled by onAttached (the sync lifecycle); it is outside RTO27's scope
         this.onAttached(hasObjects);
         break;
 
       case 'detached':
+      case 'suspended':
       case 'failed':
-        // RTO27a - the actual current state of Objects data is unknown in these states, so clear it
-        // without emitting update events (RTO27a1); the objects themselves remain in the pool.
-        this._objectsPool.clearObjectsData(false); // RTO27a1
-        this._syncObjectsPool.clear(); // RTO27a2
+        // RTO23c1 / RTO20e1 - fail any parked objects-sync waiters (get()/publishAndApply) before the
+        // RTO27a data clearing below (drain-then-clear, matching ably-cocoa). The channel's errorReason
+        // is not yet assigned when notifyState invokes this handler, so it is passed in as `reason`.
+        // Emitted unconditionally; a no-op when no waiter is parked.
+        this._eventEmitterInternal.emit(ObjectsInternalEvent.syncWaitFailed, state, reason);
+
+        if (state !== 'suspended') {
+          // RTO27a - the actual current state of Objects data is unknown in DETACHED/FAILED, so clear it
+          // without emitting update events (RTO27a1); the objects themselves remain in the pool.
+          this._objectsPool.clearObjectsData(false); // RTO27a1
+          this._syncObjectsPool.clear(); // RTO27a2
+        }
+        // RTO27b - SUSPENDED (and every unlisted state: INITIALIZED, ATTACHING, DETACHING) retains the
+        // objects data unchanged. For SUSPENDED in particular the connection may still recover, so the
+        // retained data remains a valid best-effort local copy.
         break;
-
-      // RTO27b - every other state (SUSPENDED, INITIALIZED, ATTACHING, DETACHING) is intentionally
-      // not handled here: the objects data is retained unchanged. For SUSPENDED in particular, the
-      // connection may still recover, so the retained data remains a valid best-effort local copy.
     }
   }
 
@@ -356,30 +405,10 @@ export class RealtimeObject {
         `waiting for sync to complete before applying ${syntheticMessages.length} message(s); channel=${this._channel.name}`,
       );
 
-      await new Promise<void>((resolve, reject) => {
-        const cleanup = () => {
-          this._eventEmitterInternal.off(onSynced);
-          this._channel.internalStateChanges.off(onChannelState);
-        };
-        const onSynced = () => {
-          cleanup();
-          resolve();
-        };
-        // RTO20e1
-        const onChannelState = () => {
-          cleanup();
-          reject(
-            new this._client.ErrorInfo(
-              `the operation could not be applied locally due to the channel entering the ${this._channel.state} state whilst waiting for objects sync to complete`,
-              92008,
-              400,
-              this._channel.errorReason || undefined,
-            ),
-          );
-        };
-        this._eventEmitterInternal.once(ObjectsEvent.synced, onSynced);
-        this._channel.internalStateChanges.once(['detached', 'suspended', 'failed'], onChannelState);
-      });
+      await this._waitForSyncedOrChannelFailure(
+        'the operation could not be applied locally',
+        publishSyncWaitFailureRemediation,
+      ); // RTO20e1
     }
 
     // RTO20f - Apply synthetic messages
@@ -597,6 +626,48 @@ export class RealtimeObject {
     }
   }
 
+  /**
+   * Waits for the objects sync state to reach SYNCED, rejecting with a 92008 error if the channel
+   * first transitions into DETACHED/SUSPENDED/FAILED (signalled by `actOnChannelState` via the
+   * internal `syncWaitFailed` event). Shared by `get()` (RTO23c1) and `publishAndApply` (RTO20e1),
+   * which differ in the error message's `failureDescription` prefix and in their caller-specific
+   * `remediation` (resolved per failure state, since the recovery advice depends on it); the error's
+   * code (92008), statusCode (400), and cause (the channel's errorReason) are mandated identically
+   * by both spec points, which say nothing about remediation. The cause is the state-change
+   * `reason` — the same error `notifyState` assigns to `RealtimeChannel.errorReason`; on a
+   * reason-less transition (e.g. a clean detach) the cause is deliberately absent rather than a
+   * stale prior errorReason. Both listeners are removed on either outcome (no leaks).
+   */
+  private _waitForSyncedOrChannelFailure(
+    failureDescription: string,
+    remediation: (state: ChannelState) => string,
+  ): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+      const cleanup = () => {
+        this._eventEmitterInternal.off(ObjectsEvent.synced, onSynced);
+        this._eventEmitterInternal.off(ObjectsInternalEvent.syncWaitFailed, onChannelFailure);
+      };
+      const onSynced = () => {
+        cleanup();
+        resolve();
+      };
+      const onChannelFailure = (state: ChannelState, reason?: ErrorInfo | null) => {
+        cleanup();
+        reject(
+          new this._client.ErrorInfo({
+            message: `${failureDescription} due to the channel entering the ${state} state whilst waiting for objects sync to complete`,
+            code: 92008,
+            statusCode: 400,
+            cause: reason || undefined,
+            remediation: remediation(state),
+          }),
+        );
+      };
+      this._eventEmitterInternal.once(ObjectsEvent.synced, onSynced);
+      this._eventEmitterInternal.once(ObjectsInternalEvent.syncWaitFailed, onChannelFailure);
+    });
+  }
+
   private _stateChange(state: ObjectsState): void {
     if (this._state === state) {
       return;

test/uts/objects/unit/objects_pool.test.ts

@@ -360,6 +360,69 @@ describe('uts/objects/unit/objects_pool', function () {
     expect(rto._objectsPool.get('counter:new@1000')).to.exist;
   });
 
+  // UTS: objects/unit/RTO5a5/absent-channel-serial-0
+  it('RTO5a5 - OBJECT_SYNC with no channelSerial is a single-message sync', async function () {
+    const { channel, mockWs } = await setupManualChannel('test-RTO5a5', {
+      onMessageFromClient: (msg, ws) => {
+        if (msg.action === PM_ACTION.ATTACH) {
+          ws.active_connection!.send_to_client({
+            action: PM_ACTION.ATTACHED,
+            channel: msg.channel,
+            channelSerial: 'sync1:cursor',
+            flags: HAS_OBJECTS,
+          });
+          // RTO5a5 - no channelSerial: the whole sync is contained in this one message, so the
+          // objects are applied and the sync completes (SYNCED) without waiting for a cursor-empty
+          // channelSerial (RTO5a4)
+          ws.active_connection!.send_to_client(
+            buildObjectSyncMessage(msg.channel, null as any, [
+              buildObjectState('counter:new@1000', { aaa: 't:0' }, { counter: { count: 99 } }),
+            ]),
+          );
+        }
+      },
+    });
+
+    // get() waits for SYNCED, so it doubles as the synchronization point for the applied sync
+    await channel.object.get();
+
+    const rto = getRealtimeObject(channel);
+    expect(rto._state).to.equal('synced');
+    expect(rto._objectsPool.get('counter:new@1000')).to.exist;
+  });
+
+  // UTS: objects/unit/RTO5a6/malformed-channel-serial-treated-as-absent-0
+  it('RTO5a6 - malformed channelSerial is treated as absent', async function () {
+    const { channel, mockWs } = await setupManualChannel('test-RTO5a6', {
+      onMessageFromClient: (msg, ws) => {
+        if (msg.action === PM_ACTION.ATTACH) {
+          ws.active_connection!.send_to_client({
+            action: PM_ACTION.ATTACHED,
+            channel: msg.channel,
+            channelSerial: 'sync1:cursor',
+            flags: HAS_OBJECTS,
+          });
+          // RTO5a6 - "malformedserialnocolon" has no ':' separator, so it cannot be parsed per
+          // RTO5a1; it must be handled as if the channelSerial were absent (RTO5a5): the objects
+          // are applied and the sync completes (SYNCED)
+          ws.active_connection!.send_to_client(
+            buildObjectSyncMessage(msg.channel, 'malformedserialnocolon', [
+              buildObjectState('counter:new@1000', { aaa: 't:0' }, { counter: { count: 99 } }),
+            ]),
+          );
+        }
+      },
+    });
+
+    // get() waits for SYNCED, so it doubles as the synchronization point for the applied sync
+    await channel.object.get();
+
+    const rto = getRealtimeObject(channel);
+    // Treated as absent (RTO5a5): the message was applied and the sync ended
+    expect(rto._state).to.equal('synced');
+    expect(rto._objectsPool.get('counter:new@1000')).to.exist;
+  });
+
   // UTS: objects/unit/RTO5f2a/partial-map-merge-0
   it('RTO5f2a - partial object state merge for maps', async function () {
     const { channel, mockWs } = await setupManualChannel('test-RTO5f2a', {