diff --git a/.quixos-subtree-source.json b/.quixos-subtree-source.json index a072eef..6ecbbcf 100644 --- a/.quixos-subtree-source.json +++ b/.quixos-subtree-source.json @@ -1,7 +1,7 @@ { "version": 1, "sourceRepo": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos.git", - "sourceCommit": "ddccb886544038ee85e4f1fd4db1c35b24869199", + "sourceCommit": "4a2b76ed15a52d06b422fb3d1f651be47d92ff51", "sourcePath": "quixos-instance/packages/camino-package-runtime", "exportName": "camino-package-runtime", "mirrorRemote": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/camino-package-runtime.git" diff --git a/src/index.ts b/src/index.ts index 9ede950..df770df 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,6 @@ import http from "node:http"; import { AsyncLocalStorage } from "node:async_hooks"; +import { randomUUID } from "node:crypto"; import { create } from "@bufbuild/protobuf"; import { createClient, type Client } from "@connectrpc/connect"; import { @@ -107,6 +108,48 @@ const logDerivedDependencies = (dependencies: DerivedDependency[]) => { ); }; +const logDerivedWatch = (event: string, detail: Record) => { + if (process.env.QUIXOS_DERIVED_WATCH_LOG !== "1") { + return; + } + process.stderr.write( + `[derived-watch] ${JSON.stringify({ event, ...detail })}\n`, + ); +}; + +type WatchableContext = { + camino: CaminoClient; + request: InvokeRequest; +}; + +type ActiveDerivedWatch = { + watchId: string; + context: Context; + get: RuntimeHandler; + controllers: Map; + running: boolean; + pending: boolean; + stopped: boolean; + lastValue: unknown; +}; + +const dependencyObjectIds = (dependencies: DerivedDependency[]) => + new Set(dependencies.map((dependency) => dependency.objectId)); + +const stopDerivedWatch = ( + watch: ActiveDerivedWatch, +) => { + if (watch.stopped) { + return; + } + watch.stopped = true; + for (const controller of watch.controllers.values()) { + controller.abort(); + } + watch.controllers.clear(); + logDerivedWatch("stopped", { watchId: watch.watchId }); +}; + export type Field = { get(): Promise; set(value: T): Promise; @@ -129,8 +172,116 @@ export const fieldOps = ( export const derivedWithDeps = (operation: { get: RuntimeHandler; -}): FieldOperation => - fieldOps({ +}): FieldOperation => { + type DerivedContext = Context & WatchableContext; + const watches = new Map>(); + + const rerun = async (watch: ActiveDerivedWatch) => { + if (watch.stopped) { + return; + } + if (watch.running) { + watch.pending = true; + return; + } + watch.running = true; + try { + do { + watch.pending = false; + const { value, dependencies } = await runWithDerivedDependencyTracking( + () => watch.get(watch.context), + ); + watch.lastValue = value; + logDerivedDependencies(dependencies); + logDerivedWatch("evaluated", { + watchId: watch.watchId, + value, + dependencies, + }); + reconcileSubscriptions(watch, dependencies); + } while (watch.pending && !watch.stopped); + } catch (error) { + logDerivedWatch("error", { + watchId: watch.watchId, + error: error instanceof Error ? error.message : String(error), + }); + } finally { + watch.running = false; + } + }; + + const scheduleRerun = ( + watch: ActiveDerivedWatch, + reason: string, + ) => { + if (watch.stopped) { + return; + } + logDerivedWatch("scheduled", { watchId: watch.watchId, reason }); + void rerun(watch); + }; + + const startObjectWatch = ( + watch: ActiveDerivedWatch, + objectId: string, + ) => { + const controller = new AbortController(); + watch.controllers.set(objectId, controller); + logDerivedWatch("subscribed", { watchId: watch.watchId, objectId }); + void (async () => { + try { + for await (const event of watch.context.camino.watchObject( + { + objectId, + includeSnapshot: false, + }, + { signal: controller.signal }, + )) { + if (controller.signal.aborted || watch.stopped) { + break; + } + scheduleRerun( + watch, + event.op?.id ? `op:${event.op.id}` : `object:${objectId}`, + ); + } + } catch (error) { + if (!controller.signal.aborted && !watch.stopped) { + logDerivedWatch("watch-error", { + watchId: watch.watchId, + objectId, + error: error instanceof Error ? error.message : String(error), + }); + } + } finally { + if (watch.controllers.get(objectId) === controller) { + watch.controllers.delete(objectId); + logDerivedWatch("unsubscribed", { watchId: watch.watchId, objectId }); + } + } + })(); + }; + + const reconcileSubscriptions = ( + watch: ActiveDerivedWatch, + dependencies: DerivedDependency[], + ) => { + const objectIds = dependencyObjectIds(dependencies); + for (const [objectId, controller] of watch.controllers.entries()) { + if (!objectIds.has(objectId)) { + controller.abort(); + watch.controllers.delete(objectId); + logDerivedWatch("unsubscribed", { watchId: watch.watchId, objectId }); + } + } + for (const objectId of objectIds) { + if (!watch.controllers.has(objectId)) { + startObjectWatch(watch, objectId); + } + } + }; + + return fieldOps({ async get(context) { const { value, dependencies } = await runWithDerivedDependencyTracking( () => operation.get(context), @@ -138,7 +289,43 @@ export const derivedWithDeps = (operation: { logDerivedDependencies(dependencies); return value; }, + async watchStart(context) { + const watchId = `watch:${randomUUID()}`; + const watch: ActiveDerivedWatch = { + watchId, + context: context as DerivedContext, + get: operation.get as RuntimeHandler, + controllers: new Map(), + running: false, + pending: false, + stopped: false, + lastValue: undefined, + }; + watches.set(watchId, watch); + logDerivedWatch("started", { watchId }); + await rerun(watch); + return { + watch_id: watchId, + }; + }, + watchStop(context) { + const input = protoFieldsToJs((context as DerivedContext).request.input); + const watchId = + typeof input.watch_id === "string" + ? input.watch_id + : typeof input.watchId === "string" + ? input.watchId + : ""; + const watch = watches.get(watchId); + if (!watch) { + throw new Error(`Unknown derived watch: ${watchId}`); + } + stopDerivedWatch(watch); + watches.delete(watchId); + return {}; + }, }); +}; export type RuntimeFunctionSpec = { exportName: ExportName;