From a580592e03687f36ab923857ae2ae610ce8ab404 Mon Sep 17 00:00:00 2001 From: Mathieu2301 <21021423+Mathieu2301@users.noreply.github.com> Date: Sat, 3 Oct 2026 11:34:04 +0200 Subject: [PATCH 1/3] fix(coordinator): carry acknowledged state through every re-declaration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RFC 0001 §7.5 resends all five slices on every declaration change, and STATE_SYNC carries a value for each owned path. The SDK resent the configured values, so an ACL-only change (revoking one guest) rolled every resident's view back to the initial state while reporting it current (reproduced with the real Go relay and browser host: owner 22 -> 21, stale:false). - StateManager records each batch the relay acknowledges (STATE_SET_OK), including one abandoned after handoff, tagged with the state-slice revision it targeted and its send order. Rejected and outcome_unknown batches are never recorded, so nothing unacknowledged is replayed. - DeclarationManager stages those values in every re-declaration of the same state revision (ACL, events, functions, reconnect, new epoch). An explicit state.declare/configure carries a new revision and stays authoritative; a rejected one leaves acknowledged values in force. - A batch pending when a re-declaration starts is acknowledged before that STATE_SYNC is (per-connection order), so the unhanded transaction is restarted with the acknowledged value instead of activating the stale one. Revocation never waits for a missing ACK. A protocol-violating late ACK after handoff triggers one follow-up transaction. - STATE_SYNC cannot declare a valueless path, so an acknowledged deletion is reapplied by one SDK-sent STATE_SET delete right after activation; a refusal is reported on coordinator.errors and the relay value becomes acknowledged. - 20 regression tests; README documents the behavior and the deletion window. Co-Authored-By: Claude Opus 5.5 --- README.md | 12 + src/coordinator.ts | 4 + src/internal/declarations.ts | 161 +++++++++++- src/internal/state.ts | 75 +++++- test/confirmed-state.test.ts | 484 +++++++++++++++++++++++++++++++++++ 5 files changed, 731 insertions(+), 5 deletions(-) create mode 100644 test/confirmed-state.test.ts diff --git a/README.md b/README.md index fe1d332..5b95eb3 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,18 @@ await coordinator.state.set([ ]); ``` +Every declaration change — an ACL, event or function slice, or a reconnect — +resends the complete state slice (RFC 0001 §7.5). It carries the value the relay +last acknowledged for each path, not the configured one, so revoking one user +never rolls the home back for the others. Only mutations confirmed by +`STATE_SET_OK` are carried, including one whose caller aborted after handoff; a +rejected mutation or one lost with its connection (`outcome_unknown`) is never +replayed, and the last acknowledged value is resent instead. An explicit +`state.declare` or `configure` replaces acknowledged values. `STATE_SYNC` cannot +declare a path without a value, so an acknowledged deletion is staged with the +declared value and then reapplied by one SDK-sent deletion right after the +declaration activates; users may briefly see the declared value in between. + Event publication returns a synchronous opaque ID and a transport-handoff promise. `sent` is not a delivery receipt; a later correlated relay rejection is reported through `coordinator.errors`. diff --git a/src/coordinator.ts b/src/coordinator.ts index f9c0e6b..5c0b6a5 100644 --- a/src/coordinator.ts +++ b/src/coordinator.ts @@ -282,6 +282,10 @@ class CoordinatorImpl implements this.calls.setActiveDeclarations(active); } + restoreDeletions(session: RelaySession, active: ActiveDeclarations, paths: readonly string[]): void { + this.state.restoreDeletions(session, active, paths); + } + declarationsReady(_receipt: DeclarationReceipt): void { const session = this.#session; if (session === undefined) return; diff --git a/src/internal/declarations.ts b/src/internal/declarations.ts index 06a4333..7e213ef 100644 --- a/src/internal/declarations.ts +++ b/src/internal/declarations.ts @@ -5,6 +5,7 @@ import type { EventDeclaration, FunctionHandler, ProtocolValue, + StateMutation, UserEventAccess, UserStateAccess, } from '../api.js'; @@ -76,6 +77,24 @@ export interface DeclarationHost { hasQueuedSnapshot: boolean, ): void; transportFailure(error: unknown): void; + /** + * Activation staged the declared value for paths whose acknowledged state is + * a deletion, which `STATE_SYNC` cannot express; the host reapplies them. + */ + restoreDeletions(session: RelaySession, active: ActiveDeclarations, paths: readonly string[]): void; +} + +/** + * The last value the relay acknowledged for one path, through `STATE_SET_OK`. + * `revision` is the state-slice revision the mutation was applied to: the entry + * only ever amends a re-declaration of that same slice, so an explicit + * `state.declare` or `configure` stays authoritative. + */ +interface ConfirmedEntry { + readonly revision: number; + readonly sequence: number; + readonly deleted: boolean; + readonly value: ProtocolValue | undefined; } interface PendingDeclaration { @@ -137,6 +156,28 @@ function copySnapshot(value: DeclarationSnapshot): DeclarationSnapshot { }); } +function protocolValuesEqual(left: ProtocolValue | undefined, right: ProtocolValue | undefined): boolean { + if (left === right) return true; + if (left instanceof Uint8Array || right instanceof Uint8Array) { + return left instanceof Uint8Array && right instanceof Uint8Array + && left.length === right.length && left.every((byte, index) => byte === right[index]); + } + if (Array.isArray(left) || Array.isArray(right)) { + return Array.isArray(left) && Array.isArray(right) && left.length === right.length + && left.every((item, index) => protocolValuesEqual(item, right[index])); + } + if (left === null || right === null || typeof left !== 'object' || typeof right !== 'object') { + return Number.isNaN(left) && Number.isNaN(right); + } + const leftKeys = Object.keys(left); + return leftKeys.length === Object.keys(right).length + && leftKeys.every((key) => Object.hasOwn(right, key) + && protocolValuesEqual( + (left as Record)[key], + (right as Record)[key], + )); +} + function revisionsEqual(left: DeclarationRevisions, right: DeclarationRevisions): boolean { return DECLARATION_DOMAINS.every((domain) => left[domain] === right[domain]); } @@ -189,6 +230,9 @@ export class DeclarationManager { #current: DeclarationTransaction | undefined; #transactionToken = 0; #stopped = false; + readonly #confirmed = new Map(); + /** A confirmed value arrived after the current transaction was handed off. */ + #confirmedAfterHandoff = false; constructor(host: DeclarationHost) { this.#host = host; @@ -275,9 +319,82 @@ export class DeclarationManager { synchronize(session: RelaySession): void { this.#requests.clear(); this.#current = undefined; + this.#confirmedAfterHandoff = false; this.#begin(session); } + /** + * Records a mutation batch the relay acknowledged with `STATE_SET_OK`. + * + * Every later re-declaration of the same state slice carries these values + * instead of the declared ones, because `STATE_SYNC` replaces the relay's + * values and an ACL, event or function change — or a reconnect — must not + * roll the home back. Only acknowledged batches are recorded: a rejected or + * unknown mutation never reaches this method, so it is never replayed. + * + * `sequence` is the order in which the batches were sent. The relay applies + * one connection's frames in order, so the later batch wins a path. + */ + recordConfirmed( + mutations: readonly StateMutation[], + revision: number, + sequence: number, + ): void { + let changedValue = false; + for (const mutation of mutations) { + const existing = this.#confirmed.get(mutation.path); + if (existing !== undefined && existing.revision === revision && existing.sequence > sequence) continue; + const deleted = 'delete' in mutation; + this.#confirmed.set(mutation.path, Object.freeze({ + revision, + sequence, + deleted, + value: deleted ? undefined : mutation.value, + })); + const current = this.#current; + if (!deleted && current !== undefined + && current.snapshot.revisions.state === revision + && Object.hasOwn(current.snapshot.state, mutation.path) + && !protocolValuesEqual(current.snapshot.state[mutation.path], mutation.value)) { + changedValue = true; + } + } + const current = this.#current; + if (!changedValue || current === undefined) return; + // The relay applied this batch before it staged the current STATE_SYNC, + // which captured the older value. Activating that stage would roll the + // mutation back while users see the result as current, so the stage is + // replaced: a new STATE_SYNC discards it on the relay (RFC 0001 §7.5). + if (!current.handedOff) { + this.#current = undefined; + this.#begin(current.session); + } else { + this.#confirmedAfterHandoff = true; + } + } + + /** + * The relay refused to reapply an acknowledged deletion, so it holds the + * declared value again. That value becomes the acknowledged one, unless a + * newer mutation of the path was acknowledged meanwhile. + */ + restorationRefused(paths: ReadonlyMap): void { + for (const [path, sequence] of paths) { + const entry = this.#confirmed.get(path); + if (entry?.deleted === true && entry.sequence === sequence) this.#confirmed.delete(path); + } + } + + /** The confirmed entry sequence of each path, for a restoration in flight. */ + confirmedSequences(paths: readonly string[]): Map { + const result = new Map(); + for (const path of paths) { + const entry = this.#confirmed.get(path); + if (entry !== undefined) result.set(path, entry.sequence); + } + return result; + } + handleFrame(frame: Frame): boolean { const domain = ackDomain(frame.opcode); if (domain === undefined) return false; @@ -407,7 +524,7 @@ export class DeclarationManager { const welcome = session.welcome; this.#current = { token: this.#transactionToken, - snapshot: copySnapshot(this.#desired), + snapshot: this.#withConfirmedState(copySnapshot(this.#desired)), session, receipt: Object.freeze({ sessionId: welcome.readySession.sessionId, @@ -479,6 +596,40 @@ export class DeclarationManager { }; } + /** + * The snapshot to send: the desired slices, with each declared path carrying + * the value the relay last acknowledged for that same state revision. + */ + #withConfirmedState(snapshot: DeclarationSnapshot): DeclarationSnapshot { + let state: Record | undefined; + for (const [path, entry] of this.#confirmed) { + if (entry.deleted + || entry.revision !== snapshot.revisions.state + || !Object.hasOwn(snapshot.state, path)) continue; + state ??= { ...snapshot.state }; + state[path] = entry.value as ProtocolValue; + } + return state === undefined + ? snapshot + : copySnapshot({ ...snapshot, state: Object.freeze(state) }); + } + + /** + * After activation, forgets what the activated declaration superseded and + * returns the acknowledged deletions the activation undid. + */ + #settleConfirmedState(snapshot: DeclarationSnapshot): string[] { + const deletions: string[] = []; + for (const [path, entry] of this.#confirmed) { + if (entry.revision !== snapshot.revisions.state || !Object.hasOwn(snapshot.state, path)) { + this.#confirmed.delete(path); + } else if (entry.deleted) { + deletions.push(path); + } + } + return deletions; + } + #activate(transaction: DeclarationTransaction): void { if (this.#current?.token !== transaction.token) return; this.#active = Object.freeze({ @@ -490,8 +641,12 @@ export class DeclarationManager { }); this.#current = undefined; this.#host.activeDeclarationsChanged(this.#active); + const deletions = this.#settleConfirmedState(transaction.snapshot); + if (deletions.length > 0) this.#host.restoreDeletions(transaction.session, this.#active, deletions); this.#settleThrough(transaction.snapshot.revisions, transaction.receipt, undefined); - if (!revisionsEqual(this.#desired.revisions, transaction.snapshot.revisions)) { + const refresh = this.#confirmedAfterHandoff; + this.#confirmedAfterHandoff = false; + if (refresh || !revisionsEqual(this.#desired.revisions, transaction.snapshot.revisions)) { this.#begin(transaction.session); } else { this.#host.declarationsReady(transaction.receipt); @@ -501,6 +656,8 @@ export class DeclarationManager { #rejectTransaction(transaction: DeclarationTransaction, failure: CoordinatorError): void { if (this.#current?.token !== transaction.token) return; this.#current = undefined; + // Nothing activated, so the relay still holds every acknowledged value. + this.#confirmedAfterHandoff = false; this.#settleThrough(transaction.snapshot.revisions, transaction.receipt, failure); const activeSnapshot = this.#active?.snapshot ?? emptySnapshot(); const desired = this.#desired; diff --git a/src/internal/state.ts b/src/internal/state.ts index 31b1423..dffbbc7 100644 --- a/src/internal/state.ts +++ b/src/internal/state.ts @@ -23,6 +23,7 @@ export interface StateHost { readySession(): RelaySession | undefined; activeDeclarations(): ActiveDeclarations | undefined; nextRequestId(): number; + emitFailure(failure: CoordinatorError): void; } interface PendingStateSet { @@ -30,10 +31,25 @@ interface PendingStateSet { epoch: Uint8Array; handedOff: boolean; abandoned: boolean; + /** What the relay applies if it acknowledges, recorded for re-declarations. */ + mutations: readonly StateMutation[]; + /** The active state-slice revision the mutation targets. */ + revision: number; + /** Send order on this coordinator, so the later of two batches wins a path. */ + sequence: number; signal?: AbortSignal; abort?: () => void; } +/** + * A corrective deletion the SDK itself sent after an activation, to reapply + * deletions the relay had acknowledged. It carries no application promise. + */ +interface PendingRestoration { + epoch: Uint8Array; + sequences: ReadonlyMap; +} + function requestId(frame: Frame): number | undefined { const value = frame.payload[0]; return typeof value === 'number' && Number.isSafeInteger(value) ? value : undefined; @@ -43,6 +59,8 @@ export class StateManager implements CoordinatorState { readonly #host: StateHost; readonly #declarations: DeclarationManager; readonly #pending = new Map(); + readonly #restorations = new Map(); + #sequence = 0; constructor(host: StateHost, declarations: DeclarationManager) { this.#host = host; @@ -77,11 +95,15 @@ export class StateManager implements CoordinatorState { }); const id = this.#host.nextRequestId(); const deferred = createDeferred(); + this.#sequence += 1; const pending: PendingStateSet = { deferred, epoch: session.welcome.epoch.slice(), handedOff: false, abandoned: false, + mutations: validated, + revision: active.snapshot.revisions.state, + sequence: this.#sequence, }; if (signal !== undefined) { const abort = () => { @@ -113,25 +135,70 @@ export class StateManager implements CoordinatorState { return deferred.promise; } + /** + * Reapplies acknowledged deletions that an activation just undid, because + * `STATE_SYNC` must stage a value for every declared path. One frame per + * activation, never retried: if it is lost, the next declaration of the same + * slice tries again; if it is refused, the relay's value is taken as the + * acknowledged one and the refusal is reported. + */ + restoreDeletions(session: RelaySession, active: ActiveDeclarations, paths: readonly string[]): void { + const wireMutations: ProtocolValue[] = []; + for (const path of paths) { + const pathId = active.stateIds.get(path); + if (pathId !== undefined) wireMutations.push([pathId, 1]); + } + if (wireMutations.length === 0 + || this.#pending.size + this.#restorations.size >= LIMITS.declarationsPerCoordinator) return; + const id = this.#host.nextRequestId(); + this.#restorations.set(id, { + epoch: session.welcome.epoch.slice(), + sequences: this.#declarations.confirmedSequences(paths), + }); + void session.send({ + opcode: Opcode.StateSet, + payload: [id, session.welcome.epoch, wireMutations], + }).then(() => undefined, () => { + this.#restorations.delete(id); + }); + } + handleFrame(frame: Frame): boolean { if (frame.opcode !== Opcode.StateSetOk) return false; const id = requestId(frame); if (id === undefined) throw new TypeError('STATE_SET_OK has no request ID'); + const restoration = this.#restorations.get(id); const pending = this.#pending.get(id); - if (pending === undefined) throw new TypeError('STATE_SET_OK is not correlated'); + const expectedEpoch = restoration?.epoch ?? pending?.epoch; + if (expectedEpoch === undefined) throw new TypeError('STATE_SET_OK is not correlated'); const epoch = frame.payload[1]; if (!(epoch instanceof Uint8Array) - || epoch.length !== pending.epoch.length - || epoch.some((value, index) => value !== pending.epoch[index])) { + || epoch.length !== expectedEpoch.length + || epoch.some((value, index) => value !== expectedEpoch[index])) { throw new TypeError('STATE_SET_OK uses a stale epoch'); } + if (restoration !== undefined) { + this.#restorations.delete(id); + return true; + } + if (pending === undefined) return true; this.#pending.delete(id); this.#removeAbort(pending); + // Recorded even when the caller abandoned the promise: the ACK proves the + // relay applied it, and a re-declaration must not roll it back. + this.#declarations.recordConfirmed(pending.mutations, pending.revision, pending.sequence); if (!pending.abandoned) pending.deferred.resolve(Object.freeze({ outcome: 'applied' })); return true; } handleError(id: number, code: number, retryable: boolean): boolean { + const restoration = this.#restorations.get(id); + if (restoration !== undefined) { + this.#restorations.delete(id); + this.#declarations.restorationRefused(restoration.sequences); + this.#host.emitFailure(relayFailure(code, retryable, 'not_dispatched')); + return true; + } const pending = this.#pending.get(id); if (pending === undefined) return false; this.#pending.delete(id); @@ -143,6 +210,7 @@ export class StateManager implements CoordinatorState { } disconnected(): void { + this.#restorations.clear(); for (const pending of this.#pending.values()) { this.#removeAbort(pending); pending.deferred.reject(pending.handedOff ? outcomeUnknown() : unavailable()); @@ -151,6 +219,7 @@ export class StateManager implements CoordinatorState { } stop(failure: CoordinatorError): void { + this.#restorations.clear(); for (const pending of this.#pending.values()) { this.#removeAbort(pending); pending.deferred.reject(pending.handedOff ? outcomeUnknown() : failure); diff --git a/test/confirmed-state.test.ts b/test/confirmed-state.test.ts new file mode 100644 index 0000000..340bb20 --- /dev/null +++ b/test/confirmed-state.test.ts @@ -0,0 +1,484 @@ +/** + * Acknowledged state survives every complete re-declaration. + * + * RFC 0001 §7.5 makes every declaration change — an ACL, an event, a function, + * a reconnect — resend all five slices, and `STATE_SYNC` carries a value for + * every owned path. If the SDK resends the values it was configured with, every + * ACL change silently rolls the home back to its initial state while users keep + * seeing it as current. These tests pin the conservative rule instead: a + * re-declaration carries the last value the relay *acknowledged*, never a + * mutation whose outcome is unknown, and an explicit `state.declare` stays + * authoritative. + */ +import { describe, expect, test } from 'bun:test'; +import type { CoordinatorConfiguration } from '../src/api.js'; +import { Opcode, type Frame, type ProtocolValue } from '../src/protocol/codec.js'; +import type { FakeRelayConnection } from './fakes/relay.js'; +import { flushMicrotasks } from './fakes/runtime.js'; +import { configuration, createTestHarness, isCoordinatorFailure, startReady } from './helpers.js'; + +const TEMPERATURE = 'home.temperature'; +const MODE = 'home.mode'; + +function twoPaths(): CoordinatorConfiguration { + return { ...configuration(), state: { [TEMPERATURE]: 20, [MODE]: 'eco' } }; +} + +function requestIdOf(frame: Frame): number { + const value = frame.payload[0]; + if (typeof value !== 'number') throw new TypeError('frame has no request ID'); + return value; +} + +function syncedValues(frame: Frame): Record { + expect(frame.opcode).toBe(Opcode.StateSync); + const values: Record = {}; + for (const entry of frame.payload[1] as ProtocolValue[][]) values[entry[0] as string] = entry[1] as ProtocolValue; + return values; +} + +function acknowledge(connection: FakeRelayConnection, frame: Frame): void { + connection.send({ opcode: Opcode.StateSetOk, payload: [requestIdOf(frame), connection.epoch, 9] }); +} + +function reject(connection: FakeRelayConnection, frame: Frame): void { + connection.send({ + opcode: Opcode.Error, + payload: [requestIdOf(frame), Opcode.StateSet, 1201, false, 'Denied'], + }); +} + +const OWNER_ONLY = [{ userId: 'owner', patterns: ['home.*'] }]; + +describe('acknowledged state across re-declaration', () => { + test('an ACL-only change resends the acknowledged value, not the configured one', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await expect(set).resolves.toEqual({ outcome: 'applied' }); + + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const exchange = await connection.acknowledgeDeclarations(); + await declared; + + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 22 }); + expect(harness.coordinator.status).toBe('ready'); + await harness.coordinator.stop(); + }); + + test('event, event ACL and function re-declarations preserve it too', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 23 }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await set; + + const declarations = [ + () => harness.coordinator.events.declare([]), + () => harness.coordinator.access.declareEvents([]), + () => harness.coordinator.functions.declare({ 'home.echo': (call) => call.arguments }), + ]; + for (const declare of declarations) { + const pending = declare(); + const exchange = await connection.acknowledgeDeclarations(); + await pending; + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 23 }); + } + await harness.coordinator.stop(); + }); + + test('a reconnect, even to a new epoch, resends the acknowledged value without resending the mutation', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await set; + + connection.close(); + await flushMicrotasks(); + await harness.runtime.advanceBy(60_000); + const next = await harness.relay.connectionAt(1); + await next.nextClientFrame(Opcode.Hello); + expect(next.epoch).not.toEqual(connection.epoch); + const exchange = await next.acknowledgeDeclarations(); + + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 22 }); + expect(next.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); + + test('a rejected mutation is not preserved', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 99 }]); + reject(connection, await connection.nextClientFrame(Opcode.StateSet)); + await set.catch(() => undefined); + + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const exchange = await connection.acknowledgeDeclarations(); + await declared; + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 20 }); + await harness.coordinator.stop(); + }); + + test('a mutation lost with its connection is never replayed; the last acknowledged value is', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const confirmed = harness.coordinator.state.set([{ path: TEMPERATURE, value: 21 }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await confirmed; + const unknown = harness.coordinator.state.set([{ path: TEMPERATURE, value: 25 }]); + await connection.nextClientFrame(Opcode.StateSet); + connection.close(); + const failure = await unknown.catch((error: unknown) => error); + expect(isCoordinatorFailure(failure) && failure.outcome).toBe('outcome_unknown'); + + await flushMicrotasks(); + await harness.runtime.advanceBy(60_000); + const next = await harness.relay.connectionAt(1); + await next.nextClientFrame(Opcode.Hello); + const exchange = await next.acknowledgeDeclarations(); + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 21 }); + expect(next.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); + + test('a mutation abandoned after handoff is preserved once its ACK proves it applied', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const controller = new AbortController(); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 26 }], { signal: controller.signal }); + const frame = await connection.nextClientFrame(Opcode.StateSet); + controller.abort(); + const failure = await set.catch((error: unknown) => error); + expect(isCoordinatorFailure(failure) && failure.outcome).toBe('outcome_unknown'); + acknowledge(connection, frame); + await flushMicrotasks(); + + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const exchange = await connection.acknowledgeDeclarations(); + await declared; + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 26 }); + await harness.coordinator.stop(); + }); + + test('competing mutations keep the relay order of their acknowledgements', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness, twoPaths()); + const first = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }, { path: MODE, value: 'comfort' }]); + const second = harness.coordinator.state.set([{ path: TEMPERATURE, value: 23 }]); + const third = harness.coordinator.state.set([{ path: MODE, value: 'away' }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + reject(connection, await connection.nextClientFrame(Opcode.StateSet)); + await Promise.allSettled([first, second, third]); + + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const exchange = await connection.acknowledgeDeclarations(); + await declared; + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 23, [MODE]: 'comfort' }); + await harness.coordinator.stop(); + }); +}); + +describe('a mutation still pending when a re-declaration starts', () => { + test('its ACK restarts the unhanded transaction, so the stale value is never activated', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + const setFrame = await connection.nextClientFrame(Opcode.StateSet); + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const staleSync = await connection.nextClientFrame(Opcode.StateSync); + expect(syncedValues(staleSync)).toEqual({ [TEMPERATURE]: 20 }); + + // The relay processes frames in order: the mutation applies, then the + // stale STATE_SYNC is staged. Its ACK therefore arrives first. + acknowledge(connection, setFrame); + await expect(set).resolves.toEqual({ outcome: 'applied' }); + const freshSync = await connection.nextClientFrame(Opcode.StateSync); + expect(syncedValues(freshSync)).toEqual({ [TEMPERATURE]: 22 }); + + // The stale stage's late acknowledgement is ignored; a new STATE_SYNC + // already discarded it on the relay. + connection.send({ + opcode: Opcode.StateSyncOk, + payload: [requestIdOf(staleSync), connection.epoch, 1, [[101, TEMPERATURE]]], + }); + await connection.acknowledgeDeclarations(freshSync); + await declared; + expect(harness.coordinator.status).toBe('ready'); + expect(connection.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); + + test('a rejected pending mutation does not disturb the transaction', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + const setFrame = await connection.nextClientFrame(Opcode.StateSet); + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const sync = await connection.nextClientFrame(Opcode.StateSync); + reject(connection, setFrame); + await set.catch(() => undefined); + await connection.acknowledgeDeclarations(sync); + await declared; + expect(syncedValues(sync)).toEqual({ [TEMPERATURE]: 20 }); + expect(connection.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); + + test('a revocation activates without waiting for a mutation ACK that never comes', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + await connection.nextClientFrame(Opcode.StateSet); + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const exchange = await connection.acknowledgeDeclarations(); + await expect(declared).resolves.toBeDefined(); + expect(exchange.stateAccess.payload[1]).toEqual([['owner', ['home.*']]]); + // Still unknown to the SDK, so the declaration carried the acknowledged value. + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 20 }); + void set.catch(() => undefined); + await harness.coordinator.stop(); + }); +}); + +describe('explicit state declarations stay authoritative', () => { + test('state.declare replaces acknowledged values, and later re-declarations keep the declared ones', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await set; + + const declaredState = harness.coordinator.state.declare({ [TEMPERATURE]: 30 }); + const first = await connection.acknowledgeDeclarations(); + await declaredState; + expect(syncedValues(first.state)).toEqual({ [TEMPERATURE]: 30 }); + + const declaredAccess = harness.coordinator.access.declareState(OWNER_ONLY); + const second = await connection.acknowledgeDeclarations(); + await declaredAccess; + expect(syncedValues(second.state)).toEqual({ [TEMPERATURE]: 30 }); + await harness.coordinator.stop(); + }); + + test('a mutation sent before state.declare and acknowledged during it does not override the declaration', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + const setFrame = await connection.nextClientFrame(Opcode.StateSet); + const declaredState = harness.coordinator.state.declare({ [TEMPERATURE]: 30 }); + const sync = await connection.nextClientFrame(Opcode.StateSync); + acknowledge(connection, setFrame); + await set; + await connection.acknowledgeDeclarations(sync); + await declaredState; + expect(syncedValues(sync)).toEqual({ [TEMPERATURE]: 30 }); + + const declaredAccess = harness.coordinator.access.declareState(OWNER_ONLY); + const next = await connection.acknowledgeDeclarations(); + await declaredAccess; + expect(syncedValues(next.state)).toEqual({ [TEMPERATURE]: 30 }); + await harness.coordinator.stop(); + }); + + test('a path removed by state.declare is not resurrected', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness, twoPaths()); + const set = harness.coordinator.state.set([{ path: MODE, value: 'away' }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await set; + const declaredState = harness.coordinator.state.declare({ [TEMPERATURE]: 20 }); + await connection.acknowledgeDeclarations(); + await declaredState; + const declaredAccess = harness.coordinator.access.declareState(OWNER_ONLY); + const next = await connection.acknowledgeDeclarations(); + await declaredAccess; + expect(syncedValues(next.state)).toEqual({ [TEMPERATURE]: 20 }); + await harness.coordinator.stop(); + }); + + test('a rejected state.declare leaves acknowledged values in force', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await set; + + const declaredState = harness.coordinator.state.declare({ [TEMPERATURE]: 30 }); + const sync = await connection.nextClientFrame(Opcode.StateSync); + connection.send({ + opcode: Opcode.Error, + payload: [requestIdOf(sync), Opcode.StateSync, 1201, false, 'Denied'], + }); + await declaredState.catch(() => undefined); + await flushMicrotasks(); + + const declaredAccess = harness.coordinator.access.declareState(OWNER_ONLY); + const next = await connection.acknowledgeDeclarations(); + await declaredAccess; + expect(syncedValues(next.state)).toEqual({ [TEMPERATURE]: 22 }); + await harness.coordinator.stop(); + }); +}); + +describe('acknowledged deletions', () => { + test('are restored by one corrective delete right after activation, then no further frame', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const deleted = harness.coordinator.state.set([{ path: TEMPERATURE, delete: true }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await deleted; + + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const exchange = await connection.acknowledgeDeclarations(); + // STATE_SYNC cannot declare a path without a value (RFC 0001 §7.2), so the + // declared value is staged and the acknowledged deletion is reapplied. + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 20 }); + const corrective = await connection.nextClientFrame(Opcode.StateSet); + expect(corrective.payload[1]).toEqual(connection.epoch); + expect(corrective.payload[2]).toEqual([[101, 1]]); + acknowledge(connection, corrective); + await declared; + await flushMicrotasks(); + expect(harness.coordinator.status).toBe('ready'); + expect(connection.queuedClientFrameCount).toBe(0); + + // A later value supersedes the deletion: nothing corrective follows. + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 19 }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await set; + const again = harness.coordinator.access.declareState([]); + const last = await connection.acknowledgeDeclarations(); + await again; + await flushMicrotasks(); + expect(syncedValues(last.state)).toEqual({ [TEMPERATURE]: 19 }); + expect(connection.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); + + test('an explicit state.declare cancels a pending restoration', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const deleted = harness.coordinator.state.set([{ path: TEMPERATURE, delete: true }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await deleted; + const declaredState = harness.coordinator.state.declare({ [TEMPERATURE]: 18 }); + const exchange = await connection.acknowledgeDeclarations(); + await declaredState; + await flushMicrotasks(); + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 18 }); + expect(connection.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); + + test('a refused restoration is reported and the relay value is taken as the acknowledged one', async () => { + const harness = createTestHarness(); + const failures: unknown[] = []; + harness.coordinator.errors.subscribe((failure) => failures.push(failure)); + const { connection } = await startReady(harness); + const deleted = harness.coordinator.state.set([{ path: TEMPERATURE, delete: true }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await deleted; + + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + await connection.acknowledgeDeclarations(); + reject(connection, await connection.nextClientFrame(Opcode.StateSet)); + await declared; + await flushMicrotasks(); + expect(failures).toHaveLength(1); + + const again = harness.coordinator.access.declareState([]); + await connection.acknowledgeDeclarations(); + await again; + await flushMicrotasks(); + expect(connection.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); +}); + +describe('defensive ordering', () => { + test('an out-of-order ACK cannot regress a path to an older batch', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const older = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + const newer = harness.coordinator.state.set([{ path: TEMPERATURE, value: 23 }]); + const olderFrame = await connection.nextClientFrame(Opcode.StateSet); + const newerFrame = await connection.nextClientFrame(Opcode.StateSet); + acknowledge(connection, newerFrame); + acknowledge(connection, olderFrame); + await Promise.all([older, newer]); + + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + const exchange = await connection.acknowledgeDeclarations(); + await declared; + expect(syncedValues(exchange.state)).toEqual({ [TEMPERATURE]: 23 }); + await harness.coordinator.stop(); + }); + + test('an ACK that arrives after the transaction was handed off triggers one bounded follow-up', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const set = harness.coordinator.state.set([{ path: TEMPERATURE, value: 22 }]); + const setFrame = await connection.nextClientFrame(Opcode.StateSet); + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + // A relay violating per-connection ordering: the ACK lands only after + // FUNCTION_SYNC was handed off, when the stale stage can no longer be replaced. + const sync = await connection.nextClientFrame(Opcode.StateSync); + connection.send({ + opcode: Opcode.StateSyncOk, + payload: [requestIdOf(sync), connection.epoch, 1, [[101, TEMPERATURE]]], + }); + for (const [request, reply] of [ + [Opcode.StateAclSync, Opcode.StateAclOk], + [Opcode.EventSync, Opcode.EventSyncOk], + [Opcode.EventAclSync, Opcode.EventAclOk], + ] as const) { + const frame = await connection.nextClientFrame(request); + connection.send({ + opcode: reply, + payload: reply === Opcode.EventSyncOk + ? [requestIdOf(frame), [[201, 'home.alert']]] + : [requestIdOf(frame), 1], + }); + } + const functions = await connection.nextClientFrame(Opcode.FunctionSync); + acknowledge(connection, setFrame); + await set; + connection.send({ opcode: Opcode.FunctionSyncOk, payload: [requestIdOf(functions), [[301, 'home.echo']]] }); + await declared; + + const followUp = await connection.acknowledgeDeclarations(); + expect(syncedValues(sync)).toEqual({ [TEMPERATURE]: 20 }); + expect(syncedValues(followUp.state)).toEqual({ [TEMPERATURE]: 22 }); + await flushMicrotasks(); + expect(harness.coordinator.status).toBe('ready'); + expect(connection.queuedClientFrameCount).toBe(0); + await harness.coordinator.stop(); + }); + + test('a restoration lost with its connection is attempted again after the reconnect', async () => { + const harness = createTestHarness(); + const { connection } = await startReady(harness); + const deleted = harness.coordinator.state.set([{ path: TEMPERATURE, delete: true }]); + acknowledge(connection, await connection.nextClientFrame(Opcode.StateSet)); + await deleted; + const declared = harness.coordinator.access.declareState(OWNER_ONLY); + await connection.acknowledgeDeclarations(); + await connection.nextClientFrame(Opcode.StateSet); + await declared; + connection.close(); + await flushMicrotasks(); + await harness.runtime.advanceBy(60_000); + + const next = await harness.relay.connectionAt(1); + await next.nextClientFrame(Opcode.Hello); + await next.acknowledgeDeclarations(); + const corrective = await next.nextClientFrame(Opcode.StateSet); + expect(corrective.payload[1]).toEqual(next.epoch); + expect(corrective.payload[2]).toEqual([[101, 1]]); + await harness.coordinator.stop(); + }); +}); From cf61d22c8c6bb143d3c01ea7cf04cbf1e639a16a Mon Sep 17 00:00:00 2001 From: Mathieu2301 <21021423+Mathieu2301@users.noreply.github.com> Date: Sat, 3 Oct 2026 09:37:11 +0000 Subject: [PATCH 2/3] test(cli): build the local binary before starter shadowing proof --- packages/cli/test/starter.test.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/cli/test/starter.test.ts b/packages/cli/test/starter.test.ts index 779b8f1..5dd6401 100644 --- a/packages/cli/test/starter.test.ts +++ b/packages/cli/test/starter.test.ts @@ -265,6 +265,12 @@ describe('the generated starter builds, typechecks, checks and runs', () => { }); test('the generated check uses its local CLI even when PATH contains an older global CLI', async () => { + // This test executes the packaged entry point, so it must build its own + // prerequisite rather than depend on another test having run first. + const cliBuild = Bun.spawnSync(['bun', 'run', 'build'], { + cwd: join(REPOSITORY, 'packages/cli'), stdout: 'pipe', stderr: 'pipe', + }); + expect(cliBuild.exitCode).toBe(0); const shadow = join(directory, 'shadow-bin'); await mkdir(shadow); await writeFile(join(shadow, 'miakapp'), '#!/bin/sh\necho stale-global-cli >&2\nexit 81\n'); From fad81e73e5644a25c978ebda0c5fc8c174687b8e Mon Sep 17 00:00:00 2001 From: Mathieu2301 <21021423+Mathieu2301@users.noreply.github.com> Date: Sat, 3 Oct 2026 09:37:11 +0000 Subject: [PATCH 3/3] chore: prepare coordinator SDK 4.0.0-alpha.1 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index c013d5b..f880b52 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "miakapi", - "version": "4.0.0-alpha.0", + "version": "4.0.0-alpha.1", "description": "Typed coordinator and trusted-browser SDK for Miakapp", "type": "module", "packageManager": "bun@1.2.23",