feat(node): report executor progress, logs and toggles over the side-channel

.github/actions/provision-gradle/action.yml

@@ -0,0 +1,24 @@
+name: Provision Gradle
+description: >-
+  Downloads the Gradle distribution the wrapper pins, retrying transient network
+  failures. The wrapper fetches it on first use and gives up on the first error,
+  so a reset connection to services.gradle.org fails the build before any code
+  has run. Doing it in its own step keeps that retry off the real invocation,
+  which would otherwise re-run the build to recover a download.
+
+runs:
+  using: composite
+  steps:
+    - name: Download the pinned Gradle distribution
+      working-directory: sdks/java
+      shell: bash
+      run: |
+        for attempt in 1 2 3; do
+          if ./gradlew --version --no-daemon; then
+            exit 0
+          fi
+          echo "::warning::Gradle provisioning attempt ${attempt} failed; retrying"
+          sleep $((attempt * 10))
+        done
+        echo "::error::could not download the Gradle distribution after 3 attempts"
+        exit 1

.github/workflows/ci-java.yml

@@ -55,6 +55,9 @@ jobs:
         with:
           java-version: ${{ matrix.java }}
 
+      - name: Provision Gradle
+        uses: ./.github/actions/provision-gradle
+
       # Stage the prebuilt .so where copyNative expects it, so no Rust is needed.
       - name: Download the cdylib
         uses: actions/download-artifact@v8
@@ -100,6 +103,9 @@ jobs:
       - name: Set up Java + Gradle
         uses: ./.github/actions/setup-java
 
+      - name: Provision Gradle
+        uses: ./.github/actions/provision-gradle
+
       # bash so `./gradlew` runs uniformly on the Windows runner too.
       - name: Build and test the Java SDK
         working-directory: sdks/java
@@ -128,6 +134,9 @@ jobs:
           distribution: graalvm
           github-token: ${{ secrets.GITHUB_TOKEN }}
 
+      - name: Provision Gradle
+        uses: ./.github/actions/provision-gradle
+
       - name: Build and run the native image from shipped metadata
         working-directory: sdks/java
         run: |

crates/taskito-node/src/executor.rs

@@ -17,7 +17,7 @@ use napi::threadsafe_function::ThreadsafeFunction;
 use napi_derive::napi;
 use taskito_core::worker::{
     AttachAddress, CancelSignals, ExecutorClient, ExecutorConfig, ExecutorError, ExecutorHandle,
-    ExecutorSession, WorkerDispatcher,
+    ExecutorSession, ExecutorSideChannel, WorkerDispatcher,
 };
 
 use crate::convert::{JsTaskInvocation, JsTaskOutcome};
