refactor: correctly close session on send failure

apps/agent/package.json

@@ -3,6 +3,9 @@
 	"private": true,
 	"type": "module",
 	"module": "index.ts",
+	"scripts": {
+		"tsc": "tsc --noEmit"
+	},
 	"dependencies": {
 		"@zerobyte/contracts": "workspace:*",
 		"@zerobyte/core": "workspace:*",

apps/agent/src/__tests__/controller-session.test.ts

@@ -69,3 +69,25 @@ test("emits backup.failed when a backup command hits a restic error", async () =
 		session.close();
 	}
 });
+
+test("closes the websocket when an outbound send throws", async () => {
+	const close = mock(() => undefined);
+	const session = createControllerSession(
+		fromPartial({
+			send: () => {
+				throw new Error("socket write failed");
+			},
+			close,
+		}),
+	);
+
+	try {
+		session.onOpen();
+
+		await waitForExpect(() => {
+			expect(close).toHaveBeenCalledTimes(1);
+		});
+	} finally {
+		session.close();
+	}
+});

apps/agent/src/controller-session.ts

@@ -17,6 +17,7 @@ export type ControllerSession = {
 };
 
 export const createControllerSession = (ws: WebSocket): ControllerSession => {
+	let isClosed = false;
 	const outboundQueue = Effect.runSync(Queue.bounded<AgentWireMessage>(64));
 	const inboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
 	const runningJobsRef = Effect.runSync(Ref.make<Map<string, RunningJob>>(new Map()));
@@ -63,6 +64,31 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => {
 		offerOutbound,
 	};
 
+	const closeSession = () => {
+		if (isClosed) {
+			return;
+		}
+
+		isClosed = true;
+		void Effect.runPromise(abortRunningJobs).catch(() => {});
+		void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
+		void Effect.runPromise(Fiber.interrupt(processorFiber)).catch(() => {});
+		void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});
+		void Effect.runPromise(Queue.shutdown(inboundQueue)).catch(() => {});
+	};
+
+	const handleSendFailure = (reason: string) => {
+		logger.error(`Closing agent session after an outbound websocket send failed: ${reason}`);
+
+		try {
+			ws.close();
+		} catch (error) {
+			logger.error(`Failed to close controller websocket after send failure: ${toMessage(error)}`);
+		}
+
+		closeSession();
+	};
+
 	const writerFiber = Effect.runFork(
 		Effect.forever(
 			Effect.gen(function* () {
@@ -71,7 +97,7 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => {
 					try {
 						ws.send(message);
 					} catch (error) {
-						logger.error(`Failed to send controller message: ${toMessage(error)}`);
+						handleSendFailure(toMessage(error));
 					}
 				});
 			}),
@@ -115,12 +141,6 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => {
 				logger.error(`Failed to queue inbound message: ${toMessage(error)}`);
 			});
 		},
-		close: () => {
+		close: closeSession,
-			void Effect.runPromise(abortRunningJobs).catch(() => {});
-			void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
-			void Effect.runPromise(Fiber.interrupt(processorFiber)).catch(() => {});
-			void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});
-			void Effect.runPromise(Queue.shutdown(inboundQueue)).catch(() => {});
-		},
 	};
 };