From e55182d21f11b5086cb99c61dc9248dc02ea19eb Mon Sep 17 00:00:00 2001 From: lojhan Date: Thu, 22 Oct 2026 13:07:56 +0000 Subject: [PATCH 1/2] fix: deduplicate resource creation and preserve RPC state --- src/resource-registry.ts | 11 +- src/shared-resources.ts | 238 ++++++++++++------ test/__fixtures__/in-process/all.test.mts | 11 + .../parallel/isolated-resource.test.ts | 2 +- .../parallel/multiple-resources.test.ts | 2 +- test/__fixtures__/parallel/poku.config.mts | 3 + .../parallel/resource-cleanup.test.ts | 2 +- .../parallel/shared-state.test.ts | 2 +- test/__fixtures__/server.ts | 8 +- test/__utils__/poku-cli.ts | 1 + test/integration/shared-resources.test.ts | 48 ++-- test/unit/creation-retry.test.ts | 41 +++ test/unit/execution-mode.test.ts | 35 +++ test/unit/resource-registry.test.ts | 15 ++ test/unit/resource-results.test.ts | 83 ++++++ 15 files changed, 390 insertions(+), 112 deletions(-) create mode 100644 test/__fixtures__/in-process/all.test.mts create mode 100644 test/__fixtures__/parallel/poku.config.mts create mode 100644 test/__utils__/poku-cli.ts create mode 100644 test/unit/creation-retry.test.ts create mode 100644 test/unit/resource-results.test.ts diff --git a/src/resource-registry.ts b/src/resource-registry.ts index 4b55fa4..c6fc956 100644 --- a/src/resource-registry.ts +++ b/src/resource-registry.ts @@ -1,6 +1,6 @@ export class ResourceRegistry { private registry: Record = Object.create(null); - private isRegistering = false; + private registeringDepth = 0; register(name: string, entry: T) { this.registry[name] = entry; @@ -18,14 +18,19 @@ export class ResourceRegistry { for (const key in this.registry) if (Object.prototype.hasOwnProperty.call(this.registry, key)) delete this.registry[key]; + this.registeringDepth = 0; } getIsRegistering() { - return this.isRegistering; + return this.registeringDepth > 0; } setIsRegistering(value: boolean) { - this.isRegistering = value; + if (value) { + this.registeringDepth++; + } else { + this.registeringDepth = Math.max(0, this.registeringDepth - 1); + } } getRegistry() { diff --git a/src/shared-resources.ts b/src/shared-resources.ts index 2ab2352..dbeaee2 100644 --- a/src/shared-resources.ts +++ b/src/shared-resources.ts @@ -28,9 +28,52 @@ import { ResourceRegistry } from './resource-registry.js'; const isWindows = process.platform === 'win32'; -const resourceRegistry = new ResourceRegistry(); -const moduleCounters = new Map(); -let executionMode: SharedResourceExecutionMode = 'process'; +const RUNTIME_STATE_KEY = Symbol.for('@pokujs/shared-resources.runtime-state'); + +type SharedResourcesRuntimeState = { + resourceRegistry: ResourceRegistry; + moduleCounters: Map; + executionMode: SharedResourceExecutionMode; + pendingCreations: Map>; + pendingModuleLoads: Map>; + // biome-ignore lint/suspicious/noExplicitAny: stores heterogeneous codec types + argCodecs: ArgCodec[]; +}; + +const getDefaultCodecs = () => [ + undefinedCodec, + bigIntCodec, + dateCodec, + mapCodec, + setCodec, + arrayCodec, + objectCodec, +]; + +const getRuntimeState = (): SharedResourcesRuntimeState => { + const g = globalThis as typeof globalThis & { + [RUNTIME_STATE_KEY]?: SharedResourcesRuntimeState; + }; + if (!g[RUNTIME_STATE_KEY]) { + g[RUNTIME_STATE_KEY] = { + resourceRegistry: new ResourceRegistry(), + moduleCounters: new Map(), + executionMode: 'process', + pendingCreations: new Map>(), + pendingModuleLoads: new Map>(), + argCodecs: getDefaultCodecs(), + }; + } + return g[RUNTIME_STATE_KEY]; +}; + +const runtimeState = getRuntimeState(); +const { + resourceRegistry, + moduleCounters, + pendingCreations, + pendingModuleLoads, +} = runtimeState; export const SHARED_RESOURCE_MESSAGE_TYPES = { REQUEST_RESOURCE: 'shared_resources_requestResource', @@ -42,16 +85,17 @@ export const SHARED_RESOURCE_MESSAGE_TYPES = { export const globalRegistry = resourceRegistry.getRegistry(); export const setExecutionMode = (mode: SharedResourceExecutionMode) => { - executionMode = mode; + runtimeState.executionMode = mode; }; -export const getExecutionMode = () => executionMode; +export const getExecutionMode = () => runtimeState.executionMode; export const resetSharedResourcesRuntime = () => { resourceRegistry.clear(); - resourceRegistry.setIsRegistering(false); moduleCounters.clear(); - executionMode = 'process'; + pendingCreations.clear(); + pendingModuleLoads.clear(); + runtimeState.executionMode = 'process'; }; const create = ( @@ -89,34 +133,41 @@ const use = async ( // In-process execution always resolves resources directly from the local registry. if ( - executionMode === 'in-process' || + runtimeState.executionMode === 'in-process' || !process.send || resourceRegistry.getIsRegistering() ) { - const existing = resourceRegistry.get(name); - if (existing) { - if (executionMode === 'in-process') - return constructInProcessResource( - existing.state as Record - ) as MethodsToRPC; - - return existing.state as MethodsToRPC; - } + let entry = resourceRegistry.get(name); + if (!entry) { + let pending = pendingCreations.get(name); + if (!pending) { + pending = (async () => { + const state = await context.factory(); + const created: SharedResourceEntry = { + state, + onDestroy: context.onDestroy as + | ((instance: unknown) => void | Promise) + | undefined, + }; + resourceRegistry.register(name, created); + return created; + })(); + pendingCreations.set(name, pending); + } - const state = await context.factory(); - resourceRegistry.register(name, { - state, - onDestroy: context.onDestroy as - | ((instance: unknown) => void | Promise) - | undefined, - }); + try { + entry = await pending; + } finally { + pendingCreations.delete(name); + } + } - if (executionMode === 'in-process') + if (runtimeState.executionMode === 'in-process') return constructInProcessResource( - state as Record + entry.state as Record ) as MethodsToRPC; - return state as MethodsToRPC; + return entry.state as MethodsToRPC; } if (!context.module) @@ -156,6 +207,12 @@ export const sendIPCMessage = ( const cleanup = () => { if (timer) clearTimeout(timer); emitter.off('message', handleMessage); + if (emitter === process && emitter.listenerCount('message') === 0) { + const procWithChannel = process as typeof process & { + channel?: { unref?: () => void }; + }; + procWithChannel.channel?.unref?.(); + } }; if (typeof timeout === 'number' && timeout > 0) { @@ -165,6 +222,12 @@ export const sendIPCMessage = ( }, timeout); } + if (emitter === process) { + const procWithChannel = process as typeof process & { + channel?: { ref?: () => void }; + }; + procWithChannel.channel?.ref?.(); + } emitter.on('message', handleMessage); try { @@ -177,7 +240,7 @@ export const sendIPCMessage = ( }; const requestResource = async (name: string, module: string) => { - if (executionMode === 'in-process') { + if (runtimeState.executionMode === 'in-process') { throw new Error( 'Cannot request shared resources through IPC while running in in-process mode.' ); @@ -201,16 +264,14 @@ const requestResource = async (name: string, module: string) => { message.id === requestId, }); - if (response.error || !response.value || !response.rpcs) + if (response.error || response.value === undefined || !response.rpcs) throw new Error( response.error ?? `Invalid response for resource "${name}"` ); - return constructSharedResourceWithRPCs( - response.value as Record, - response.rpcs, - name - ); + const decodedState = decodeArg(response.value) as Record; + + return constructSharedResourceWithRPCs(decodedState, response.rpcs, name); }; const remoteProcedureCall = async ( @@ -218,7 +279,7 @@ const remoteProcedureCall = async ( method: string, args: unknown[] ) => { - if (executionMode === 'in-process') { + if (runtimeState.executionMode === 'in-process') { throw new Error( 'Cannot run shared resource RPCs through IPC while running in in-process mode.' ); @@ -257,19 +318,6 @@ const ENC_TAG = '__sr_enc'; const tagToWire = (tag: string | symbol): string => typeof tag === 'symbol' ? Symbol.keyFor(tag)! : tag; -// Checked in declaration order: first match wins, so user-registered codecs -// (prepended by configureCodecs) always take precedence over built-ins. -// biome-ignore lint/suspicious/noExplicitAny: stores heterogeneous codec types; is() always guards encode() calls -let argCodecs: ArgCodec[] = [ - undefinedCodec, - bigIntCodec, - dateCodec, - mapCodec, - setCodec, - arrayCodec, - objectCodec, -]; - /** * Registers (or merges) custom codecs into the global codec registry. * New codecs are prepended so they are checked before built-ins, allowing @@ -278,11 +326,14 @@ let argCodecs: ArgCodec[] = [ // biome-ignore lint/suspicious/noExplicitAny: see argCodecs export const configureCodecs = (codecs: ArgCodec[]): void => { const incoming = new Map(codecs.map((c) => [c.tag, c])); - argCodecs = [...codecs, ...argCodecs.filter((c) => !incoming.has(c.tag))]; + runtimeState.argCodecs = [ + ...codecs, + ...runtimeState.argCodecs.filter((c) => !incoming.has(c.tag)), + ]; }; export const encodeArg = (v: unknown): unknown => { - for (const codec of argCodecs) + for (const codec of runtimeState.argCodecs) if (codec.is(v)) return { [ENC_TAG]: 'c', @@ -304,7 +355,7 @@ export const encodeArg = (v: unknown): unknown => { const decodeEncoded = (enc: Record): unknown => { if (enc[ENC_TAG] !== 'c') return enc; - const codec = argCodecs.find((c) => + const codec = runtimeState.argCodecs.find((c) => typeof c.tag === 'symbol' ? Symbol.keyFor(c.tag) === enc.t : c.tag === enc.t ); if (!codec) @@ -320,7 +371,7 @@ export const decodeArg = (v: unknown): unknown => { }; const tryReconcileInPlace = (original: unknown, mutated: unknown): boolean => { - for (const codec of argCodecs) { + for (const codec of runtimeState.argCodecs) { if (codec.writeBack && codec.is(original) && codec.is(mutated)) { codec.writeBack(original, mutated, tryReconcileInPlace); return true; @@ -373,7 +424,7 @@ export const setupSharedResourceIPC = ( child: IPCEventEmitter | ChildProcess, registry: Record = globalRegistry ): void => { - if (executionMode === 'in-process') return; + if (runtimeState.executionMode === 'in-process') return; child.on('message', async (message: IPCMessage) => { if (message.type === SHARED_RESOURCE_MESSAGE_TYPES.REQUEST_RESOURCE) @@ -385,31 +436,45 @@ export const setupSharedResourceIPC = ( }); }; -const loadModuleResources = async (module: string) => { +const isResourceContext = ( + exported: unknown +): exported is ResourceContext => + Boolean( + exported && + typeof exported === 'object' && + 'factory' in exported && + typeof exported.factory === 'function' + ); + +const importAndRegisterModuleResources = async (module: string) => { resourceRegistry.setIsRegistering(true); try { const modulePath = isWindows ? pathToFileURL(module).href : module; const mod: Record = await import(modulePath); - for (const key in mod) { - if (!Object.prototype.hasOwnProperty.call(mod, key)) continue; - - const exported = mod[key]; - - if ( - exported && - typeof exported === 'object' && - 'factory' in exported && - typeof exported.factory === 'function' - ) - await use(exported as ResourceContext); + for (const exported of Object.values(mod)) { + if (isResourceContext(exported)) await use(exported); } } finally { resourceRegistry.setIsRegistering(false); } }; +const loadModuleResources = async (module: string) => { + let pending = pendingModuleLoads.get(module); + if (!pending) { + pending = importAndRegisterModuleResources(module); + pendingModuleLoads.set(module, pending); + } + + try { + await pending; + } finally { + pendingModuleLoads.delete(module); + } +}; + export const handleRequestResource = async ( message: IPCRequestResourceMessage, registry: Record, @@ -434,7 +499,7 @@ export const handleRequestResource = async ( child.send({ type: SHARED_RESOURCE_MESSAGE_TYPES.RESOURCE_RESULT, name: message.name, - value: entry.state, + value: encodeArg(entry.state), rpcs, id: message.id, } satisfies IPCResourceResultMessage); @@ -493,8 +558,8 @@ export const handleRemoteProcedureCall = async ( type: SHARED_RESOURCE_MESSAGE_TYPES.REMOTE_PROCEDURE_CALL_RESULT, id: message.id, value: { - result, - latest: state, + result: encodeArg(result), + latest: encodeArg(state) as Record, mutatedArgs: callArgs.map(encodeArg), }, } satisfies IPCResponse); @@ -507,6 +572,27 @@ export const handleRemoteProcedureCall = async ( } }; +const applyRpcResult = ( + target: Record, + rpcs: string[], + args: unknown[], + rpcResult: NonNullable +) => { + const decodedMutatedArgs = rpcResult.mutatedArgs.map(decodeArg); + + for (let i = 0; i < args.length; i++) + writeBack(args[i], decodedMutatedArgs[i]); + + const decodedLatest = decodeArg(rpcResult.latest) as Record; + + if (decodedLatest && typeof decodedLatest === 'object') { + for (const rpcKey of rpcs) delete decodedLatest[rpcKey]; + writeBack(target, decodedLatest); + } + + return decodeArg(rpcResult.result); +}; + const constructSharedResourceWithRPCs = ( target: Record, rpcs: string[], @@ -519,19 +605,7 @@ const constructSharedResourceWithRPCs = ( if (typeof prop === 'string' && rpcs.includes(prop)) { return async (...args: unknown[]) => { const rpcResult = await remoteProcedureCall(name, prop, args); - const decodedMutatedArgs = rpcResult.mutatedArgs.map(decodeArg); - - for (let i = 0; i < args.length; i++) - writeBack(args[i], decodedMutatedArgs[i]); - - for (const rpcKey of rpcs) { - if (rpcKey in rpcResult.latest) { - delete rpcResult.latest[rpcKey]; - } - } - - Object.assign(target, rpcResult.latest); - return rpcResult.result; + return applyRpcResult(target, rpcs, args, rpcResult); }; } diff --git a/test/__fixtures__/in-process/all.test.mts b/test/__fixtures__/in-process/all.test.mts new file mode 100644 index 0000000..e719311 --- /dev/null +++ b/test/__fixtures__/in-process/all.test.mts @@ -0,0 +1,11 @@ +import { createRequire } from 'node:module'; + +const require = createRequire(import.meta.url); +const scenarios = [ + require('../parallel/multiple-resources.test.ts'), + require('../parallel/isolated-resource.test.ts'), + require('../parallel/shared-state.test.ts'), + require('../parallel/resource-cleanup.test.ts'), +] as Array<{ result: Promise }>; + +await Promise.all(scenarios.map((scenario) => scenario.result)); diff --git a/test/__fixtures__/parallel/isolated-resource.test.ts b/test/__fixtures__/parallel/isolated-resource.test.ts index e00467e..186371f 100644 --- a/test/__fixtures__/parallel/isolated-resource.test.ts +++ b/test/__fixtures__/parallel/isolated-resource.test.ts @@ -2,7 +2,7 @@ import { assert, test, waitForExpectedResult } from 'poku'; import { resource } from '../../../src/index.js'; import { FlagContext } from '../counter.js'; -test('Test second resource only', async () => { +export const result = test('Test second resource only', async () => { const flag = await resource.use(FlagContext); await waitForExpectedResult(flag.isActive, true); diff --git a/test/__fixtures__/parallel/multiple-resources.test.ts b/test/__fixtures__/parallel/multiple-resources.test.ts index 8695a9b..f9e7fbc 100644 --- a/test/__fixtures__/parallel/multiple-resources.test.ts +++ b/test/__fixtures__/parallel/multiple-resources.test.ts @@ -2,7 +2,7 @@ import { assert, test, waitForExpectedResult } from 'poku'; import { resource } from '../../../src/index.js'; import { CounterContext, FlagContext } from '../counter.js'; -test('Test two Resources from Same File', async () => { +export const result = test('Test two Resources from Same File', async () => { const counter = await resource.use(CounterContext); const flag = await resource.use(FlagContext); diff --git a/test/__fixtures__/parallel/poku.config.mts b/test/__fixtures__/parallel/poku.config.mts new file mode 100644 index 0000000..2f50a3e --- /dev/null +++ b/test/__fixtures__/parallel/poku.config.mts @@ -0,0 +1,3 @@ +import { sharedResources } from '../../../src/index.js'; + +export default { plugins: [sharedResources()], concurrency: 0 }; diff --git a/test/__fixtures__/parallel/resource-cleanup.test.ts b/test/__fixtures__/parallel/resource-cleanup.test.ts index 7094bd0..8f703b6 100644 --- a/test/__fixtures__/parallel/resource-cleanup.test.ts +++ b/test/__fixtures__/parallel/resource-cleanup.test.ts @@ -2,7 +2,7 @@ import { assert, test } from 'poku'; import { resource } from '../../../src/index.js'; import { ServerContext } from '../server.js'; -test('Server: Query and Verify Response', async () => { +export const result = test('Server: Query and Verify Response', async () => { const server = await resource.use(ServerContext); const result = await server.query('/'); diff --git a/test/__fixtures__/parallel/shared-state.test.ts b/test/__fixtures__/parallel/shared-state.test.ts index 32623de..4fbd653 100644 --- a/test/__fixtures__/parallel/shared-state.test.ts +++ b/test/__fixtures__/parallel/shared-state.test.ts @@ -2,7 +2,7 @@ import { assert, test, waitForExpectedResult } from 'poku'; import { resource } from '../../../src/index.js'; import { CounterContext } from '../counter.js'; -test('Verify Counter State', async () => { +export const result = test('Verify Counter State', async () => { const counter = await resource.use(CounterContext); await waitForExpectedResult(counter.getCount, 1); diff --git a/test/__fixtures__/server.ts b/test/__fixtures__/server.ts index 1f26b0d..898c706 100644 --- a/test/__fixtures__/server.ts +++ b/test/__fixtures__/server.ts @@ -9,7 +9,13 @@ export const ServerContext = resource.create( res.end(JSON.stringify({ solution: 2 })); }); - await new Promise((resolve) => server.listen(0, resolve)); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, () => { + server.off('error', reject); + resolve(); + }); + }); const { port } = server.address() as AddressInfo; return { diff --git a/test/__utils__/poku-cli.ts b/test/__utils__/poku-cli.ts new file mode 100644 index 0000000..aba11f2 --- /dev/null +++ b/test/__utils__/poku-cli.ts @@ -0,0 +1 @@ +import '../../node_modules/poku/lib/bin/index.js'; diff --git a/test/integration/shared-resources.test.ts b/test/integration/shared-resources.test.ts index b8370f3..ff52f07 100644 --- a/test/integration/shared-resources.test.ts +++ b/test/integration/shared-resources.test.ts @@ -1,47 +1,51 @@ -import { assert, describe, it, poku } from 'poku'; +import { assert, describe, it } from 'poku'; import { inspectPoku } from 'poku/plugins'; -import { sharedResources } from '../../src/index.js'; + +const FACTORY_ERROR_PATTERN = /Intentional factory error/; describe('Shared Resources', async () => { await it('Parallel tests', async () => { - const code = await poku('test/__fixtures__/parallel', { - noExit: true, - plugins: [sharedResources()], - concurrency: 0, - timeout: 10000, + const { exitCode, stdout, stderr } = await inspectPoku({ + bin: 'test/__utils__/poku-cli.ts', + spawnOptions: { timeout: 15000 }, + command: + '--config=test/__fixtures__/parallel/poku.config.mts --isolation=process --timeout=10000 test/__fixtures__/parallel', }); - assert.strictEqual(code, 0, 'Exit Code needs to be 0'); + assert.strictEqual(exitCode, 0, `${stdout}\n${stderr}`); }); await it('Runs on isolation: none', async () => { - const code = await poku('test/__fixtures__/parallel', { - noExit: true, - plugins: [sharedResources()], - isolation: 'none', - concurrency: 0, - timeout: 10000, + const { exitCode, stdout, stderr } = await inspectPoku({ + bin: 'test/__utils__/poku-cli.ts', + spawnOptions: { timeout: 15000 }, + command: + '--config=test/__fixtures__/parallel/poku.config.mts --isolation=none --timeout=10000 test/__fixtures__/in-process', }); - assert.strictEqual(code, 0, 'Exit Code needs to be 0'); + assert.strictEqual(exitCode, 0, `${stdout}\n${stderr}`); }); await it('Error tests', async () => { - const { exitCode } = await inspectPoku({ + const { exitCode, stdout, stderr } = await inspectPoku({ + bin: 'test/__utils__/poku-cli.ts', + spawnOptions: { timeout: 15000 }, command: - '--config="test/__fixtures__/error/poku.config.ts" test/__fixtures__/error', + '--config=test/__fixtures__/parallel/poku.config.mts test/__fixtures__/error', }); assert.strictEqual(exitCode, 1, 'Exit Code needs to be 1'); + assert.match(`${stdout}\n${stderr}`, FACTORY_ERROR_PATTERN); }); await it('Reference tests', async () => { - const code = await poku('test/__fixtures__/references', { - noExit: true, - plugins: [sharedResources()], - concurrency: 0, + const { exitCode, stdout, stderr } = await inspectPoku({ + bin: 'test/__utils__/poku-cli.ts', + spawnOptions: { timeout: 15000 }, + command: + '--config=test/__fixtures__/parallel/poku.config.mts --isolation=process test/__fixtures__/references', }); - assert.strictEqual(code, 0, 'Exit Code needs to be 0'); + assert.strictEqual(exitCode, 0, `${stdout}\n${stderr}`); }); }); diff --git a/test/unit/creation-retry.test.ts b/test/unit/creation-retry.test.ts new file mode 100644 index 0000000..a1fad7e --- /dev/null +++ b/test/unit/creation-retry.test.ts @@ -0,0 +1,41 @@ +import nodeAssert from 'node:assert/strict'; +import { test } from 'poku'; +import { + resetSharedResourcesRuntime, + resource, + setExecutionMode, +} from '../../src/shared-resources.js'; + +test('Concurrent callers share factory failures and can retry creation', async () => { + resetSharedResourcesRuntime(); + setExecutionMode('in-process'); + let calls = 0; + const factoryError = new Error('first creation failed'); + const context = resource.create(async () => { + calls++; + await Promise.resolve(); + if (calls === 1) throw factoryError; + return { value: 42 }; + }); + try { + const first = await Promise.allSettled([ + resource.use(context), + resource.use(context), + ]); + nodeAssert.strictEqual(calls, 1); + for (const result of first) { + nodeAssert.strictEqual(result.status, 'rejected'); + if (result.status === 'rejected') + nodeAssert.strictEqual(result.reason, factoryError); + } + const next = await Promise.all([ + resource.use(context), + resource.use(context), + ]); + nodeAssert.strictEqual(calls, 2); + nodeAssert.strictEqual(next[0].value, 42); + nodeAssert.strictEqual(next[1].value, 42); + } finally { + resetSharedResourcesRuntime(); + } +}); diff --git a/test/unit/execution-mode.test.ts b/test/unit/execution-mode.test.ts index c7a0fb7..8cc72cb 100644 --- a/test/unit/execution-mode.test.ts +++ b/test/unit/execution-mode.test.ts @@ -2,6 +2,7 @@ import { assert, test } from 'poku'; import { getExecutionMode, resetSharedResourcesRuntime, + resource, setExecutionMode, setupSharedResourceIPC, } from '../../src/shared-resources.js'; @@ -53,4 +54,38 @@ test('Execution mode', async () => { 'Reset should restore process mode' ); }); + + await test('should deduplicate concurrent resource.use() calls for the same async resource', async () => { + resetSharedResourcesRuntime(); + setExecutionMode('in-process'); + + let factoryCalls = 0; + const AsyncStore = resource.create(async () => { + factoryCalls++; + await new Promise((resolve) => setTimeout(resolve, 15)); + return { + items: new Map([['initial', 1n]]), + getCalls() { + return factoryCalls; + }, + }; + }); + + const [first, second, third] = await Promise.all([ + resource.use(AsyncStore), + resource.use(AsyncStore), + resource.use(AsyncStore), + ]); + + assert.strictEqual( + factoryCalls, + 1, + 'Factory should run only once for concurrent callers' + ); + assert.strictEqual(await first.getCalls(), 1); + assert.strictEqual(second.items.get('initial'), 1n); + assert.strictEqual(third.items.get('initial'), 1n); + + resetSharedResourcesRuntime(); + }); }); diff --git a/test/unit/resource-registry.test.ts b/test/unit/resource-registry.test.ts index 351ccb0..4efea5e 100644 --- a/test/unit/resource-registry.test.ts +++ b/test/unit/resource-registry.test.ts @@ -40,6 +40,21 @@ test('ResourceRegistry', () => { false, 'State updated to false' ); + + registry.setIsRegistering(true); + registry.setIsRegistering(true); + registry.setIsRegistering(false); + assert.strictEqual( + registry.getIsRegistering(), + true, + 'Nested registration stays true until outer scope exits' + ); + registry.setIsRegistering(false); + assert.strictEqual( + registry.getIsRegistering(), + false, + 'Nested registration returns to false when depth reaches 0' + ); }); test('should return the entire registry', () => { diff --git a/test/unit/resource-results.test.ts b/test/unit/resource-results.test.ts new file mode 100644 index 0000000..eec0c6b --- /dev/null +++ b/test/unit/resource-results.test.ts @@ -0,0 +1,83 @@ +import type { + IPCRemoteProcedureCallResultMessage, + IPCResourceResultMessage, +} from '../../src/types.js'; +import nodeAssert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { test } from 'poku'; +import { + decodeArg, + encodeArg, + handleRemoteProcedureCall, + handleRequestResource, + SHARED_RESOURCE_MESSAGE_TYPES, +} from '../../src/shared-resources.js'; + +test('Resource state and RPC results survive JSON IPC transport', async () => { + const messages: unknown[] = []; + const child = Object.assign(new EventEmitter(), { + send(message: unknown) { + messages.push(JSON.parse(JSON.stringify(message))); + return true; + }, + }); + const createdAt = new Date('2026-01-01T00:00:00.000Z'); + const state = { + createdAt, + values: new Map([['initial', 1n]]), + update(input: Map) { + input.set('changed', 2n); + this.values.set('updated', 3n); + return new Set([4n]); + }, + }; + const registry = { resource: { state } }; + await handleRequestResource( + { + type: SHARED_RESOURCE_MESSAGE_TYPES.REQUEST_RESOURCE, + name: 'resource', + module: 'unused', + id: 'request', + }, + registry, + child + ); + const response = messages.shift() as IPCResourceResultMessage; + nodeAssert.strictEqual(response.error, undefined); + nodeAssert.deepStrictEqual(response.rpcs, ['update']); + const decodedState = decodeArg(response.value) as typeof state; + nodeAssert.deepStrictEqual(decodedState.createdAt, createdAt); + nodeAssert.deepStrictEqual(decodedState.values, new Map([['initial', 1n]])); + nodeAssert.strictEqual('update' in decodedState, false); + + await handleRemoteProcedureCall( + { + type: SHARED_RESOURCE_MESSAGE_TYPES.REMOTE_PROCEDURE_CALL, + name: 'resource', + method: 'update', + args: [encodeArg(new Map([['original', 0n]]))], + id: 'rpc', + }, + registry, + child + ); + const result = messages.shift() as IPCRemoteProcedureCallResultMessage; + nodeAssert.strictEqual(result.error, undefined); + nodeAssert.ok(result.value); + nodeAssert.deepStrictEqual(decodeArg(result.value.result), new Set([4n])); + const latest = decodeArg(result.value.latest) as typeof state; + nodeAssert.deepStrictEqual( + latest.values, + new Map([ + ['initial', 1n], + ['updated', 3n], + ]) + ); + nodeAssert.deepStrictEqual( + decodeArg(result.value.mutatedArgs[0]), + new Map([ + ['original', 0n], + ['changed', 2n], + ]) + ); +}); From 063b18c95a3adf7eb51faefedeb9d5e987fc451f Mon Sep 17 00:00:00 2001 From: lojhan Date: Thu, 22 Oct 2026 13:11:36 +0000 Subject: [PATCH 2/2] test: infer resource module paths in ESM fixtures --- test/__fixtures__/counter.ts | 28 ++++++++++------------------ 1 file changed, 10 insertions(+), 18 deletions(-) diff --git a/test/__fixtures__/counter.ts b/test/__fixtures__/counter.ts index 98d1354..886523b 100644 --- a/test/__fixtures__/counter.ts +++ b/test/__fixtures__/counter.ts @@ -1,8 +1,5 @@ -import { env } from 'node:process'; import { resource } from '../../src/index.js'; -const runtime = env.POKU_RUNTIME || 'node'; - export const CounterContext = resource.create(() => ({ count: 0, increment() { @@ -14,18 +11,13 @@ export const CounterContext = resource.create(() => ({ }, })); -export const FlagContext = resource.create( - () => ({ - active: false, - activate() { - this.active = true; - return this.active; - }, - isActive() { - return this.active; - }, - }), - { - module: runtime === 'deno' ? undefined : __filename, - } -); +export const FlagContext = resource.create(() => ({ + active: false, + activate() { + this.active = true; + return this.active; + }, + isActive() { + return this.active; + }, +}));