Balance pooled job distribution across runner children (first-fit → round-robin)

spec/background-jobs/worker-pooled-distribution-spec.js

@@ -0,0 +1,93 @@
+// @ts-check
+
+import BackgroundJobsWorker from "../../src/background-jobs/worker.js"
+import {describe, expect, it} from "../../src/testing/test.js"
+
+/**
+ * @param {string} id
+ * @returns {import("../../src/background-jobs/types.js").BackgroundJobPayload & {id: string}}
+ */
+function fakePayload(id) {
+  return /** @type {import("../../src/background-jobs/types.js").BackgroundJobPayload & {id: string}} */ (/** @type {unknown} */ ({id}))
+}
+
+/**
+ * Adds a pooled child to a worker without forking a real process, so pooled child
+ * selection can be exercised directly.
+ * @param {BackgroundJobsWorker} worker - Worker under test.
+ * @param {{inflight?: number, lastDispatchSeq?: number, retiring?: boolean}} [args] - Seed state.
+ * @returns {import("node:child_process").ChildProcess} - The fake child.
+ */
+function addFakePooledChild(worker, {inflight = 0, lastDispatchSeq = 0, retiring = false} = {}) {
+  const child = /** @type {import("node:child_process").ChildProcess} */ (/** @type {unknown} */ ({send() {}}))
+  /** @type {Map<string, {payload: import("../../src/background-jobs/types.js").BackgroundJobPayload & {id: string}}>} */
+  const inflightMap = new Map()
+
+  for (let index = 0; index < inflight; index++) {
+    inflightMap.set(`seed-${worker.pooledChildren.size}-${index}`, {payload: fakePayload(`seed-${index}`)})
+  }
+
+  worker.pooledChildren.add(child)
+  worker.pooledChildStates.set(child, {createdAtMs: 0, inflight: inflightMap, jobsRun: 0, lastDispatchSeq, retiring})
+
+  return child
+}
+
+/**
+ * @param {BackgroundJobsWorker} worker - Worker under test.
+ * @param {import("node:child_process").ChildProcess} child - Pooled child.
+ * @returns {number} - Its in-flight job count.
+ */
+function inflightSize(worker, child) {
+  const state = worker.pooledChildStates.get(child)
+
+  if (!state) throw new Error("Missing pooled child state")
+
+  return state.inflight.size
+}
+
+describe("Background jobs - pooled distribution", () => {
+  it("spreads jobs evenly across children instead of first-fit packing the earliest", () => {
+    const worker = new BackgroundJobsWorker({pooledRunnerConcurrency: 4, pooledRunnerCount: 3})
+    const first = addFakePooledChild(worker)
+    const second = addFakePooledChild(worker)
+    const third = addFakePooledChild(worker)
+
+    for (let index = 0; index < 6; index++) worker._runPooledJob(fakePayload(`job-${index}`))
+
+    // Round-robin: two each. First-fit would have packed [4, 2, 0].
+    expect([inflightSize(worker, first), inflightSize(worker, second), inflightSize(worker, third)]).toEqual([2, 2, 2])
+  })
+
+  it("does not burst a freshly added child to catch up to already-loaded children", () => {
+    const worker = new BackgroundJobsWorker({pooledRunnerConcurrency: 10, pooledRunnerCount: 3})
+
+    worker._pooledDispatchSeq = 2
+
+    const loadedA = addFakePooledChild(worker, {inflight: 2, lastDispatchSeq: 1})
+    const loadedB = addFakePooledChild(worker, {inflight: 2, lastDispatchSeq: 2})
+    const fresh = addFakePooledChild(worker, {inflight: 0, lastDispatchSeq: 0})
+
+    for (let index = 0; index < 3; index++) worker._runPooledJob(fakePayload(`job-${index}`))
+
+    // The fresh child takes one job as its turn comes up, then the rotation moves on to
+    // the others — not a burst of all three to "catch up" (which naive least-loaded,
+    // always picking the lowest in-flight, would do).
+    expect(inflightSize(worker, fresh)).toEqual(1)
+    expect(inflightSize(worker, loadedA)).toEqual(3)
+    expect(inflightSize(worker, loadedB)).toEqual(3)
+  })
+
+  it("skips retiring children and lazily spawns only when every non-retiring child is full", () => {
+    const worker = new BackgroundJobsWorker({pooledRunnerConcurrency: 2, pooledRunnerCount: 4})
+    const retiring = addFakePooledChild(worker, {retiring: true})
+    const full = addFakePooledChild(worker, {inflight: 2})
+
+    // No non-retiring child has a free slot, so selection returns undefined (the caller
+    // then spawns a new child).
+    expect(worker._selectPooledChild()).toEqual(undefined)
+    // The retiring child is never selected even though it has open slots.
+    expect(inflightSize(worker, retiring)).toEqual(0)
+    expect(inflightSize(worker, full)).toEqual(2)
+  })
+})