@@ -60,6 +60,9 @@ pub struct JsExecutor {
     /// Cancels delivered as protocol frames. The JS side polls this instead of
     /// a storage flag, which a detached executor does not have.
     cancels: Arc<CancelSignals>,
+    /// Progress, task logs and toggles — the storage-shaped operations the
+    /// scheduler performs on this executor's behalf.
+    side_channel: ExecutorSideChannel,
     scheduler_id: String,
     executor_id: String,
     peer: String,
@@ -100,6 +103,52 @@ impl JsExecutor {
         self.cancels.is_cancelled(&job_id)
     }
 
+    /// Whether the scheduler applies progress and task logs on our behalf.
+    ///
+    /// False against a scheduler with no storage configured for it, or one
+    /// built before the side-channel existed. The methods below are no-ops
+    /// either way; this exists so the shell can say so once rather than
+    /// silently dropping a task's progress bar.
+    #[napi]
+    pub fn supports_side_channel(&self) -> bool {
+        self.side_channel.is_supported()
+    }
+
+    /// Report a running job's progress (0-100).
+    ///
+    /// An executor has no storage of its own, so this travels to the scheduler
+    /// instead. Fire-and-forget: it never blocks the calling task and never
+    /// fails its job.
+    #[napi]
+    pub fn report_progress(&self, job_id: String, progress: i32) {
+        self.side_channel.report_progress(&job_id, progress);
+    }
+
+    /// Write one structured log line for a running job. A published partial is
+    /// this at level `result`, with the value as `extra`.
+    #[napi]
+    pub fn write_task_log(
+        &self,
+        job_id: String,
+        task_name: String,
+        level: String,
+        message: String,
+        extra: Option<String>,
+    ) {
+        self.side_channel
+            .write_task_log(&job_id, &task_name, &level, &message, extra.as_deref());
+    }
+
+    /// Middleware the operator has disabled for a running job's task.
+    ///
+    /// Resolved by the scheduler at dispatch and carried on the job frame, so
+    /// a dashboard toggle is honoured without the settings read this process
+    /// has no storage to perform.
+    #[napi]
+    pub fn disabled_middleware(&self, job_id: String) -> Vec<String> {
+        self.side_channel.disabled_middleware(&job_id)
+    }
+
     /// Resolve once the scheduler ends the session — a `shutdown` frame, or the
     /// connection dropping. Does not drain; call `shutdown()` for that.
     #[napi]
@@ -217,6 +266,7 @@ pub async fn start_executor(
         Ok(JsExecutor {
             executor_id: handle.executor_id().to_string(),
             session: handle.session(),
+            side_channel: handle.side_channel(),
             cancels,
             handle: Arc::new(Mutex::new(Some(handle))),
             scheduler_id,

sdks/node/src/executor.ts

@@ -43,7 +43,14 @@ export interface ExecutorStartParams {
   tasks: ReadonlyMap<string, RegisteredTask>;
   serializer: Serializer;
   codecs?: ReadonlyMap<string, PayloadCodec>;
-  middlewareFor: (taskName: string) => readonly Middleware[];
+  /**
+   * The middleware chain for a task, minus `disabled`.
+   *
+   * Takes the disable list rather than reading one: an executor has no
+   * settings store, so the scheduler resolves it and attaches it to each
+   * dispatch.
+   */
+  middlewareFor: (taskName: string, disabled: readonly string[]) => readonly Middleware[];
   emitter: Emitter;
   resources: ResourceRuntime;
   run?: ExecutorRunOptions;
@@ -87,20 +94,41 @@ export class Executor {
     // The executor does not exist yet, and the callback it needs must already
     // be able to reach it — a cancel frame lands in native state that a running
     // handler polls. Resolved through this holder, assigned once the attach
-    // succeeds; until then nothing is running, so nothing can be cancelled.
+    // succeeds.
+    //
+    // The native attach starts its job loop before this promise resolves, so
+    // the scheduler can dispatch into that window. An invocation waits for the
+    // holder to be filled rather than reading it empty, which would run a
+    // middleware the dispatch said was disabled and drop the job's progress.
     let attached: NativeExecutor | undefined;
+    let markAttached: () => void = () => {};
+    const attachedReady = new Promise<void>((resolve) => {
+      markAttached = resolve;
+    });
 
-    const taskCallback = createTaskCallback({
+    const invoke = createTaskCallback({
       tasks,
       serializer,
       codecs,
-      middlewareFor,
+      // Every one of these reaches for the executor rather than for storage,
+      // which this process deliberately has none of: the scheduler holds the
+      // connection and does the work on its behalf.
+      middlewareFor: (taskName, jobId) =>
+        middlewareFor(taskName, attached?.disabledMiddleware(jobId) ?? []),
       emitter,
       resources,
       queue,
       isCancelled: (jobId) => attached?.isCancelRequested(jobId) ?? false,
+      setProgress: (jobId, progress) => attached?.reportProgress(jobId, progress),
+      writeTaskLog: (jobId, taskName, level, message, extra) =>
+        attached?.writeTaskLog(jobId, taskName, level, message, extra),
     });
 
+    const taskCallback: typeof invoke = async (invocation) => {
+      await attachedReady;
+      return invoke(invocation);
+    };
+
     const native = await startNativeExecutor(taskCallback, {
       address,
       tasks: advertised,
@@ -115,6 +143,7 @@ export class Executor {
     });
 
     attached = native;
+    markAttached();
     try {
       // Only lease the resource runtime once the attach actually succeeded, so a
       // refused handshake leaks nothing.

sdks/node/src/queue.ts

@@ -1658,18 +1658,17 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
    * without polling storage itself. Hold the returned {@link Executor}.
    */
   async runExecutor(options?: ExecutorRunOptions): Promise<Executor> {
-    const disables = new MiddlewareDisableStore(this.native);
     const executor: Executor = await Executor.start(this.native, {
       onStopped: () => this.liveExecutors.delete(executor),
       tasks: this.tasks,
       serializer: this.serializer,
       codecs: this.codecs,
-      middlewareFor: (taskName) => {
-        const disabled = disables.getFor(taskName);
-        return disabled.length === 0
+      // The list is supplied rather than read: an executor opens no storage,
+      // so the scheduler resolves the toggles and sends them with the job.
+      middlewareFor: (_taskName, disabled) =>
+        disabled.length === 0
           ? this.middleware
-          : this.middleware.filter((mw, index) => !disabled.includes(middlewareKey(mw, index)));
-      },
+          : this.middleware.filter((mw, index) => !disabled.includes(middlewareKey(mw, index))),
       emitter: this.emitter,
       resources: this.resources,
       run: options,

sdks/node/src/task-callback.ts

@@ -21,8 +21,15 @@ export interface TaskCallbackDeps {
   serializer: Serializer;
   /** Named codec registry for per-task payload decode (see `TaskOptions.codecs`). */
   codecs?: ReadonlyMap<string, PayloadCodec>;
-  /** The middleware chain for a task, after dashboard disables are applied. */
-  middlewareFor: (taskName: string) => readonly Middleware[];
+  /**
+   * The middleware chain for a task, after dashboard disables are applied.
+   *
+   * Takes the job id because an attached executor resolves disables per
+   * dispatch — the scheduler attaches the list to the job frame, since the
+   * executor has no settings store to read it from. A worker ignores it and
+   * reads storage by task name.
+   */
+  middlewareFor: (taskName: string, jobId: string) => readonly Middleware[];
   emitter: Emitter;
   resources: ResourceRuntime;
   /** Backs progress, published partials, and the cancel-flag poll. */
@@ -35,6 +42,21 @@ export interface TaskCallbackDeps {
    * the native state those land in.
    */
   isCancelled?: (jobId: string) => boolean;
+  /**
+   * Overrides where a task's progress goes.
+   *
+   * Same reason as {@link isCancelled}: an executor has no storage, so it
+   * sends progress to the scheduler, which applies it.
+   */
+  setProgress?: (jobId: string, progress: number) => void;
+  /** Overrides where a task's log lines and published partials go. */
+  writeTaskLog?: (
+    jobId: string,
+    taskName: string,
+    level: string,
+    message: string,
+    extra?: string,
+  ) => void;
 }
 
 /**
@@ -49,6 +71,13 @@ export function createTaskCallback(
 ): (invocation: JsTaskInvocation) => Promise<JsTaskOutcome> {
   const { tasks, serializer, codecs, middlewareFor, emitter, resources, queue } = deps;
   const isCancelled = deps.isCancelled ?? ((jobId: string) => queue.isCancelRequested(jobId));
+  const setProgress =
+    deps.setProgress ??
+    ((jobId: string, progress: number) => queue.updateProgress(jobId, progress));
+  const writeTaskLog =
+    deps.writeTaskLog ??
+    ((jobId: string, taskName: string, level: string, message: string, extra?: string) =>
+      queue.writeTaskLog(jobId, taskName, level, message, extra));
 
   return async (invocation: JsTaskInvocation): Promise<JsTaskOutcome> => {
     // Built-in workflow cache-return: echo the single (cached) arg as the result.
@@ -74,16 +103,16 @@ export function createTaskCallback(
     // Resolve the middleware chain BEFORE allocating the cancel poller and
     // task scope — it reads storage and may throw, and nothing would clean
     // those up yet.
-    const chain = middlewareFor(invocation.taskName);
+    const chain = middlewareFor(invocation.taskName, invocation.id);
 
     // Cooperative cancel signal + job context exposed to the handler.
     const controller = new AbortController();
     const context: JobContext = {
       jobId: invocation.id,
       signal: controller.signal,
-      setProgress: (progress) => queue.updateProgress(invocation.id, progress),
+      setProgress: (progress) => setProgress(invocation.id, progress),
       publish: (value) =>
-        queue.writeTaskLog(invocation.id, invocation.taskName, "result", "", JSON.stringify(value)),
+        writeTaskLog(invocation.id, invocation.taskName, "result", "", JSON.stringify(value)),
     };
     const poller = setInterval(() => {
       try {

sdks/node/src/worker.ts

@@ -103,6 +103,8 @@ export class Worker {
     // every invocation (live toggles); task/queue overrides apply here, at
     // worker startup.
     const disables = new MiddlewareDisableStore(queue);
+    // The job id is unused here: a worker has storage, so it reads the live
+    // toggle list by task name rather than taking one off the dispatch.
     const middlewareFor = (taskName: string): readonly Middleware[] => {
       const disabled = disables.getFor(taskName);
       if (disabled.length === 0) {

sdks/node/test/integrations/cli.test.ts

@@ -76,11 +76,14 @@ describe("taskito CLI", () => {
         stdio: "ignore",
         env: { ...process.env, TASKITO_INDEX: indexUrl, TASKITO_DB: db, TASKITO_MARKER: marker },
       });
-      expect(await waitForFile(marker, 6000)).toBe(true);
+      expect(await waitForFile(marker, 25_000)).toBe(true);
     } finally {
       child?.kill("SIGTERM");
     }
-  });
+    // Spawning a second Node, importing the SDK and opening SQLite before the
+    // job can run costs more than the file's default budget on a cold Windows
+    // runner, where this timed out at just over six seconds.
+  }, 30_000);
 });
 
 async function waitForFile(path: string, timeoutMs: number): Promise<boolean> {