fix(client): keep publication order while an instance hydrates

src/trame_dataclass/module/protocol_v2.py

@@ -1,3 +1,6 @@
+import weakref
+from copy import deepcopy
+
 from wslink import register as export_rpc
 from wslink.websocket import LinkProtocol
 
@@ -33,6 +36,13 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.class_definitions = {}
         self.next_class_definition_id = 1
+        # Revisions belong to live model instances and are keyed by the instance's
+        # own ``_id``, not by the instance: keying by instance would require every
+        # ``StateDataModel`` subclass to be hashable (a subclass defining ``__eq__``
+        # without ``__hash__`` is not), and two equal-but-distinct models would share
+        # one revision row. ``_id`` comes from a monotonic counter, so unlike ``id()``
+        # it is never reused by a later instance.
+        self.state_revisions = {}
 
     @export_rpc("trame.dataclass.register")
     def register_instance(self, trame_dataclass):
@@ -41,6 +51,20 @@ def register_instance(self, trame_dataclass):
             self.register_definition(trame_dataclass.__class__)
             trame_dataclass.register_flush_implementation(self.push_delta)
 
+    def _track_revisions(self, obj):
+        """The revision key for *obj*, with the weak lifetime the table used to get free.
+
+        A ``finalize`` and not a ``__del__`` hook: it fires when the model is collected,
+        it does not keep the model alive, and the registry holds it for us. Retired
+        geometry therefore leaves no row behind -- which is the property
+        ``test_revision_bookkeeping_does_not_retain_a_deleted_model`` pins.
+        """
+        key = obj._id
+        if key not in self.state_revisions:
+            self.state_revisions[key] = 0
+            weakref.finalize(obj, self.state_revisions.pop, key, None)
+        return key
+
     def register_definition(self, trame_dataclass_class):
         """Register a model class definition, returning the existing one if already known."""
         if not issubclass(trame_dataclass_class, StateDataModel):
@@ -87,7 +111,10 @@ def get_state(self, instance_id):
         return {
             "id": instance_id,
             "definition": self.class_definitions[obj.__class__]["id"],
-            "state": obj.client_state,
+            # The reply may be serialized after another publication. Freeze its
+            # contents so the revision describes this snapshot, not a mutable cache.
+            "state": deepcopy(obj.client_state),
+            "revision": self.state_revisions.get(obj._id, 0),
         }
 
     @export_rpc("trame.dataclass.state.update")
@@ -103,4 +130,11 @@ def update_state(self, msg):
 
     def push_delta(self, msg):
         """Publish a state delta to all connected clients via the ``trame.dataclass.publish`` topic."""
-        self.publish("trame.dataclass.publish", msg)
+        obj = get_instance(msg["id"])
+        if obj is None:
+            self.publish("trame.dataclass.publish", msg)
+            return
+        key = self._track_revisions(obj)
+        revision = self.state_revisions[key] + 1
+        self.state_revisions[key] = revision
+        self.publish("trame.dataclass.publish", {**deepcopy(msg), "revision": revision})

tests/test_protocol_revisions.py

