Function createFleetQueueClient

showcase/harness/src/fleet/queue-client.ts:781–2330

(
  config: FleetQueueClientConfig,
)

}

export function createFleetQueueClient(
  config: FleetQueueClientConfig,
): FleetQueueClientWithSweepOutcome {
  const { pb, claim, logger } = config;
  const rng = config.rng ?? Math.random;
  const now = config.now ?? Date.now;
  const sleep =
    config.sleep ??
    ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
  const stalePolicy = config.stalePending ?? {};
  const staleExpiryPeriods =
    stalePolicy.expiryPeriods ?? DEFAULT_STALE_PENDING_EXPIRY_PERIODS;
  const staleDefaultPeriodMs =
    stalePolicy.defaultPeriodMs ?? DEFAULT_STALE_PENDING_FAMILY_PERIOD_MS;
  const stalePeriodMsFor = (family: string): number =>
    stalePolicy.familyPeriodsMs?.[family] ?? staleDefaultPeriodMs;

  // Per-client cache of a claimed job's decoded payload, keyed by jobId. The
  // renew CAS returns the lifecycle columns but NOT the payload, and the
  // convenience re-read used to re-hydrate it can momentarily fail (a PB read
  // blip). Throwing on that blip permanently stops the worker's heartbeat,
  // after which the sweeper reclaims the still-live job and synthesizes a FALSE
  // `worker-crashed-mid-job` comm error. So we remember the payload at claim
  // time and reuse it on renew, making the re-read a non-fatal convenience.
  const payloadCache = new Map<string, ServiceJobPayload>();

  // Last-known GOOD lease per held job, set at claim time and refreshed on
  // every successful renew. This is what an INDETERMINATE renew (a thrown
  // claim.renewLease — 5xx or 2xx-unreadable; the renew may or may not have
  // committed) hands back to the heartbeat: the worker heartbeat treats a
  // renewLease THROW like a lost lease (it breaks), so letting the throw
  // escape kills the heartbeat → the sweeper reclaims a LIVE job → a FALSE
  // worker-crashed-mid-job. Returning the current lease unchanged keeps the
  // heartbeat alive so the NEXT beat retries; only a definitive
  // `renewed: false` stops it. Evicted exactly where payloadCache is.
  const leaseCache = new Map<string, JobLease>();

  // CONCURRENT-SWEEP LATCH (see sweepExpired): the in-flight sweep's
  // promise, or null when none is running. A sweep arriving while one is in
  // flight piggybacks on this promise instead of racing the in-flight
  // sweep's per-call grace set.
  let sweepInFlight: Promise<SweepResultWithIndeterminate> | null = null;

  /**
   * Bounded-retry persistence of a per-service result onto an ALREADY
   * TERMINAL probe_jobs row — the SEPARATE record write that follows a
   * release CAS (migration 1779989700 adds `result` + `result_processed`).
   * Never throws: returns `{ ok: true }` on success, or `{ ok: false,
   * lastErr }` after RESULT_WRITE_MAX_ATTEMPTS attempts so the caller
   * decides what the loss means (`report` — distinct "result lost" error).
   * Used by `report()` ONLY: the decode-failure attribution in `claimNext`
   * is a deliberate single-attempt write (its retry pacing would stall the
   * claim race, and the consumer's crash synthesis is its backstop).
   * `result_processed` seeds false: the consumer latches it true after
   * aggregating exactly once.
   */
  async function writeResult(
    jobId: string,