danvics/zerobyte
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: manual promise state to effect Deferred

Browse source
This commit is contained in:
Nicolas Meienberger 2026-05-06 22:16:43 +02:00 committed by Nico
parent deedf2a203
commit 8cabaef17c
1 changed files with 54 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

107
app/server/modules/agents/controller/session.ts
View file

@ -1,4 +1,4 @@
import { Effect, Queue, Ref, type Scope } from "effect"; import { Deferred, Effect, Queue, Ref, type Scope } from "effect";
import type { AgentKind } from "../../../db/schema"; import type { AgentKind } from "../../../db/schema";
import { import {
createControllerMessage, createControllerMessage,
@ -29,10 +29,8 @@ type SessionState = {
lastPongAt: number | null; lastPongAt: number | null;
}; };
type PendingVolumeCommand = { type InFlightVolumeCommand = {
resolve: (payload: VolumeCommandResponsePayload) => void; deferred: Deferred.Deferred<VolumeCommandResponsePayload, Error>;
reject: (error: Error) => void;
timeout: ReturnType<typeof setTimeout>;
}; };
export type ControllerAgentSessionEvent = export type ControllerAgentSessionEvent =
@ -58,7 +56,7 @@ export const createControllerAgentSession = (
Effect.gen(function* () { Effect.gen(function* () {
let isClosed = false; let isClosed = false;
const outboundQueue = yield* Queue.bounded<ControllerWireMessage>(64); const outboundQueue = yield* Queue.bounded<ControllerWireMessage>(64);
const pendingVolumeCommands = new Map<string, PendingVolumeCommand>(); const inFlightVolumeCommands = yield* Ref.make(new Map<string, InFlightVolumeCommand>());
const state = yield* Ref.make<SessionState>({ const state = yield* Ref.make<SessionState>({
isReady: false, isReady: false,
lastSeenAt: null, lastSeenAt: null,
@ -77,18 +75,33 @@ export const createControllerAgentSession = (
const updateState = (update: (current: SessionState) => SessionState) => Ref.update(state, update); const updateState = (update: (current: SessionState) => SessionState) => Ref.update(state, update);
const rejectPendingVolumeCommands = () => { const setInFlightVolumeCommand = (commandId: string, inFlight: InFlightVolumeCommand) =>
for (const [commandId, pending] of pendingVolumeCommands) { Ref.update(inFlightVolumeCommands, (current) => new Map(current).set(commandId, inFlight));
clearTimeout(pending.timeout);
pending.reject(new Error(`Agent session closed before volume command ${commandId} completed`)); const removeInFlightVolumeCommand = (commandId: string) =>
Ref.modify(inFlightVolumeCommands, (current) => {
const inFlight = current.get(commandId) ?? null;
const next = new Map(current);
next.delete(commandId);
return [inFlight, next];
});
const rejectInFlightVolumeCommands = Effect.gen(function* () {
const inFlightCommands = yield* Ref.get(inFlightVolumeCommands);
yield* Ref.set(inFlightVolumeCommands, new Map());
for (const [commandId, inFlight] of inFlightCommands) {
yield* Deferred.fail(
inFlight.deferred,
new Error(`Agent session closed before volume command ${commandId} completed`),
);
} }
pendingVolumeCommands.clear(); });
};
const releaseSession = Effect.gen(function* () { const releaseSession = Effect.gen(function* () {
const disconnectedAt = Date.now(); const disconnectedAt = Date.now();
yield* updateState((current) => ({ ...current, isReady: false, lastSeenAt: disconnectedAt })); yield* updateState((current) => ({ ...current, isReady: false, lastSeenAt: disconnectedAt }));
yield* Effect.sync(rejectPendingVolumeCommands); yield* rejectInFlightVolumeCommands;
yield* onEvent({ type: "agent.disconnected" }); yield* onEvent({ type: "agent.disconnected" });
yield* Queue.shutdown(outboundQueue); yield* Queue.shutdown(outboundQueue);
@ -152,17 +165,16 @@ export const createControllerAgentSession = (
return yield* Effect.never; return yield* Effect.never;
}); });
const handleVolumeCommandResult = (payload: VolumeCommandResponsePayload) => { const handleVolumeCommandResult = (payload: VolumeCommandResponsePayload) =>
const pending = pendingVolumeCommands.get(payload.commandId); Effect.gen(function* () {
if (!pending) { const inFlight = yield* removeInFlightVolumeCommand(payload.commandId);
logger.warn(`Received response for unknown volume command ${payload.commandId}`); if (!inFlight) {
return; yield* logger.effect.warn(`Received response for unknown volume command ${payload.commandId}`);
} return;
}
pendingVolumeCommands.delete(payload.commandId); yield* Deferred.succeed(inFlight.deferred, payload);
clearTimeout(pending.timeout); });
pending.resolve(payload);
};
const handleAgentMessage = (message: AgentMessage) => const handleAgentMessage = (message: AgentMessage) =>
Effect.gen(function* () { Effect.gen(function* () {
@ -181,7 +193,7 @@ export const createControllerAgentSession = (
break; break;
} }
case "volume.commandResult": { case "volume.commandResult": {
yield* Effect.sync(() => handleVolumeCommandResult(message.payload)); yield* handleVolumeCommandResult(message.payload);
break; break;
} }
default: { default: {
@ -212,37 +224,26 @@ export const createControllerAgentSession = (
}, },
sendBackup: (payload) => offerOutbound(createControllerMessage("backup.run", payload)), sendBackup: (payload) => offerOutbound(createControllerMessage("backup.run", payload)),
sendBackupCancel: (payload) => offerOutbound(createControllerMessage("backup.cancel", payload)), sendBackupCancel: (payload) => offerOutbound(createControllerMessage("backup.cancel", payload)),
runVolumeCommand: (command) => { runVolumeCommand: (command) =>
return Effect.tryPromise({ Effect.gen(function* () {
try: async () => { const commandId = Bun.randomUUIDv7();
const commandId = Bun.randomUUIDv7(); const deferred = yield* Deferred.make<VolumeCommandResponsePayload, Error>();
const response = new Promise<VolumeCommandResponsePayload>((resolve, reject) => { yield* setInFlightVolumeCommand(commandId, { deferred });
const timeout = setTimeout(() => {
pendingVolumeCommands.delete(commandId);
reject(new Error(`Volume command ${command.name} timed out`));
}, 60_000);
pendingVolumeCommands.set(commandId, { resolve, reject, timeout }); const queued = yield* offerOutbound(createControllerMessage("volume.command", { commandId, command }));
}); if (!queued) {
yield* removeInFlightVolumeCommand(commandId);
return yield* Effect.fail(new Error(`Failed to queue volume command ${command.name}`));
}
const queued = await Effect.runPromise( return yield* Deferred.await(deferred).pipe(
offerOutbound(createControllerMessage("volume.command", { commandId, command })), Effect.timeoutFail({
); duration: "60 seconds",
onTimeout: () => new Error(`Volume command ${command.name} timed out`),
if (!queued) { }),
const pending = pendingVolumeCommands.get(commandId); Effect.ensuring(removeInFlightVolumeCommand(commandId)),
if (pending) { );
clearTimeout(pending.timeout); }),
pendingVolumeCommands.delete(commandId);
}
throw new Error(`Failed to queue volume command ${command.name}`);
}
return response;
},
catch: (error) => (error instanceof Error ? error : new Error(toMessage(error))),
});
},
isReady: () => Ref.get(state).pipe(Effect.map((current) => current.isReady)), isReady: () => Ref.get(state).pipe(Effect.map((current) => current.isReady)),
run, run,
}; };