From 0179d3d088664c84fd08aa3e4b12472c255d9c43 Mon Sep 17 00:00:00 2001 From: Sakana <15715093608@163.com> Date: Wed, 21 Oct 2026 21:31:16 +0800 Subject: [PATCH 1/2] fix(shared-state): keep initial snapshots local --- .../src/client/rpc-shared-state.test.ts | 75 +++++++++++++++++++ .../devframe/src/client/rpc-shared-state.ts | 14 ++-- 2 files changed, 84 insertions(+), 5 deletions(-) diff --git a/packages/devframe/src/client/rpc-shared-state.test.ts b/packages/devframe/src/client/rpc-shared-state.test.ts index f5779a06..c56899ea 100644 --- a/packages/devframe/src/client/rpc-shared-state.test.ts +++ b/packages/devframe/src/client/rpc-shared-state.test.ts @@ -1,5 +1,6 @@ import { createEventEmitter } from 'devframe/utils/events' import { describe, expect, it } from 'vitest' +import { DEVFRAME_EVENTS } from '../events' import { createRpcSharedStateClientHost } from './rpc-shared-state' function makeFakeRpc() { @@ -50,3 +51,77 @@ describe('client shared state', () => { expect(setCalls).toHaveLength(1) // exactly one server-state:set, not two }) }) + +describe('initial shared-state snapshots', () => { + function setup() { + const { rpc, setCalls } = makeFakeRpc() + const snapshot = Promise.withResolvers() + const handlers = new Map void>() + rpc.isTrusted = true + rpc.call = () => snapshot.promise + rpc.client.register = (definition: any) => handlers.set(definition.name, definition.handler) + const host = createRpcSharedStateClientHost(rpc) + return { rpc, host, snapshot, handlers, setCalls } + } + + it('reads a snapshot without writing it back to the node side', async () => { + const { host, snapshot, setCalls } = setup() + const state = await host.get('k', { initialValue: { count: 0, localDefault: true } }) + snapshot.resolve({ count: 1 }) + await snapshot.promise + expect(state.value()).toEqual({ count: 1, localDefault: true }) + expect(setCalls).toEqual([]) + state.mutate((draft) => { + draft.count = 2 + }) + expect(setCalls[0]?.[1]).toEqual({ count: 2, localDefault: true }) + }) + + it('keeps an update received while the initial snapshot is pending', async () => { + const { host, snapshot, handlers } = setup() + const state = await host.get('k', { initialValue: { count: 0 } }) + handlers.get(DEVFRAME_EVENTS.broadcast.clientStateUpdated)!('k', { count: 2 }, 'newer') + snapshot.resolve({ count: 1 }) + await snapshot.promise + expect(state.value()).toEqual({ count: 2 }) + }) + + it('keeps a local write made while the initial snapshot is pending', async () => { + const { host, snapshot, setCalls } = setup() + const state = await host.get('k', { initialValue: { count: 0 } }) + state.mutate((draft) => { + draft.count = 2 + }) + snapshot.resolve({ count: 1 }) + await snapshot.promise + expect(state.value()).toEqual({ count: 2 }) + expect(setCalls).toHaveLength(1) + }) + + it('forwards a local write triggered by the initial snapshot', async () => { + const { host, snapshot, setCalls } = setup() + const state = await host.get('k', { initialValue: { count: 0 } }) + state.on('updated', (value) => { + if (value.count === 1) { + state.mutate((draft) => { + draft.count = 2 + }) + } + }) + snapshot.resolve({ count: 1 }) + await snapshot.promise + expect(state.value()).toEqual({ count: 2 }) + expect(setCalls).toHaveLength(1) + expect(setCalls[0]?.[1]).toEqual({ count: 2 }) + }) + + it('ignores a snapshot after its state is deleted', async () => { + const { host, snapshot, setCalls } = setup() + const state = await host.get('k', { initialValue: { count: 0 } }) + host.delete('k') + snapshot.resolve({ count: 1 }) + await snapshot.promise + expect(state.value()).toEqual({ count: 0 }) + expect(setCalls).toEqual([]) + }) +}) diff --git a/packages/devframe/src/client/rpc-shared-state.ts b/packages/devframe/src/client/rpc-shared-state.ts index 629aee9d..8d211110 100644 --- a/packages/devframe/src/client/rpc-shared-state.ts +++ b/packages/devframe/src/client/rpc-shared-state.ts @@ -1,6 +1,7 @@ import type { RpcSharedStateGetOptions, RpcSharedStateHost } from 'devframe/types' import type { SharedState, SharedStatePatch } from 'devframe/utils/shared-state' import type { DevframeRpcClient } from './rpc' +import { nanoid } from 'devframe/utils/nanoid' import { createSharedState } from 'devframe/utils/shared-state' import { DEVFRAME_EVENTS } from '../events' @@ -42,10 +43,10 @@ export function createRpcSharedStateClientHost(rpc: DevframeRpcClient): RpcShare }, }) - function registerSharedState(key: string, state: SharedState) { + function registerSharedState(key: string, state: SharedState, initialSyncId?: string) { const offs: (() => void)[] = [] offs.push(state.on('updated', (fullState, patches, syncId) => { - if (isStaticBackend) + if (isStaticBackend || syncId === initialSyncId) return if (patches) { rpc.callEvent('devframe:rpc:server-state:patch', key, patches, syncId) @@ -99,15 +100,18 @@ export function createRpcSharedStateClientHost(rpc: DevframeRpcClient): RpcShare sharedState.set(key, state) for (const fn of keyAddedListeners) fn(key) + const initialState = state.value() + const initialSyncId = nanoid() rpc.call('devframe:rpc:server-state:get', key) .then((serverState) => { - if (serverState !== undefined) - state.mutate(() => mergeWithInitialValue(key, serverState)) + // A pending read must not replace a newer update or a deleted state. + if (serverState !== undefined && sharedState.get(key) === state && state.value() === initialState) + state.mutate(() => mergeWithInitialValue(key, serverState), initialSyncId) }) .catch((error) => { console.error('Error getting server state', error) }) - stateDisposers.set(key, registerSharedState(key, state)) + stateDisposers.set(key, registerSharedState(key, state, initialSyncId)) return state } else { From e4951237fa468d87a74f108924ec58edf1ac8e57 Mon Sep 17 00:00:00 2001 From: Sakana <15715093608@163.com> Date: Sun, 25 Oct 2026 13:33:48 +0800 Subject: [PATCH 2/2] fix(shared-state): preserve local undo during initial reads Track updates while the initial snapshot is pending. Restoring the original object after a local edit must not make a stale snapshot current again. Add an undo regression and retain initial synchronization after unchanged mutations. All required checks and the WebSocket/SSE service-discovery workflows pass. --- .../src/client/rpc-shared-state.test.ts | 29 +++++++++++++++++++ .../devframe/src/client/rpc-shared-state.ts | 8 +++-- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/packages/devframe/src/client/rpc-shared-state.test.ts b/packages/devframe/src/client/rpc-shared-state.test.ts index c56899ea..f47b9581 100644 --- a/packages/devframe/src/client/rpc-shared-state.test.ts +++ b/packages/devframe/src/client/rpc-shared-state.test.ts @@ -98,6 +98,35 @@ describe('initial shared-state snapshots', () => { expect(setCalls).toHaveLength(1) }) + it('keeps a local undo that restores the initial object while the snapshot is pending', async () => { + const { host, snapshot, setCalls } = setup() + const state = await host.get('k', { initialValue: { count: 0 } }) + const initialValue = state.value() + state.mutate((draft) => { + draft.count = 2 + }) + state.mutate(() => initialValue) + expect(state.value()).toBe(initialValue) + expect(setCalls.map(([, value]) => value)).toEqual([{ count: 2 }, { count: 0 }]) + + snapshot.resolve({ count: 1 }) + await snapshot.promise + expect(state.value()).toBe(initialValue) + expect(setCalls).toHaveLength(2) + }) + + it('applies the initial snapshot after a mutation that leaves the value unchanged', async () => { + const { host, snapshot, setCalls } = setup() + const state = await host.get('k', { initialValue: { count: 0 } }) + state.mutate((draft) => { + draft.count = 0 + }) + snapshot.resolve({ count: 1 }) + await snapshot.promise + expect(state.value()).toEqual({ count: 1 }) + expect(setCalls).toEqual([]) + }) + it('forwards a local write triggered by the initial snapshot', async () => { const { host, snapshot, setCalls } = setup() const state = await host.get('k', { initialValue: { count: 0 } }) diff --git a/packages/devframe/src/client/rpc-shared-state.ts b/packages/devframe/src/client/rpc-shared-state.ts index 8d211110..9b302929 100644 --- a/packages/devframe/src/client/rpc-shared-state.ts +++ b/packages/devframe/src/client/rpc-shared-state.ts @@ -100,17 +100,21 @@ export function createRpcSharedStateClientHost(rpc: DevframeRpcClient): RpcShare sharedState.set(key, state) for (const fn of keyAddedListeners) fn(key) - const initialState = state.value() + let stateUpdated = false + const stopTrackingUpdates = state.on('updated', () => { + stateUpdated = true + }) const initialSyncId = nanoid() rpc.call('devframe:rpc:server-state:get', key) .then((serverState) => { // A pending read must not replace a newer update or a deleted state. - if (serverState !== undefined && sharedState.get(key) === state && state.value() === initialState) + if (serverState !== undefined && sharedState.get(key) === state && !stateUpdated) state.mutate(() => mergeWithInitialValue(key, serverState), initialSyncId) }) .catch((error) => { console.error('Error getting server state', error) }) + .finally(stopTrackingUpdates) stateDisposers.set(key, registerSharedState(key, state, initialSyncId)) return state }