src/background-jobs/worker.js

@@ -188,8 +188,11 @@ export default class BackgroundJobsWorker {
     this.inflightPooledJobs = new Set()
     /** @type {Set<import("node:child_process").ChildProcess>} */
     this.pooledChildren = new Set()
-    /** @type {Map<import("node:child_process").ChildProcess, {createdAtMs: number, jobsRun: number, inflight: Map<string, {payload: import("./types.js").BackgroundJobPayload & {id: string}, resolve?: (value: void) => void, timeoutTimer?: ReturnType<typeof setTimeout> | null}>, retiring: boolean, settling?: boolean, timeoutSigkillTimer?: ReturnType<typeof setTimeout> | null}>} */
+    /** @type {Map<import("node:child_process").ChildProcess, {createdAtMs: number, jobsRun: number, inflight: Map<string, {payload: import("./types.js").BackgroundJobPayload & {id: string}, resolve?: (value: void) => void, timeoutTimer?: ReturnType<typeof setTimeout> | null}>, lastDispatchSeq: number, retiring: boolean, settling?: boolean, timeoutSigkillTimer?: ReturnType<typeof setTimeout> | null}>} */
     this.pooledChildStates = new Map()
+    // Monotonic dispatch counter for round-robin child selection: each dispatch stamps
+    // the chosen child, and selection prefers the child dispatched least recently.
+    this._pooledDispatchSeq = 0
   }
 
   /**
@@ -643,18 +646,13 @@ export default class BackgroundJobsWorker {
    * @returns {Promise<void>} - Resolves after the durable report.
    */
   _runPooledJob(payload) {
-    let child
-    for (const candidate of this.pooledChildren) {
-      const state = this.pooledChildStates.get(candidate)
-      if (state && !state.retiring && state.inflight.size < this.pooledRunnerConcurrency) {
-        child = candidate
-        break
-      }
-    }
-    if (!child) child = this._createPooledChild()
+    const child = this._selectPooledChild() || this._createPooledChild()
     const state = this.pooledChildStates.get(child)
     if (!state) throw new Error("Pooled runner state missing")
 
+    // Stamp the round-robin cursor so the next dispatch prefers a different child.
+    state.lastDispatchSeq = ++this._pooledDispatchSeq
+
     return new Promise((resolve) => {
       const timeoutTimer = this._armPooledJobTimeout({child, jobId: payload.id})
 
@@ -667,6 +665,36 @@ export default class BackgroundJobsWorker {
     })
   }
 
+  /**
+   * Selects a pooled child to run the next job, or undefined when every non-retiring
+   * child is already full (the caller then lazily spawns one). Among children with a
+   * free concurrency slot, picks the one dispatched least recently — a round-robin that
+   * spreads jobs (notably multi-minute RunBuildJobs, each pinning a tenant connection
+   * for its whole run) evenly across children instead of first-fit packing the earliest
+   * one until it is full. A freshly spawned or replacement child therefore takes its
+   * fair share one job at a time as its turn comes up, rather than absorbing a burst to
+   * "catch up" to the others.
+   * @returns {import("node:child_process").ChildProcess | undefined} - The chosen child, or undefined when all non-retiring children are full.
+   */
+  _selectPooledChild() {
+    /** @type {import("node:child_process").ChildProcess | undefined} */
+    let selected
+    let selectedSeq = Infinity
+
+    for (const child of this.pooledChildren) {
+      const state = this.pooledChildStates.get(child)
+
+      if (!state || state.retiring || state.inflight.size >= this.pooledRunnerConcurrency) continue
+
+      if (state.lastDispatchSeq < selectedSeq) {
+        selected = child
+        selectedSeq = state.lastDispatchSeq
+      }
+    }
+
+    return selected
+  }
+
   /**
    * Arms a per-job wall-clock backstop for a pooled job. A pooled child hosts many
    * concurrent jobs, so a single genuinely-hung job would otherwise pin its
@@ -733,7 +761,7 @@ export default class BackgroundJobsWorker {
     })
     this.pooledChildren.add(child)
     this.inflightProcessChildren.add(child)
-    this.pooledChildStates.set(child, {createdAtMs: Date.now(), jobsRun: 0, inflight: new Map(), retiring: false})
+    this.pooledChildStates.set(child, {createdAtMs: Date.now(), jobsRun: 0, inflight: new Map(), lastDispatchSeq: 0, retiring: false})
     child.on("message", (message) => this._handlePooledChildMessage({child, message}))
     child.once("exit", (code, signal) => this._handlePooledChildFailure({child, error: new Error(`Pooled background job runner exited: code=${code} signal=${signal || "none"}`)}))
     child.once("error", (error) => this._handlePooledChildFailure({child, error}))