Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions .changeset/job-run-degraded-status.md
Original file line number Diff line number Diff line change
@@ -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).
2 changes: 2 additions & 0 deletions packages/services/service-automation/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,10 @@
},
"devDependencies": {
"@objectstack/driver-sql": "workspace:*",
"@objectstack/metadata-core": "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"
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,209 @@
// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.

import { describe, it, expect } from 'vitest';
import { assertEngineUpdateDispatch } from '@objectstack/metadata-core';
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, options?: any) {
// Same binding as the sibling double in `service-job`: the fake refuses
// exactly what `ObjectQLEngine.update` refuses, so the audit writes this
// test asserts are writes a real server would have accepted.
assertEngineUpdateDispatch(patch, options);
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();
});
});
25 changes: 23 additions & 2 deletions packages/services/service-automation/src/builtin/wait-node.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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;
};
}

Expand Down
1 change: 1 addition & 0 deletions packages/services/service-job/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
"croner": "^10.0.1"
},
"devDependencies": {
"@objectstack/metadata-core": "workspace:*",
"@types/node": "^26.1.2",
"typescript": "^6.0.3",
"vitest": "^4.1.10"
Expand Down
12 changes: 10 additions & 2 deletions packages/services/service-job/src/cron-job-adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading
Loading