@@ -0,0 +1,167 @@
+import gc
+from unittest.mock import Mock
+
+from trame_dataclass.module.protocol_v2 import TrameDataclassProtocol
+from trame_dataclass.v2 import StateDataModel, Sync
+
+
+class Sample(StateDataModel):
+    value = Sync(int, 0)
+    items = Sync(list, list)
+
+
+def protocol():
+    result = TrameDataclassProtocol()
+    result.class_definitions[Sample] = {"id": 1}
+    result.publish = Mock()
+    return result
+
+
+def test_snapshot_and_delta_share_a_monotonic_instance_revision():
+    p, a, b = protocol(), Sample(), Sample()
+    assert p.get_state(a._id)["revision"] == 0
+    p.push_delta({"id": a._id, "state": {"value": 1}})
+    assert p.publish.call_args.args[1]["revision"] == 1
+    assert p.get_state(a._id)["revision"] == 1
+    p.push_delta({"id": a._id, "state": {"value": 2}})
+    assert p.publish.call_args.args[1]["revision"] == 2
+    p.push_delta({"id": b._id, "state": {"value": 3}})
+    assert p.publish.call_args.args[1]["revision"] == 1
+
+
+def test_a_snapshot_cannot_change_after_its_revision_is_returned():
+    p, a = protocol(), Sample()
+    a.items = [1, {"value": 2}]
+    reply = p.get_state(a._id)
+    a.items[1]["value"] = 3
+    a.items.append(4)
+    assert reply["state"]["items"] == [1, {"value": 2}]
+
+
+def test_a_publication_cannot_change_after_its_revision_is_assigned():
+    p, a = protocol(), Sample()
+    payload = {"id": a._id, "state": {"items": [1]}}
+    p.push_delta(payload)
+    payload["state"]["items"].append(2)
+    assert p.publish.call_args.args[1]["state"]["items"] == [1]
+    assert "revision" not in payload
+
+
+def test_revision_bookkeeping_does_not_retain_a_deleted_model():
+    p, a = protocol(), Sample()
+    p.push_delta({"id": a._id, "state": {"value": 1}})
+    assert len(p.state_revisions) == 1
+    del a
+    gc.collect()
+    assert len(p.state_revisions) == 0
+
+
+def test_a_retired_id_keeps_the_existing_missing_state_protocol():
+    p = protocol()
+    assert p.get_state("gone") == {"id": "gone", "state": None}
+    msg = {"id": "gone", "state": {"value": 2}}
+    p.push_delta(msg)
+    p.publish.assert_called_once_with("trame.dataclass.publish", msg)
+
+
+class ValueEqual(StateDataModel):
+    """A model that compares by value -- and is therefore UNHASHABLE.
+
+    Declaring ``__eq__`` sets ``__hash__ = None`` unless the class says otherwise, which
+    is ordinary Python. ``get_state`` must not require models to be hashable.
+    """
+
+    value = Sync(int, 0)
+
+    __hash__ = None
+
+    def __eq__(self, other):
+        return isinstance(other, ValueEqual) and other.value == self.value
+
+
+def test_a_model_that_compares_by_value_can_still_be_read_and_revised():
+    """MUTATION (run): key ``state_revisions`` by ``obj``. Both calls raise TypeError."""
+    p, a = protocol(), ValueEqual()
+    p.class_definitions[ValueEqual] = {"id": 2}
+    assert p.get_state(a._id)["revision"] == 0
+    p.push_delta({"id": a._id, "state": {"value": 1}})
+    assert p.publish.call_args.args[1]["revision"] == 1
+    assert p.get_state(a._id)["revision"] == 1
+
+
+def test_two_equal_but_distinct_models_keep_separate_revisions():
+    """Equality is the user's business; identity is the protocol's.
+
+    MUTATION (run): key by ``obj.value`` or any other user-visible field. The two share
+    one counter and ``b``'s first delta is announced as revision 2.
+    """
+    p, a, b = protocol(), ValueEqual(), ValueEqual()
+    p.class_definitions[ValueEqual] = {"id": 2}
+    assert a == b
+    assert a._id != b._id
+    p.push_delta({"id": a._id, "state": {"value": 7}})
+    p.push_delta({"id": a._id, "state": {"value": 8}})
+    assert p.get_state(a._id)["revision"] == 2
+    p.push_delta({"id": b._id, "state": {"value": 9}})
+    assert p.publish.call_args.args[1]["revision"] == 1
+    assert p.get_state(b._id)["revision"] == 1
+
+
+def test_collecting_one_of_two_equal_models_leaves_the_other_alone():
+    """The weak lifetime has to survive the key change, and has to be per INSTANCE.
+
+    MUTATION (run): drop the ``weakref.finalize`` in ``_track_revisions``. ``a``'s row
+    outlives it and the table grows for the life of the session.
+    """
+    p, a, b = protocol(), ValueEqual(), ValueEqual()
+    p.class_definitions[ValueEqual] = {"id": 2}
+    p.push_delta({"id": a._id, "state": {"value": 1}})
+    p.push_delta({"id": b._id, "state": {"value": 1}})
+    surviving = b._id
+    assert len(p.state_revisions) == 2
+    del a
+    gc.collect()
+    assert list(p.state_revisions) == [surviving]
+    p.push_delta({"id": b._id, "state": {"value": 2}})
+    assert p.publish.call_args.args[1]["revision"] == 2
+
+
+class HashableValueEqual(ValueEqual):
+    """Equal by value AND hashable: the shape where the defect is silent, not loud.
+
+    Nothing raises here. Two distinct instances simply shared one row, so both read
+    revision 2 after one delta each; collecting either one took the row with it and
+    reset the SURVIVOR to 0, whose next publication announced revision 1 — which a
+    client already holding 2 then discards as stale. A live update, dropped, with no
+    error anywhere. The tests below pin per-instance revisions.
+    """
+
+    def __hash__(self):
+        return 0
+
+
+def test_state_snapshot_supports_unhashable_value_equal_models():
+    p = protocol()
+    p.class_definitions[ValueEqual] = {"id": 2}
+    model = ValueEqual()
+    assert p.get_state(model._id)["state"]["value"] == 0
+    p.push_delta({"id": model._id, "state": {"value": 1}})
+    assert p.get_state(model._id)["revision"] == 1
+
+
+def test_distinct_equal_instances_keep_independent_revision_lifetimes():
+    """MUTATION (run): key by ``obj``. ``b`` reads 2 before the delete and 0 after it."""
+    p = protocol()
+    p.class_definitions[HashableValueEqual] = {"id": 3}
+    a, b = HashableValueEqual(), HashableValueEqual()
+    assert a == b
+    assert a._id != b._id
+    p.push_delta({"id": a._id, "state": {"value": 0}})
+    p.push_delta({"id": b._id, "state": {"value": 0}})
+    before = p.get_state(b._id)["revision"]
+    assert before == 1, "each instance counts its own deltas"
+    del a
+    gc.collect()
+    assert p.get_state(b._id)["revision"] == before, "retiring a reset live b revision"
+    p.push_delta({"id": b._id, "state": {"value": 0}})
+    assert p.get_state(b._id)["revision"] == before + 1

