Close the julia oneShot worker on failed runs (fixes #649)

crates/quarto-core/tests/fixtures/extensions/julia-engine/_extensions/julia-engine/Project.toml

@@ -2,4 +2,4 @@
 QuartoNotebookRunner = "4c0109c6-14e9-4c88-93f0-2b974d3468f4"
 
 [compat]
-QuartoNotebookRunner = "=0.17.4"
+QuartoNotebookRunner = "=0.18.2"

crates/quarto-core/tests/fixtures/extensions/julia-engine/_extensions/julia-engine/_extension.yml

@@ -1,6 +1,6 @@
 title: Quarto Julia Engine Extension
 author: Quarto Julia Engine
-version: 0.1.0
+version: 0.2.1
 quarto-required: ">=1.9.0"
 contributes:
   engines:

crates/quarto-core/tests/fixtures/extensions/julia-engine/_extensions/julia-engine/julia-engine.js

@@ -621,6 +621,14 @@ function encodeBase64(data) {
   return new TextDecoder().decode(output);
 }
 
+// deno:https://jsr.io/@std/io/0.224.0/write_all.ts
+async function writeAll(writer, data) {
+  let nwritten = 0;
+  while (nwritten < data.length) {
+    nwritten += await writer.write(data.subarray(nwritten));
+  }
+}
+
 // src/worker-close.ts
 function isWorkerBusyError(e) {
   return e instanceof Error && /worker is busy/i.test(e.message);
@@ -668,6 +676,33 @@ async function postRunClose(writeCommand, file, warn) {
 ${message}`);
   }
 }
+async function errorRunClose(writeCommand, file, warn) {
+  try {
+    try {
+      await writeCommand({
+        type: "close",
+        content: {
+          file
+        }
+      });
+    } catch (e) {
+      if (isWorkerBusyError(e)) {
+        await writeCommand({
+          type: "forceclose",
+          content: {
+            file
+          }
+        });
+      } else {
+        throw e;
+      }
+    }
+  } catch (e) {
+    const message = e instanceof Error ? e.message : String(e);
+    warn(`Julia worker close after a failed run also failed; the worker may be leaked.
+${message}`);
+  }
+}
 
 // src/constants.ts
 var kJuliaEngine = "julia";
@@ -679,6 +714,7 @@ var kFigFormat = "fig-format";
 var kFigPos = "fig-pos";
 var kIpynbProduceSourceNotebook = "produce-source-notebook";
 var kKeepHidden = "keep-hidden";
+var kKeepIpynb = "keep-ipynb";
 
 // src/julia-engine.ts
 var isWindows2 = Deno.build.os === "windows";
@@ -797,6 +833,11 @@ var juliaEngineDiscovery = {
           language: "julia"
         };
         const assets = quarto.jupyter.assets(options.target.input, options.format.pandoc.to);
+        if (options.format.execute[kKeepIpynb]) {
+          const stem = options.target.source.replace(/\.[^.]+$/, "");
+          const ipynbPath = stem + ".ipynb";
+          Deno.writeTextFileSync(ipynbPath, JSON.stringify(nb, null, 2));
+        }
         const result = await quarto.jupyter.toMarkdown(nb, {
           executeOptions: options,
           language: nb.metadata.kernelspec.language.toLowerCase(),
@@ -1138,23 +1179,31 @@ async function executeJulia(options) {
     await preRunClose(writeCloseCommand, file);
   }
   const sourceRanges = buildSourceRanges(options.target.markdown);
-  const response = await writeJuliaCommand(conn, {
-    type: "run",
-    content: {
-      file,
-      options,
-      sourceRanges
+  let response;
+  try {
+    response = await writeJuliaCommand(conn, {
+      type: "run",
+      content: {
+        file,
+        options,
+        sourceRanges
+      }
+    }, transportOptions.key, options, (update) => {
+      const n = update.nChunks.toString();
+      const i = update.chunkIndex.toString();
+      const i_padded = `${" ".repeat(n.length - i.length)}${i}`;
+      const ncols = getConsoleColumns() ?? 80;
+      const firstPart = `Running [${i_padded}/${n}] at line ${update.line}:  `;
+      const firstPartLength = firstPart.length;
+      const sigLine = firstSignificantLine(update.source, Math.max(0, ncols - firstPartLength));
+      quarto.console.info(`${firstPart}${sigLine}`);
+    });
+  } catch (e) {
+    if (options.oneShot) {
+      await errorRunClose(writeCloseCommand, file, (message) => quarto.console.warning(message));
     }
-  }, transportOptions.key, options, (update) => {
-    const n = update.nChunks.toString();
-    const i = update.chunkIndex.toString();
-    const i_padded = `${" ".repeat(n.length - i.length)}${i}`;
-    const ncols = getConsoleColumns() ?? 80;
-    const firstPart = `Running [${i_padded}/${n}] at line ${update.line}:  `;
-    const firstPartLength = firstPart.length;
-    const sigLine = firstSignificantLine(update.source, Math.max(0, ncols - firstPartLength));
-    quarto.console.info(`${firstPart}${sigLine}`);
-  });
+    throw e;
+  }
   if (options.oneShot) {
     await postRunClose(writeCloseCommand, file, (message) => quarto.console.warning(message));
   }
@@ -1184,10 +1233,7 @@ async function writeJuliaCommand(conn, command, secret, options, onProgressUpdat
   }) + "\n";
   const messageBytes = new TextEncoder().encode(message);
   trace(options, `write command "${command.type}" to socket server`);
-  const bytesWritten = await conn.write(messageBytes);
-  if (bytesWritten !== messageBytes.length) {
-    throw new Error("Internal Error");
-  }
+  await writeAll(conn, messageBytes);
   let restOfPreviousResponse = new Uint8Array(512);
   let restLength = 0;
   while (true) {

crates/quarto-core/tests/fixtures/extensions/julia-engine/src/constants.ts

@@ -13,3 +13,4 @@ export const kFigFormat = "fig-format";
 export const kFigPos = "fig-pos";
 export const kIpynbProduceSourceNotebook = "produce-source-notebook";
 export const kKeepHidden = "keep-hidden";
+export const kKeepIpynb = "keep-ipynb";

crates/quarto-core/tests/fixtures/extensions/julia-engine/src/julia-engine.ts

@@ -8,6 +8,7 @@
 import { dirname, fromFileUrl, join, resolve } from "path";
 import { existsSync } from "fs/exists";
 import { encodeBase64 } from "encoding/base64";
+import { writeAll } from "jsr:@std/io@0.224.0/write-all";
 
 // Type imports from Quarto via import map
 import type {
@@ -30,6 +31,7 @@ import type {
 // unit-testable independent of the socket)
 import {
   type CloseCommandWriter,
+  errorRunClose,
   postRunClose,
   preRunClose,
 } from "./worker-close.ts";
@@ -45,6 +47,7 @@ import {
   kIpynbProduceSourceNotebook,
   kJuliaEngine,
   kKeepHidden,
+  kKeepIpynb,
 } from "./constants.ts";
 
 // Platform detection
@@ -228,6 +231,14 @@ export const juliaEngineDiscovery: ExecutionEngineDiscovery = {
           options.format.pandoc.to,
         );
 
+        // Write notebook to file if keep-ipynb is set (must happen before
+        // toMarkdown which mutates nb in place)
+        if (options.format.execute[kKeepIpynb]) {
+          const stem = options.target.source.replace(/\.[^.]+$/, "");
+          const ipynbPath = stem + ".ipynb";
+          Deno.writeTextFileSync(ipynbPath, JSON.stringify(nb, null, 2));
+        }
+
         // NOTE: for perforance reasons the 'nb' is mutated in place
         // by jupyterToMarkdown (we don't want to make a copy of a
         // potentially very large notebook) so should not be relied
@@ -722,25 +733,40 @@ async function executeJulia(
 
   const sourceRanges = buildSourceRanges(options.target.markdown);
 
-  const response = await writeJuliaCommand(
-    conn,
-    { type: "run", content: { file, options, sourceRanges } },
-    transportOptions.key,
-    options,
-    (update: ProgressUpdate) => {
-      const n = update.nChunks.toString();
-      const i = update.chunkIndex.toString();
-      const i_padded = `${" ".repeat(n.length - i.length)}${i}`;
-      const ncols = getConsoleColumns() ?? 80;
-      const firstPart = `Running [${i_padded}/${n}] at line ${update.line}:  `;
-      const firstPartLength = firstPart.length;
-      const sigLine = firstSignificantLine(
-        update.source,
-        Math.max(0, ncols - firstPartLength),
+  let response;
+  try {
+    response = await writeJuliaCommand(
+      conn,
+      { type: "run", content: { file, options, sourceRanges } },
+      transportOptions.key,
+      options,
+      (update: ProgressUpdate) => {
+        const n = update.nChunks.toString();
+        const i = update.chunkIndex.toString();
+        const i_padded = `${" ".repeat(n.length - i.length)}${i}`;
+        const ncols = getConsoleColumns() ?? 80;
+        const firstPart = `Running [${i_padded}/${n}] at line ${update.line}:  `;
+        const firstPartLength = firstPart.length;
+        const sigLine = firstSignificantLine(
+          update.source,
+          Math.max(0, ncols - firstPartLength),
+        );
+        quarto.console.info(`${firstPart}${sigLine}`);
+      },
+    );
+  } catch (e) {
+    // A failed run must still close the oneShot worker, or it leaks on the
+    // shared control server (and blocks the server's idle timeout). Best
+    // effort — the run error rethrown below is the diagnostic that matters.
+    if (options.oneShot) {
+      await errorRunClose(
+        writeCloseCommand,
+        file,
+        (message) => quarto.console.warning(message),
       );
-      quarto.console.info(`${firstPart}${sigLine}`);
-    },
-  );
+    }
+    throw e;
+  }
 
   if (options.oneShot) {
     await postRunClose(
@@ -834,10 +860,7 @@ async function writeJuliaCommand<T extends ServerCommand["type"]>(
   const messageBytes = new TextEncoder().encode(message);
 
   trace(options, `write command "${command.type}" to socket server`);
-  const bytesWritten = await conn.write(messageBytes);
-  if (bytesWritten !== messageBytes.length) {
-    throw new Error("Internal Error");
-  }
+  await writeAll(conn, messageBytes);
 
   // a string of bytes received from the server could start with a
   // partial message, contain multiple complete messages (separated by newlines) after that

crates/quarto-core/tests/fixtures/extensions/julia-engine/src/worker-close.ts

@@ -16,21 +16,22 @@ export type CloseCommandWriter = (
 ) => Promise<unknown>;
 
 // A close fails with "worker is busy" when the file's worker is still running
-// (QNR does not interrupt a running task on a plain close). See task-p0-report.md.
+// (QNR does not interrupt a running task on a plain close).
 export function isWorkerBusyError(e: unknown): boolean {
   return e instanceof Error && /worker is busy/i.test(e.message);
 }
 
 // Pre-run close (julia-engine.ts oneShot / daemon-restart path). If the file's
 // worker is busy we recover with a forceclose rather than surfacing the bare
 // protocol error: the busy worker is an ABANDONED one (a prior client vanished
-// mid-run and left the shared server's worker orphaned — the Bug A scenario),
-// and forceclose reclaims the file so this fresh render can proceed.
+// mid-run and left the shared server's worker orphaned), and forceclose
+// reclaims the file so this fresh render can proceed.
 //
-// CAVEAT (documented for the upstream PR, deliberately not special-cased here):
-// a worker busy serving a *live concurrent* render on a shared server would also
-// be force-closed, killing legitimate work. Distinguishing abandoned-vs-live is
-// the deeper oneShot-server-reuse design question — see the q2 compat log.
+// CAVEAT (deliberately documented rather than special-cased): a worker busy
+// serving a *live concurrent* render on a shared server would also be
+// force-closed, killing legitimate work. Distinguishing abandoned-vs-live
+// workers is part of the wider question of whether a oneShot render should
+// reuse a daemon-started server at all.
 export async function preRunClose(
   writeCommand: CloseCommandWriter,
   file: string,
@@ -69,3 +70,41 @@ export async function postRunClose(
     );
   }
 }
+
+// Error-path close (julia-engine.ts oneShot cleanup when the run FAILED).
+// Without this, the throw skips the post-run close and the file's worker
+// leaks on the shared control server — forever, since nothing ever runs that
+// file again (and open workers also block the server's idle timeout).
+//
+// Semantics: best-effort, never throws.
+// - Plain close first: the common failure is QNR reporting a cell error
+//   in-band, after which the worker is idle and healthy.
+// - Busy → forceclose: a transport-level failure can leave the worker still
+//   running; we are abandoning it either way, and reclaiming it now prevents
+//   creating the abandoned-busy state preRunClose has to recover from later.
+// - Any remaining failure warns instead of throwing — deliberate asymmetry
+//   with preRunClose's forceclose-propagates contract, because here a real
+//   diagnostic (the run error) is already in flight and a cleanup failure
+//   must not mask it.
+export async function errorRunClose(
+  writeCommand: CloseCommandWriter,
+  file: string,
+  warn: (message: string) => void,
+): Promise<void> {
+  try {
+    try {
+      await writeCommand({ type: "close", content: { file } });
+    } catch (e) {
+      if (isWorkerBusyError(e)) {
+        await writeCommand({ type: "forceclose", content: { file } });
+      } else {
+        throw e;
+      }
+    }
+  } catch (e) {
+    const message = e instanceof Error ? e.message : String(e);
+    warn(
+      `Julia worker close after a failed run also failed; the worker may be leaked.\n${message}`,
+    );
+  }
+}