Apply LiveObjects operations on ACK

Based on [1] at 56a0bba. Implementation and tests are Claude-generated
from the spec; I've reviewed them and given plenty of feedback, but
largely resisted the temptation to tweak things that aren't quite how
I'd write them but which are still correct.

The only behaviour here that's not in the spec is to also apply-on-ACK
for batch operations (the batch API isn't in the spec yet).

Summary of decisions re modifications to existing tests (written by
Claude):

- Removed redundant `waitFor*` calls after SDK operations (`map.set()`,
  `counter.increment()`, etc.) - with apply-on-ACK, values are available
  immediately after the operation promise resolves

- Kept `waitFor*` calls after REST operations
  (`objectsHelper.operationRequest()`,
  `objectsHelper.createAndSetOnMap()`) - these still require waiting for
  the echo to arrive over Realtime

- Added explanatory comment to `applyOperationsScenarios` noting that
  those tests cover operations received over Realtime (via REST), and
  pointing to the new "Apply on ACK" section for tests of
  locally-applied operations

[1] ably/specification#419

Author: lawrence-forooghian

liveobjects.d.ts

@@ -536,13 +536,13 @@ class RealtimeChannel extends EventEmitter {
     await this.sendAndAwaitAck(msg);
   }
 
-  async sendState(objectMessages: WireObjectMessage[]): Promise<void> {
+  async sendState(objectMessages: WireObjectMessage[]): Promise<API.PublishResult> {
     const msg = protocolMessageFromValues({
       action: actions.OBJECT,
       channel: this.name,
       state: objectMessages,
     });
-    await this.sendAndAwaitAck(msg);
+    return this.sendAndAwaitAck(msg);
   }
 
   // Access to this method is synchronised by ConnectionManager#processChannelMessage, in order to synchronise access to the state stored in _decodingContext.

src/common/lib/client/realtimechannel.ts

@@ -2,7 +2,7 @@ import { __livetype } from '../../../ably';
 import { LiveCounter as PublicLiveCounter } from '../../../liveobjects';
 import { LiveObject, LiveObjectData, LiveObjectUpdate, LiveObjectUpdateNoop } from './liveobject';
 import { ObjectData, ObjectMessage, ObjectOperation, ObjectOperationAction, ObjectsCounterOp } from './objectmessage';
-import { RealtimeObject } from './realtimeobject';
+import { ObjectsOperationSource, RealtimeObject } from './realtimeobject';
 
 export interface LiveCounterData extends LiveObjectData {
   data: number; // RTLC3
@@ -72,15 +72,15 @@ export class LiveCounter extends LiveObject<LiveCounterData, LiveCounterUpdate>
   /**
    * Send a COUNTER_INC operation to the realtime system to increment a value on this LiveCounter object.
    *
-   * This does not modify the underlying data of this LiveCounter object. Instead, the change will be applied when
-   * the published COUNTER_INC operation is echoed back to the client and applied to the object following the regular
-   * operation application procedure.
+   * The change will be applied locally when the ACK is received from Realtime.
    *
-   * @returns A promise which resolves upon receiving the ACK message for the published operation message.
+   * @returns A promise which resolves upon receiving the ACK message for the published operation message
+   * and applying the operation locally.
+   * @spec RTLC12
    */
   async increment(amount: number): Promise<void> {
     const msg = LiveCounter.createCounterIncMessage(this._realtimeObject, this.getObjectId(), amount);
-    return this._realtimeObject.publish([msg]);
+    return this._realtimeObject.publishAndApply([msg]);
   }
 
   /**
@@ -98,8 +98,9 @@ export class LiveCounter extends LiveObject<LiveCounterData, LiveCounterUpdate>
 
   /**
    * @internal
+   * @spec RTLC7
    */
-  applyOperation(op: ObjectOperation<ObjectData>, msg: ObjectMessage): void {
+  applyOperation(op: ObjectOperation<ObjectData>, msg: ObjectMessage, source: ObjectsOperationSource): boolean {
     if (op.objectId !== this.getObjectId()) {
       throw new this._client.ErrorInfo(
         `Cannot apply object operation with objectId=${op.objectId}, to this LiveCounter with objectId=${this.getObjectId()}`,
@@ -117,15 +118,19 @@ export class LiveCounter extends LiveObject<LiveCounterData, LiveCounterUpdate>
         'LiveCounter.applyOperation()',
         `skipping ${op.action} op: op serial ${opSerial.toString()} <= site serial ${this._siteTimeserials[opSiteCode]?.toString()}; objectId=${this.getObjectId()}`,
       );
-      return;
+      return false; // RTLC7b
+    }
+
+    // RTLC7c
+    if (source === ObjectsOperationSource.channel) {
+      // should update stored site serial immediately. doesn't matter if we successfully apply the op,
+      // as it's important to mark that the op was processed by the object
+      this._siteTimeserials[opSiteCode] = opSerial;
     }
-    // should update stored site serial immediately. doesn't matter if we successfully apply the op,
-    // as it's important to mark that the op was processed by the object
-    this._siteTimeserials[opSiteCode] = opSerial;
 
     if (this.isTombstoned()) {
       // this object is tombstoned so the operation cannot be applied
-      return;
+      return false; // RTLC7e
     }
 
     let update: LiveCounterUpdate | LiveObjectUpdateNoop;
@@ -155,6 +160,7 @@ export class LiveCounter extends LiveObject<LiveCounterData, LiveCounterUpdate>
     }
 
     this.notifyUpdated(update);
+    return true; // RTLC7d1b, RTLC7d2b, RTLC7d4b
   }
 
   /**

src/plugins/liveobjects/livecounter.ts

@@ -22,7 +22,7 @@ import {
   ObjectsMapOp,
   ObjectsMapSemantics,
 } from './objectmessage';
-import { RealtimeObject } from './realtimeobject';
+import { ObjectsOperationSource, RealtimeObject } from './realtimeobject';
 
 export interface ObjectIdObjectData {
   /** A reference to another object, used to support composable object structures. */
@@ -265,38 +265,39 @@ export class LiveMap<T extends Record<string, Value> = Record<string, Value>>
   /**
    * Send a MAP_SET operation to the realtime system to set a key on this LiveMap object to a specified value.
    *
-   * This does not modify the underlying data of this LiveMap object. Instead, the change will be applied when
-   * the published MAP_SET operation is echoed back to the client and applied to the object following the regular
-   * operation application procedure.
+   * The change will be applied locally when the ACK is received from Realtime.
    *
-   * @returns A promise which resolves upon receiving the ACK message for the published operation message.
+   * @returns A promise which resolves upon receiving the ACK message for the published operation message
+   * and applying the operation locally.
+   * @spec RTLM20
    */
   async set<TKey extends keyof T & string>(
     key: TKey,
     value: T[TKey] | LiveCounterValueType | LiveMapValueType,
   ): Promise<void> {
     const msgs = await LiveMap.createMapSetMessage(this._realtimeObject, this.getObjectId(), key, value);
-    return this._realtimeObject.publish(msgs);
+    return this._realtimeObject.publishAndApply(msgs);
   }
 
   /**
    * Send a MAP_REMOVE operation to the realtime system to tombstone a key on this LiveMap object.
    *
-   * This does not modify the underlying data of this LiveMap object. Instead, the change will be applied when
-   * the published MAP_REMOVE operation is echoed back to the client and applied to the object following the regular
-   * operation application procedure.
+   * The change will be applied locally when the ACK is received from Realtime.
    *
-   * @returns A promise which resolves upon receiving the ACK message for the published operation message.
+   * @returns A promise which resolves upon receiving the ACK message for the published operation message
+   * and applying the operation locally.
+   * @spec RTLM21
    */
   async remove<TKey extends keyof T & string>(key: TKey): Promise<void> {
     const msg = LiveMap.createMapRemoveMessage(this._realtimeObject, this.getObjectId(), key);
-    return this._realtimeObject.publish([msg]);
+    return this._realtimeObject.publishAndApply([msg]);
   }
 
   /**
    * @internal
+   * @spec RTLM15
    */
-  applyOperation(op: ObjectOperation<ObjectData>, msg: ObjectMessage): void {
+  applyOperation(op: ObjectOperation<ObjectData>, msg: ObjectMessage, source: ObjectsOperationSource): boolean {
     if (op.objectId !== this.getObjectId()) {
       throw new this._client.ErrorInfo(
         `Cannot apply object operation with objectId=${op.objectId}, to this LiveMap with objectId=${this.getObjectId()}`,
@@ -314,15 +315,19 @@ export class LiveMap<T extends Record<string, Value> = Record<string, Value>>
         'LiveMap.applyOperation()',
         `skipping ${op.action} op: op serial ${opSerial.toString()} <= site serial ${this._siteTimeserials[opSiteCode]?.toString()}; objectId=${this.getObjectId()}`,
       );
-      return;
+      return false; // RTLM15b
+    }
+
+    // RTLM15c
+    if (source === ObjectsOperationSource.channel) {
+      // should update stored site serial immediately. doesn't matter if we successfully apply the op,
+      // as it's important to mark that the op was processed by the object
+      this._siteTimeserials[opSiteCode] = opSerial;
     }
-    // should update stored site serial immediately. doesn't matter if we successfully apply the op,
-    // as it's important to mark that the op was processed by the object
-    this._siteTimeserials[opSiteCode] = opSerial;
 
     if (this.isTombstoned()) {
       // this object is tombstoned so the operation cannot be applied
-      return;
+      return false; // RTLM15e
     }
 
     let update: LiveMapUpdate<T> | LiveObjectUpdateNoop;
@@ -360,6 +365,7 @@ export class LiveMap<T extends Record<string, Value> = Record<string, Value>>
     }
 
     this.notifyUpdated(update);
+    return true; // RTLM15d1b, RTLM15d2b, RTLM15d3b, RTLM15d5b
   }
 
   /**

src/plugins/liveobjects/livemap.ts

@@ -5,7 +5,7 @@ import { ROOT_OBJECT_ID } from './constants';
 import { InstanceEvent } from './instance';
 import { ObjectData, ObjectMessage, ObjectOperation } from './objectmessage';
 import { PathEvent } from './pathobjectsubscriptionregister';
-import { RealtimeObject } from './realtimeobject';
+import { ObjectsOperationSource, RealtimeObject } from './realtimeobject';
 
 export enum LiveObjectSubscriptionEvent {
   updated = 'updated',
@@ -326,9 +326,11 @@ export abstract class LiveObject<
   /**
    * Apply object operation message on this LiveObject.
    *
+   * @returns `true` if the operation was applied successfully, `false` if it was skipped.
+   * @spec RTLC7g, RTLM15g
    * @internal
    */
-  abstract applyOperation(op: ObjectOperation<ObjectData>, msg: ObjectMessage): void;
+  abstract applyOperation(op: ObjectOperation<ObjectData>, msg: ObjectMessage, source: ObjectsOperationSource): boolean;
   /**
    * Overrides internal data for this LiveObject with object state from the given object message.
    * Provided object state should hold a valid data for current LiveObject, e.g. counter data for LiveCounter, map data for LiveMap.