From b3307e56ad90552e555106088e4c57bae80bed76 Mon Sep 17 00:00:00 2001
From: Ahinoam Pollack <ahinoamp@gmail.com>
Date: Fri, 2 Oct 2026 13:54:46 +0200
Subject: [PATCH] fix(client): keep publication order through hydration

A publication could be lost or applied out of order while an instance was
still hydrating:

- a delta arriving before the class definition threw on
  `dataclass_containers` and was dropped;
- a delta arriving while nested refs were still being fetched wrote to a
  ref that did not exist yet;
- a delta arriving during the first `state.get` was matched to the pending
  fetch and never applied;
- an older async handler awaiting a child fetch could finish after a newer
  one and overwrite its value.

Client: every publication now waits for the instance's whole `fetchState`
(nested children included) and runs through a per-instance promise queue;
a failed update settles only its own entry.

Server: snapshots and deltas carry a per-instance revision, so a replayed
delta that a newer snapshot already contains is dropped instead of rolling
the value back. Revisions are keyed by the model's `_id`, so models need
not be hashable and equal-but-distinct models keep separate counters.
Servers without revisions keep the previous behaviour.

Adds tests/test_protocol_revisions.py and vue-components/tests/
publication-order.cjs (`npm test`).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/trame_dataclass/module/protocol_v2.py  |  38 ++-
 tests/test_protocol_revisions.py           | 167 ++++++++++++++
 vue-components/package.json                |   3 +-
 vue-components/src/core.js                 |  74 ++++--
 vue-components/tests/publication-order.cjs | 255 +++++++++++++++++++++
 5 files changed, 511 insertions(+), 26 deletions(-)
 create mode 100644 tests/test_protocol_revisions.py
 create mode 100644 vue-components/tests/publication-order.cjs

diff --git a/src/trame_dataclass/module/protocol_v2.py b/src/trame_dataclass/module/protocol_v2.py
index 7efee0a..2c3d976 100644
--- a/src/trame_dataclass/module/protocol_v2.py
+++ b/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})
diff --git a/tests/test_protocol_revisions.py b/tests/test_protocol_revisions.py
new file mode 100644
index 0000000..06f1f02
--- /dev/null
+++ b/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
diff --git a/vue-components/package.json b/vue-components/package.json
index 1a23399..706beb3 100644
--- a/vue-components/package.json
+++ b/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": {
diff --git a/vue-components/src/core.js b/vue-components/src/core.js
index 3472b22..f252305 100644
--- a/vue-components/src/core.js
+++ b/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];
   }
 
