fix: shared-state trust-flip re-init, atomic storage writes, falsy RPC cache (#80)

Author: antfubot

packages/devframe/src/client/rpc-shared-state.test.ts

@@ -0,0 +1,36 @@
+import { createEventEmitter } from 'devframe/utils/events'
+import { describe, expect, it } from 'vitest'
+import { createRpcSharedStateClientHost } from './rpc-shared-state'
+
+function makeFakeRpc() {
+  const events = createEventEmitter<any>()
+  const setCalls: any[][] = []
+  const rpc = {
+    connectionMeta: { backend: 'websocket' },
+    isTrusted: false,
+    events,
+    client: { register: () => {} },
+    callEvent: (name: string, ...args: any[]) => {
+      if (name === 'devframe:rpc:server-state:set')
+        setCalls.push(args)
+    },
+    call: async () => undefined,
+  } as any
+  return { rpc, events, setCalls }
+}
+
+describe('client shared state', () => {
+  it('registers the server-sync bridge once across repeated trust flips', async () => {
+    const { rpc, events, setCalls } = makeFakeRpc()
+    const host = createRpcSharedStateClientHost(rpc)
+    const state = await host.get('k', { initialValue: { a: 1 } })
+
+    events.emit('rpc:is-trusted:updated', true)
+    events.emit('rpc:is-trusted:updated', true) // second flip must not re-register
+
+    state.mutate((d: any) => {
+      d.a = 2
+    })
+    expect(setCalls).toHaveLength(1) // exactly one server-state:set, not two
+  })
+})

packages/devframe/src/client/rpc-shared-state.ts

@@ -114,8 +114,10 @@ export function createRpcSharedStateClientHost(rpc: DevframeRpcClient): RpcShare
       return new Promise<SharedState<T>>((resolve) => {
         if (!rpc.isTrusted) {
           resolve(state)
+          let initialized = false
           rpc.events.on('rpc:is-trusted:updated', (isTrusted) => {
-            if (isTrusted) {
+            if (isTrusted && !initialized) {
+              initialized = true
               initSharedState()
             }
           })

packages/devframe/src/client/rpc.ts

@@ -316,14 +316,11 @@ export async function getDevframeRpcClient(
           async onRequest(req, next, resolve) {
             await rpcOptions.onRequest?.call(this, req, next, resolve)
             if (cacheOptions && cacheManager?.validate(req.m)) {
-              const cached = cacheManager.cached(req.m, req.a)
-              if (cached) {
-                return resolve(cached)
-              }
-              else {
-                const res = await next(req)
-                cacheManager?.apply(req, res)
+              if (cacheManager.has(req.m, req.a)) {
+                return resolve(cacheManager.cached(req.m, req.a))
               }
+              const res = await next(req)
+              cacheManager.apply(req, res)
             }
             else {
               await next(req)

packages/devframe/src/node/__tests__/storage.test.ts

@@ -40,4 +40,67 @@ describe('createStorage', () => {
       fs.rmSync(dir, { recursive: true, force: true })
     }
   })
+
+  it('writes atomically via temp file + rename', async () => {
+    const dir = fs.mkdtempSync(join(os.tmpdir(), 'devframe-storage-'))
+    const filepath = join(dir, 'state.json')
+
+    try {
+      const state = createStorage({
+        filepath,
+        initialValue: { count: 1 },
+        debounce: 0,
+      })
+
+      state.mutate((draft) => {
+        draft.count = 2
+      })
+
+      await wait(20)
+
+      // no leftover temp file
+      const entries = fs.readdirSync(dir)
+      expect(entries).toEqual(['state.json'])
+
+      const saved = JSON.parse(fs.readFileSync(filepath, 'utf-8'))
+      expect(saved).toEqual({ count: 2 })
+    }
+    finally {
+      fs.rmSync(dir, { recursive: true, force: true })
+    }
+  })
+
+  it('reports a write failure instead of throwing', async () => {
+    const dir = fs.mkdtempSync(join(os.tmpdir(), 'devframe-storage-'))
+    const filepath = join(dir, 'state.json')
+
+    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {})
+    const renameSpy = vi.spyOn(fs, 'renameSync').mockImplementation(() => {
+      throw new Error('EACCES: permission denied')
+    })
+
+    try {
+      const state = createStorage({
+        filepath,
+        initialValue: { count: 1 },
+        debounce: 0,
+      })
+
+      expect(() => {
+        state.mutate((draft) => {
+          draft.count = 2
+        })
+      }).not.toThrow()
+
+      await wait(20)
+
+      expect(errorSpy).toHaveBeenCalled()
+      expect(fs.existsSync(filepath)).toBe(false)
+    }
+    finally {
+      renameSpy.mockRestore()
+      errorSpy.mockRestore()
+      fs.rmSync(dir, { recursive: true, force: true })
+    }
+  })
 })

packages/devframe/src/node/diagnostics.ts

@@ -64,5 +64,9 @@ export const diagnostics = defineDiagnostics({
         `Scoped RPC registration for namespace "${p.namespace}" received an already-namespaced function name "${p.name}".`,
       fix: 'A scoped context auto-namespaces ids. Pass a bare name without a ":" separator (e.g. `register({ name: "get-cwd" })`), or use the unscoped `ctx.base.rpc.register` for a fully-qualified name.',
     },
+    DF0035: {
+      why: (p: { filepath: string }) => `Failed to persist storage file: ${p.filepath}`,
+      fix: 'Check that the storage directory is writable and has free space.',
+    },
   },
 })

packages/devframe/src/node/storage.ts

@@ -1,4 +1,5 @@
 import fs from 'node:fs'
+import process from 'node:process'
 import { destr } from 'destr'
 import { createSharedState } from 'devframe/utils/shared-state'
 import { dirname } from 'pathe'
@@ -39,8 +40,16 @@ export function createStorage<T extends object>(options: CreateStorageOptions<T>
   state.on(
     'updated',
     debounce((newState) => {
-      fs.mkdirSync(dirname(options.filepath), { recursive: true })
-      fs.writeFileSync(options.filepath, `${JSON.stringify(newState, null, 2)}\n`)
+      try {
+        const dir = dirname(options.filepath)
+        fs.mkdirSync(dir, { recursive: true })
+        const tmp = `${options.filepath}.${process.pid}.tmp`
+        fs.writeFileSync(tmp, `${JSON.stringify(newState, null, 2)}\n`)
+        fs.renameSync(tmp, options.filepath) // atomic replace on same filesystem
+      }
+      catch (error) {
+        diagnostics.DF0035({ filepath: options.filepath, cause: error }, { method: 'error' })
+      }
     }, debounceTime),
   )
 

packages/devframe/src/rpc/cache.test.ts

@@ -19,3 +19,46 @@ it('cache', async () => {
   cache.clear()
   expect(cache.cached<number>('fn2', [3, 4])).toBeUndefined()
 })
+
+it('serves falsy cached values via `has` (presence, not truthiness)', () => {
+  const cache = new RpcCacheManager({ functions: ['fn'] })
+
+  // absent key: `has` reports false, distinct from a stored falsy value
+  expect(cache.has('fn', ['zero'])).toBe(false)
+
+  cache.apply({ m: 'fn', a: ['zero'] }, 0)
+  cache.apply({ m: 'fn', a: ['false'] }, false)
+  cache.apply({ m: 'fn', a: ['empty'] }, '')
+  cache.apply({ m: 'fn', a: ['null'] }, null)
+
+  expect(cache.has('fn', ['zero'])).toBe(true)
+  expect(cache.has('fn', ['false'])).toBe(true)
+  expect(cache.has('fn', ['empty'])).toBe(true)
+  expect(cache.has('fn', ['null'])).toBe(true)
+
+  expect(cache.cached('fn', ['zero'])).toBe(0)
+  expect(cache.cached('fn', ['false'])).toBe(false)
+  expect(cache.cached('fn', ['empty'])).toBe('')
+  expect(cache.cached('fn', ['null'])).toBe(null)
+})
+
+it('the client cache path (has-then-cached) serves a falsy value without a second fetch', async () => {
+  const cache = new RpcCacheManager({ functions: ['fn'] })
+  let nextCalls = 0
+  const next = async (req: { m: string, a: unknown[] }) => {
+    nextCalls++
+    cache.apply(req, 0)
+    return 0
+  }
+
+  // mirrors the onRequest cache path in client/rpc.ts
+  async function callThroughCache(req: { m: string, a: unknown[] }) {
+    if (cache.has(req.m, req.a))
+      return cache.cached(req.m, req.a)
+    return next(req)
+  }
+
+  expect(await callThroughCache({ m: 'fn', a: [] })).toBe(0)
+  expect(await callThroughCache({ m: 'fn', a: [] })).toBe(0)
+  expect(nextCalls).toBe(1) // second call served from cache, not re-fetched
+})

packages/devframe/src/rpc/cache.ts

@@ -33,6 +33,10 @@ export class RpcCacheManager {
     return undefined
   }
 
+  has(m: string, a: unknown[]): boolean {
+    return this.cacheMap.get(m)?.has(this.keySerializer(a)) ?? false
+  }
+
   apply(req: { m: string, a: unknown[] }, res: unknown): void {
     const methodCache = this.cacheMap.get(req.m) || new Map<string, unknown>()
     methodCache.set(this.keySerializer(req.a), res)