vue-components/package.json

@@ -15,7 +15,8 @@
     "dev": "vite",
     "build": "vite build --emptyOutDir",
     "lint": "eslint . --ext .vue,.js,.jsx,.cjs,.mjs --fix --ignore-path .gitignore --ignore-pattern public",
-    "semantic-release": "semantic-release"
+    "semantic-release": "semantic-release",
+    "test": "node --test tests/*.cjs"
   },
   "dependencies": {},
   "peerDependencies": {

vue-components/src/core.js

@@ -38,6 +38,8 @@ export class DataclassManager {
     this.deepReactiveWatchers = {};
     this.pendingDeepReactives = {};
     this.pendingFetch = {};
+    this.pendingPublications = {};
+    this.revisions = {};
   }
 
   connect(client) {
@@ -47,28 +49,53 @@ export class DataclassManager {
     this.subscription = client
       .getConnection()
       .getSession()
-      .subscribe("trame.dataclass.publish", async ([event]) => {
-        const { id, state } = event;
-        if (!this.dataStates[id]) {
-          await this.fetchState(id);
-          return;
-        }
+      .subscribe("trame.dataclass.publish", ([event]) =>
+        this.enqueuePublication(event),
+      );
+  }
 
-        // Capture server state for update comparison
-        updateServerState(this.dataStates[id].server, state);
-
-        for (const [key, value] of Object.entries(state)) {
-          if (this.isDataClass(id, key)) {
-            await this.handleNestedDataClass(id, key, value);
-          } else {
-            this.dataStates[id].refs[key].value = this.wrapValue(
-              id,
-              key,
-              value,
-            );
-          }
+  enqueuePublication(event) {
+    const { id } = event;
+    // Capture discovery now, before a promise callback can let a later message
+    // mistake the same instance for another first publication.
+    const discovery = !this.pendingFetch[id] && !this.dataStates[id];
+    const hydration = this.fetchState(id);
+    const previous = this.pendingPublications[id] || Promise.resolve();
+    const next = previous.then(async () => {
+      await hydration;
+      const { state, revision } = event;
+      const snapshotRevision = this.revisions[id];
+      if (Number.isInteger(revision) && Number.isInteger(snapshotRevision)) {
+        if (revision <= snapshotRevision) return;
+      } else if (discovery) {
+        // Legacy servers do not carry revisions. Their discovery publication
+        // precedes the state request and is already represented by that reply.
+        return;
+      }
+
+      // A whole asynchronous publication owns the instance until nested fetches
+      // finish. No newer callback can overwrite its cache and then be rolled back.
+      updateServerState(this.dataStates[id].server, state);
+      for (const [key, value] of Object.entries(state)) {
+        if (this.isDataClass(id, key)) {
+          await this.handleNestedDataClass(id, key, value);
+        } else {
+          this.dataStates[id].refs[key].value = this.wrapValue(id, key, value);
         }
-      });
+      }
+      if (Number.isInteger(revision)) this.revisions[id] = revision;
+    });
+    // A failed update is reported, but cannot poison all future queue entries.
+    const settled = next.catch((error) => {
+      console.error(`Dataclass publication failed for ${id}`, error);
+    });
+    this.pendingPublications[id] = settled;
+    settled.then(() => {
+      if (this.pendingPublications[id] === settled) {
+        delete this.pendingPublications[id];
+      }
+    });
+    return settled;
   }
 
   updateServer(id, name, value) {
@@ -296,7 +323,7 @@ export class DataclassManager {
     if (this.pendingFetch[id]) {
       return await this.pendingFetch[id];
     }
-    this.pendingFetch[id] = new Promise(async (resolve) => {
+    this.pendingFetch[id] = (async () => {
       const refs = { _id: id };
       const data = await this.client
         .getConnection()
@@ -349,8 +376,9 @@ export class DataclassManager {
           this.dataStates[id].refs,
         );
       }
-      resolve(this.dataStates[id]);
-    });
+      this.revisions[id] = data.revision;
+      return this.dataStates[id];
+    })();
     return await this.pendingFetch[id];
   }