diff --git a/vue-components/tests/publication-order.cjs b/vue-components/tests/publication-order.cjs
new file mode 100644
index 0000000..dddd6cf
--- /dev/null
+++ b/vue-components/tests/publication-order.cjs
@@ -0,0 +1,255 @@
+/* eslint-env node */
+const assert = require("node:assert/strict");
+const test = require("node:test");
+const fs = require("node:fs");
+const path = require("node:path");
+const vm = require("node:vm");
+const Vue = require("vue");
+const source =
+  fs
+    .readFileSync(
+      process.env.DATACLASS_CORE || path.join(__dirname, "../src/core.js"),
+      "utf8",
+    )
+    .replace(
+      'import { ref, reactive, watch } from "vue";',
+      "const { ref, reactive, watch } = Vue;",
+    )
+    .replace("export class DataclassManager", "class DataclassManager") +
+  "\nmodule.exports = { DataclassManager };";
+function deferred() {
+  let resolve, reject;
+  const promise = new Promise((a, b) => {
+    resolve = a;
+    reject = b;
+  });
+  return { promise, resolve, reject };
+}
+const tick = () => new Promise((resolve) => setImmediate(resolve));
+const definition = (containers = []) => ({
+  dataclass_containers: containers,
+  client_only: [],
+  deep_reactive: [],
+});
+const state = (values, type = "Layer", revision = 0, hold = null) => ({
+  definition: type,
+  state: values,
+  revision,
+  hold,
+});
+function fixture(states, definitions = { Layer: definition() }) {
+  const errors = [],
+    calls = [];
+  const context = {
+    Vue,
+    module: { exports: {} },
+    console: { error: (...args) => errors.push(args) },
+  };
+  vm.runInNewContext(source, context);
+  const manager = new context.module.exports.DataclassManager();
+  let publication;
+  const session = {
+    async call(name, args) {
+      calls.push([name, args]);
+      if (name === "trame.dataclass.state.get") {
+        const original = states[args[0]];
+        const snapshot = structuredClone({
+          definition: original.definition,
+          state: original.state,
+          revision: original.revision,
+        });
+        if (original.hold) await original.hold.promise;
+        return snapshot;
+      }
+      if (name === "trame.dataclass.definition.get") {
+        const result = definitions[args[0]];
+        return result.promise || result;
+      }
+      return null;
+    },
+    subscribe(topic, callback) {
+      publication = callback;
+      return {};
+    },
+  };
+  manager.connect({ getConnection: () => ({ getSession: () => session }) });
+  return {
+    manager,
+    errors,
+    calls,
+    publish: (id, values, revision) =>
+      publication([{ id, state: values, revision }]),
+    value: (id, key) => manager.dataStates[id].refs[key].value,
+  };
+}
+
+test("publication waits for the missing class definition", async () => {
+  const gate = deferred(),
+    f = fixture({ layer: state({ visible: false }) }, { Layer: gate });
+  const hydration = f.manager.fetchState("layer");
+  await tick();
+  const update = f.publish("layer", { visible: true }, 1);
+  await tick();
+  assert.equal(f.manager.dataStates.layer.refs.visible, undefined);
+  gate.resolve(definition());
+  await Promise.all([hydration, update]);
+  assert.equal(f.value("layer", "visible"), true);
+  assert.deepEqual(f.errors, []);
+});
+
+test("publication waits for nested refs after the definition arrives", async () => {
+  const gate = deferred(),
+    f = fixture(
+      {
+        parent: state({ child: "child", title: "old" }, "Parent"),
+        child: state({ value: 1 }, "Child"),
+      },
+      { Parent: definition(["child"]), Child: gate },
+    );
+  const hydration = f.manager.fetchState("parent");
+  await tick();
+  const update = f.publish("parent", { title: "new" }, 1);
+  await tick();
+  assert.equal(f.manager.dataStates.parent.refs.title, undefined);
+  gate.resolve(definition());
+  await Promise.all([hydration, update]);
+  assert.equal(f.value("parent", "title"), "new");
+  assert.deepEqual(f.errors, []);
+});
+
+test("a newer delta survives the first outstanding state reply", async () => {
+  const gate = deferred(),
+    f = fixture({ layer: state({ visible: false }, "Layer", 0, gate) });
+  const hydration = f.manager.fetchState("layer");
+  const update = f.publish("layer", { visible: true }, 1);
+  await tick();
+  gate.resolve();
+  await Promise.all([hydration, update]);
+  assert.equal(f.value("layer", "visible"), true);
+  assert.deepEqual(f.errors, []);
+});
+
+test("a newer publication cannot overtake an older callback fetching a child", async () => {
+  const gate = deferred(),
+    f = fixture(
+      {
+        parent: state({ children: [], visible: false }, "Parent"),
+        child: state({ value: 1 }, "Child"),
+      },
+      { Parent: definition(["children"]), Child: gate },
+    );
+  await f.manager.fetchState("parent");
+  const older = f.publish("parent", { children: ["child"], visible: false }, 1);
+  await tick();
+  const newer = f.publish("parent", { visible: true }, 2);
+  await tick();
+  assert.equal(f.value("parent", "visible"), false);
+  gate.resolve(definition());
+  await Promise.all([older, newer]);
+  await Vue.nextTick();
+  assert.equal(f.value("parent", "visible"), true);
+  assert.deepEqual(f.errors, []);
+  assert.equal(
+    f.calls.filter(([name]) => name === "trame.dataclass.state.update").length,
+    0,
+  );
+});
+
+test("a snapshot revision subsumes older and equal publications without rollback", async () => {
+  const f = fixture({ layer: state({ value: "newest" }, "Layer", 3) });
+  const first = f.publish("layer", { value: "old" }, 1);
+  const equal = f.publish("layer", { value: "also old" }, 3);
+  await Promise.all([first, equal]);
+  assert.equal(f.value("layer", "value"), "newest");
+  await f.publish("layer", { value: "later" }, 4);
+  assert.equal(f.value("layer", "value"), "later");
+});
+
+test("a legacy discovery snapshot is not overwritten by its older discovery message", async () => {
+  const f = fixture({ layer: state({ value: "snapshot" }, "Layer", null) });
+  await f.publish("layer", { value: "discovery" });
+  assert.equal(f.value("layer", "value"), "snapshot");
+  await f.publish("layer", { value: "subsequent" });
+  assert.equal(f.value("layer", "value"), "subsequent");
+});
+
+test("a legacy in-flight snapshot still replays a newer publication", async () => {
+  const gate = deferred(),
+    f = fixture({ layer: state({ value: "old" }, "Layer", null, gate) });
+  const hydration = f.manager.fetchState("layer");
+  const update = f.publish("layer", { value: "new" });
+  gate.resolve();
+  await Promise.all([hydration, update]);
+  assert.equal(f.value("layer", "value"), "new");
+});
+
+test("one blocked instance does not hold unrelated updates", async () => {
+  const gate = deferred(),
+    f = fixture({
+      blocked: state({ value: 0 }, "Layer", 0, gate),
+      live: state({ value: 0 }),
+    });
+  const hydration = f.manager.fetchState("blocked");
+  const blocked = f.publish("blocked", { value: 1 }, 1);
+  await f.manager.fetchState("live");
+  await f.publish("live", { value: 2 }, 1);
+  assert.equal(f.value("live", "value"), 2);
+  gate.resolve();
+  await Promise.all([hydration, blocked]);
+  assert.equal(f.value("blocked", "value"), 1);
+});
+
+test("a rejected publication is reported and does not poison subsequent publications", async () => {
+  const f = fixture({ layer: state({ value: 0 }) });
+  await f.manager.fetchState("layer");
+  await f.publish("layer", { undeclared: 1 }, 1);
+  assert.equal(f.errors.length, 1);
+  await f.publish("layer", { value: 2 }, 2);
+  assert.equal(f.value("layer", "value"), 2);
+});
+
+test("settling an old publication does not remove a newer active queue", async () => {
+  const gate = deferred(),
+    f = fixture(
+      {
+        parent: state({ children: [], visible: false }, "Parent"),
+        child: state({ value: 1 }, "Child"),
+      },
+      { Parent: definition(["children"]), Child: gate },
+    );
+  await f.manager.fetchState("parent");
+  const first = f.publish("parent", { visible: true }, 1);
+  const second = f.publish(
+    "parent",
+    { children: ["child"], visible: false },
+    2,
+  );
+  await first;
+  await tick();
+  const third = f.publish("parent", { visible: true }, 3);
+  await tick();
+  gate.resolve(definition());
+  await Promise.all([second, third]);
+  assert.equal(f.value("parent", "visible"), true);
+  assert.deepEqual(f.errors, []);
+});
+
+test("an RPC rejection rejects hydration rather than leaving a promise pending", async () => {
+  const gate = deferred(),
+    f = fixture({ layer: state({ value: 0 }, "Layer", 0, gate) });
+  const hydration = f.manager.fetchState("layer");
+  const rejected = assert.rejects(hydration, /offline/);
+  gate.reject(new Error("offline"));
+  await rejected;
+});
+
+test("only an integer revision advances the last applied server version", async () => {
+  const f = fixture({ layer: state({ value: "snapshot" }, "Layer", 3) });
+  await f.manager.fetchState("layer");
+  await f.publish("layer", { value: "four" }, 4);
+  await f.publish("layer", { value: "stale duplicate" }, 4);
+  assert.equal(f.value("layer", "value"), "four");
+  await f.publish("layer", { value: "legacy" });
+  await f.publish("layer", { value: "still stale" }, 3);
+  assert.equal(f.value("layer", "value"), "legacy");
+});