fix(service-job,service-automation): map a degraded job outcome to sys_job_run.status instead of success (#5548)

Author: os-help

.changeset/job-run-degraded-status.md

@@ -0,0 +1,29 @@
+---
+'@objectstack/service-job': patch
+'@objectstack/service-automation': patch
+---
+
+Job runs that finish without doing their work are now audited as `degraded`, not `success` (#5548)
+
+`DbJobAdapter` decided a run's outcome solely by whether the handler threw, so a
+handler that failed internally and deliberately did not throw was recorded as
+`sys_job_run.status: 'success'` — the audit surface Studio's jobs view reads
+reported the one thing that had definitely not happened.
+
+The adapters now consume the `JobRunOutcome` channel `JobHandler` gained in
+#6617, using the `degraded` status vocabulary added in #7072:
+
+- a handler resolving `{ outcome: 'degraded', reason? }` lands
+  `sys_job_run.status: 'degraded'` with the reason in `error`, and mirrors onto
+  `sys_job.last_status` / `last_error`;
+- `degraded` is not a failure: `failure_count` stays flat and nothing retries
+  (retry keys on a rejected promise only, unchanged);
+- `IntervalJobAdapter` / `CronJobAdapter` report the same verdict through
+  `getExecutions()`, so the in-memory history and the persisted row agree.
+
+Strictly additive: a handler that resolves `undefined` — every handler written
+before #6617 — is still recorded as `success`, byte for byte as before.
+
+The first adopter is the `wait` node's timer wake-up: a shot that fires into an
+unreachable suspended-run store now reports `degraded` / `STORE_UNAVAILABLE`
+while still keeping its one-shot armed and its `sys_job` row active (#5529).

packages/services/service-automation/package.json

@@ -26,6 +26,7 @@
     "@objectstack/driver-sql": "workspace:*",
     "@objectstack/objectql": "workspace:*",
     "@objectstack/plugin-security": "workspace:*",
+    "@objectstack/service-job": "workspace:*",
     "@types/node": "^26.1.2",
     "typescript": "^6.0.3",
     "vitest": "^4.1.10"

packages/services/service-automation/src/builtin/wait-node-degraded-run.test.ts

@@ -0,0 +1,204 @@
+// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.
+
+import { describe, it, expect } from 'vitest';
+import { DbJobAdapter } from '@objectstack/service-job';
+import type { IJobService, JobSchedule, JobHandler } from '@objectstack/spec/contracts';
+import { AutomationEngine } from '../engine.js';
+import type { NodeExecutor } from '../engine.js';
+import { InMemorySuspendedRunStore } from '../suspended-run-store.js';
+import { registerWaitNode, rearmSuspendedWaitTimers } from './wait-node.js';
+
+/**
+ * #5548, end to end: the scenario that produced the finding, driven through the
+ * REAL job adapter rather than a fake that records calls.
+ *
+ * The specimen is #5529's wait wake-up firing into an unreachable durable store.
+ * That shot consumes nothing — the run stays parked, and the one-shot is kept
+ * ARMED on purpose so it can be re-fired — and it deliberately does **not**
+ * throw, because a throw is the retry signal `IJobService` implementations key
+ * on (which is why option A was rejected). The consequence, until now, was that
+ * the job's audit row said `success`: the operator-facing surface reported the
+ * one thing that definitely did not happen.
+ *
+ * Why the real `DbJobAdapter` and not a spy: the defect lives in the mapping
+ * from "what the handler reported" to "what got written", so a case that
+ * asserts the handler was called, or that it did not throw, cannot see it —
+ * that criterion IS the defect. Every assertion below reads the value in the
+ * persisted `sys_job_run` / `sys_job` cell.
+ */
+
+function silentLogger() {
+  return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } } as any;
+}
+
+/** A fake job service for "process 1", which only has to park the run. */
+function fakeJobCtx() {
+  const scheduled: Array<{ name: string; schedule: JobSchedule; handler: JobHandler }> = [];
+  const cancelled: string[] = [];
+  const job: IJobService = {
+    async schedule(name, schedule, handler) { scheduled.push({ name, schedule, handler }); },
+    async cancel(name) { cancelled.push(name); },
+    async trigger() {},
+  };
+  const ctx = { logger: silentLogger(), getService: (id: string) => (id === 'job' ? job : undefined) } as any;
+  return { ctx, scheduled, cancelled };
+}
+
+function markerExecutor(ran: string[]): NodeExecutor {
+  return { type: 'mark', async execute(node) { ran.push(node.id); return { success: true }; } };
+}
+
+/** Minimal ObjectQL stand-in for the two audit tables `DbJobAdapter` writes. */
+function makeFakeEngine() {
+  const tables = new Map<string, any[]>();
+  return {
+    tables,
+    async find(table: string, opts: any = {}) {
+      const t = tables.get(table) ?? [];
+      const out = opts.where
+        ? t.filter((r) => Object.entries(opts.where).every(([k, v]) => r[k] === v))
+        : [...t];
+      return opts.limit ? out.slice(0, opts.limit) : out;
+    },
+    async insert(table: string, data: any) {
+      const t = tables.get(table) ?? [];
+      t.push({ ...data });
+      tables.set(table, t);
+      return { id: data.id };
+    },
+    async update(table: string, patch: any) {
+      const t = tables.get(table) ?? [];
+      const r = t.find((x) => x.id === patch.id);
+      if (!r) throw new Error(`row ${patch.id} not in ${table}`);
+      Object.assign(r, patch);
+      return r;
+    },
+  };
+}
+
+const waitFlow = (waitConfig: Record<string, unknown>) => ({
+  name: 'wait_flow',
+  label: 'Wait Flow',
+  type: 'autolaunched',
+  nodes: [
+    { id: 'start', type: 'start', label: 'Start' },
+    { id: 'pause', type: 'wait', label: 'Wait', waitEventConfig: waitConfig },
+    { id: 'after', type: 'mark', label: 'After' },
+    { id: 'end', type: 'end', label: 'End' },
+  ],
+  edges: [
+    { id: 'e1', source: 'start', target: 'pause' },
+    { id: 'e2', source: 'pause', target: 'after' },
+    { id: 'e3', source: 'after', target: 'end' },
+  ],
+});
+
+const config = { eventType: 'timer', timerDuration: 'P1D' };
+
+/** A store whose resume-time `load` is unreachable; everything else works. */
+function storeWithUnreadableLoad(inner: InMemorySuspendedRunStore) {
+  return {
+    inner,
+    async save(run: any) { return inner.save(run); },
+    async load(_runId: string): Promise<any> { throw new Error('connection refused'); },
+    async delete(runId: string) { return inner.delete(runId); },
+    async list() { return inner.list(); },
+  };
+}
+
+/**
+ * Park a run in "process 1", then cold-boot "process 2" whose durable read is
+ * broken and whose job service is a real `DbJobAdapter`. Returns the adapter,
+ * the fake ObjectQL tables, and the wake-up job's name.
+ */
+async function coldBootOntoDbJobAdapter(broken: boolean) {
+  const inner = new InMemorySuspendedRunStore();
+  const boot1 = fakeJobCtx();
+  const e1 = new AutomationEngine(silentLogger());
+  e1.registerNodeExecutor(markerExecutor([]));
+  registerWaitNode(e1, boot1.ctx);
+  e1.setSuspendedRunStore(inner);
+  e1.registerFlow('wait_flow', waitFlow(config));
+  const paused = await e1.execute('wait_flow');
+  expect(paused.status).toBe('paused');
+
+  const store = broken ? (storeWithUnreadableLoad(inner) as any) : inner;
+  const ran: string[] = [];
+  const objectql = makeFakeEngine();
+  const jobService = new DbJobAdapter({ engine: objectql });
+  const e2 = new AutomationEngine(silentLogger());
+  e2.registerNodeExecutor(markerExecutor(ran));
+  // The wait node is registered against the REAL adapter, so the teardown that
+  // fires when the run leaves the node (#5512) goes through it too.
+  registerWaitNode(e2, {
+    logger: silentLogger(),
+    getService: (id: string) => (id === 'job' ? jobService : undefined),
+  } as any);
+  e2.setSuspendedRunStore(store);
+  e2.registerFlow('wait_flow', waitFlow(config));
+  // The re-arm pass registers the one-shot on the real adapter — from here on
+  // every run of that job goes through `DbJobAdapter.wrap`.
+  expect(await rearmSuspendedWaitTimers(e2, store, jobService, silentLogger())).toBe(1);
+
+  return { paused, inner, ran, objectql, jobService, jobName: `flow-wait:${paused.runId}:pause` };
+}
+
+describe('#5548 — the #5529 wait wake-up that consumed nothing is audited as degraded, not success', () => {
+  it('the wake-up into an unreachable store lands sys_job_run.status = "degraded"', async () => {
+    const { inner, ran, objectql, jobService, jobName, paused } = await coldBootOntoDbJobAdapter(true);
+
+    // Fire the wake-up the way an operator or the timer would.
+    await jobService.trigger(jobName);
+
+    // The pause really was not consumed — the run is still parked, its row still
+    // there. This is the premise the audit row has to reflect.
+    expect(ran).toEqual([]);
+    expect((await inner.list()).map((r) => r.runId)).toEqual([paused.runId]);
+
+    const runs = objectql.tables.get('sys_job_run') ?? [];
+    expect(runs).toHaveLength(1);
+    expect(runs[0].job_name).toBe(jobName);
+    // The cell this card exists for. Before the wiring it read 'success'.
+    expect(runs[0].status).toBe('degraded');
+    expect(runs[0].error).toBe('STORE_UNAVAILABLE');
+
+    await jobService.destroy();
+  });
+
+  it('the job row mirrors it, stays active, and does NOT count as a failure', async () => {
+    const { objectql, jobService, jobName } = await coldBootOntoDbJobAdapter(true);
+    await jobService.trigger(jobName);
+
+    const [job] = objectql.tables.get('sys_job') ?? [];
+    expect(job.last_status).toBe('degraded');
+    expect(job.last_error).toBe('STORE_UNAVAILABLE');
+    // #5529's half is untouched: the one-shot is kept ARMED so the stuck run
+    // stays visible and the wake-up re-firable.
+    expect(job.active).toBe(true);
+    // `degraded` is not a failure: the retry/alerting signal does not move.
+    expect(job.failure_count).toBe(0);
+    expect(job.run_count).toBe(1);
+
+    await jobService.destroy();
+  });
+
+  it('a wake-up that DOES resume the run is still audited as success (control)', async () => {
+    const { ran, objectql, jobService, jobName } = await coldBootOntoDbJobAdapter(false);
+    await jobService.trigger(jobName);
+
+    // The pause was consumed and traversal continued…
+    expect(ran).toEqual(['after']);
+    const runs = objectql.tables.get('sys_job_run') ?? [];
+    expect(runs).toHaveLength(1);
+    // …so the row says success, exactly as before this change.
+    expect(runs[0].status).toBe('success');
+    expect(runs[0].error).toBeNull();
+    const [job] = objectql.tables.get('sys_job') ?? [];
+    expect(job.last_status).toBe('success');
+    // The one-shot had its shot and settled the pause, so it disarms — the
+    // `sys_job` row goes inactive, which is the OPPOSITE of the degraded case.
+    expect(job.active).toBe(false);
+
+    await jobService.destroy();
+  });
+});

packages/services/service-automation/src/builtin/wait-node.ts

@@ -2,7 +2,7 @@
 
 import type { PluginContext } from '@objectstack/core';
 import { defineActionDescriptor } from '@objectstack/spec/automation';
-import type { IJobService } from '@objectstack/spec/contracts';
+import type { IJobService, JobRunOutcome } from '@objectstack/spec/contracts';
 import type { AutomationEngine, SuspendedRunStore } from '../engine.js';
 import { describeThrownForLog, type ThrownCauseMeta } from '../thrown-cause-diagnostics.js';
 
@@ -56,6 +56,13 @@ interface WaitTimerLogger {
  *    healthy timer is the durability degradation AGENTS.md's log-level rule is
  *    about (#4632), and this path was previously silent — the result was
  *    discarded by the callback without so much as a `warn`.
+ *
+ *    Since #5548 this outcome is also RESOLVED as `{ outcome: 'degraded',
+ *    reason: 'STORE_UNAVAILABLE' }`, so the job service records the run as
+ *    `degraded` rather than `success` — the log line said the shot missed, the
+ *    audit row said it succeeded. The `reason` is the short code only: the
+ *    driver's own message can be multi-line and belongs in the log record's
+ *    `meta` (#5737), not in an audit column.
  *  - **everything else** — cancel, exactly as before. Success consumed the
  *    pause; `RESUME_IN_PROGRESS` means a concurrent resume is consuming it (and
  *    #5512's `onSuspensionReleased` drops this job when it does); a machine-state
@@ -89,15 +96,26 @@ function makeWaitTimerJobHandler(
   runId: string,
   jobName: string,
   logger: WaitTimerLogger,
-): () => Promise<void> {
+): () => Promise<void | JobRunOutcome> {
   return async () => {
     // Set only on the one outcome that must NOT disarm the job. A thrown
     // `resume` leaves it false, so the `finally` still cancels.
     let keepArmed = false;
+    // #5548 — what this shot reports to the JOB service, as opposed to what it
+    // logs. `STORE_UNAVAILABLE` is the specimen the ruling names: the handler
+    // completes normally (deliberately — #5529 refused to make it throw,
+    // because a throw is the retry signal third-party `IJobService`
+    // implementations key on), so before the `JobRunOutcome` channel existed
+    // the run was recorded as `success` on an audit surface whose whole job is
+    // to say whether the work happened. Resolving `degraded` instead moves the
+    // `sys_job_run` row and nothing else: still no throw, still no retry, and
+    // the job still stays ARMED and `active` exactly as #5529 fixed it.
+    let outcome: JobRunOutcome | undefined;
     try {
       const result = await engine.resume(runId);
       if (result?.code === 'STORE_UNAVAILABLE') {
         keepArmed = true;
+        outcome = { outcome: 'degraded', reason: 'STORE_UNAVAILABLE' };
         // #5737 — the cause goes to `meta`, never into the message. This one is
         // NOT a thrown value and so does NOT go through `describeThrownForLog`:
         // `AutomationResult.error` is a STRING the engine already composed
@@ -138,6 +156,9 @@ function makeWaitTimerJobHandler(
         }
       }
     }
+    // Resolved, never thrown — the report rides the RETURN value precisely so
+    // the failure semantics of `IJobService` stay untouched (#6617).
+    return outcome;
   };
 }
 

packages/services/service-job/src/cron-job-adapter.ts

@@ -155,8 +155,16 @@ export class CronJobAdapter implements IJobService {
     };
     const startMs = Date.now();
     try {
-      await runWithPolicy(record.name, () => record.handler({ jobId: record.name, data }), record.options);
-      execution.status = 'success';
+      const outcome = await runWithPolicy(record.name, () => record.handler({ jobId: record.name, data }), record.options);
+      // #5548 — same mapping as `IntervalJobAdapter.executeJob`, deliberately
+      // one shape and not two spellings: a resolved `degraded` outcome is a
+      // completed run whose work did not happen, never a `success`.
+      if (outcome && outcome.outcome === 'degraded') {
+        execution.status = 'degraded';
+        execution.error = outcome.reason;
+      } else {
+        execution.status = 'success';
+      }
     } catch (err) {
       execution.status = err instanceof JobTimeoutError ? 'timeout' : 'failed';
       execution.error = err instanceof Error ? err.message : String(err);