|
| 1 | +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. |
| 2 | + |
| 3 | +import { describe, it, expect } from 'vitest'; |
| 4 | +import { assertEngineUpdateDispatch } from '@objectstack/metadata-core'; |
| 5 | +import { DbJobAdapter } from '@objectstack/service-job'; |
| 6 | +import type { IJobService, JobSchedule, JobHandler } from '@objectstack/spec/contracts'; |
| 7 | +import { AutomationEngine } from '../engine.js'; |
| 8 | +import type { NodeExecutor } from '../engine.js'; |
| 9 | +import { InMemorySuspendedRunStore } from '../suspended-run-store.js'; |
| 10 | +import { registerWaitNode, rearmSuspendedWaitTimers } from './wait-node.js'; |
| 11 | + |
| 12 | +/** |
| 13 | + * #5548, end to end: the scenario that produced the finding, driven through the |
| 14 | + * REAL job adapter rather than a fake that records calls. |
| 15 | + * |
| 16 | + * The specimen is #5529's wait wake-up firing into an unreachable durable store. |
| 17 | + * That shot consumes nothing — the run stays parked, and the one-shot is kept |
| 18 | + * ARMED on purpose so it can be re-fired — and it deliberately does **not** |
| 19 | + * throw, because a throw is the retry signal `IJobService` implementations key |
| 20 | + * on (which is why option A was rejected). The consequence, until now, was that |
| 21 | + * the job's audit row said `success`: the operator-facing surface reported the |
| 22 | + * one thing that definitely did not happen. |
| 23 | + * |
| 24 | + * Why the real `DbJobAdapter` and not a spy: the defect lives in the mapping |
| 25 | + * from "what the handler reported" to "what got written", so a case that |
| 26 | + * asserts the handler was called, or that it did not throw, cannot see it — |
| 27 | + * that criterion IS the defect. Every assertion below reads the value in the |
| 28 | + * persisted `sys_job_run` / `sys_job` cell. |
| 29 | + */ |
| 30 | + |
| 31 | +function silentLogger() { |
| 32 | + return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } } as any; |
| 33 | +} |
| 34 | + |
| 35 | +/** A fake job service for "process 1", which only has to park the run. */ |
| 36 | +function fakeJobCtx() { |
| 37 | + const scheduled: Array<{ name: string; schedule: JobSchedule; handler: JobHandler }> = []; |
| 38 | + const cancelled: string[] = []; |
| 39 | + const job: IJobService = { |
| 40 | + async schedule(name, schedule, handler) { scheduled.push({ name, schedule, handler }); }, |
| 41 | + async cancel(name) { cancelled.push(name); }, |
| 42 | + async trigger() {}, |
| 43 | + }; |
| 44 | + const ctx = { logger: silentLogger(), getService: (id: string) => (id === 'job' ? job : undefined) } as any; |
| 45 | + return { ctx, scheduled, cancelled }; |
| 46 | +} |
| 47 | + |
| 48 | +function markerExecutor(ran: string[]): NodeExecutor { |
| 49 | + return { type: 'mark', async execute(node) { ran.push(node.id); return { success: true }; } }; |
| 50 | +} |
| 51 | + |
| 52 | +/** Minimal ObjectQL stand-in for the two audit tables `DbJobAdapter` writes. */ |
| 53 | +function makeFakeEngine() { |
| 54 | + const tables = new Map<string, any[]>(); |
| 55 | + return { |
| 56 | + tables, |
| 57 | + async find(table: string, opts: any = {}) { |
| 58 | + const t = tables.get(table) ?? []; |
| 59 | + const out = opts.where |
| 60 | + ? t.filter((r) => Object.entries(opts.where).every(([k, v]) => r[k] === v)) |
| 61 | + : [...t]; |
| 62 | + return opts.limit ? out.slice(0, opts.limit) : out; |
| 63 | + }, |
| 64 | + async insert(table: string, data: any) { |
| 65 | + const t = tables.get(table) ?? []; |
| 66 | + t.push({ ...data }); |
| 67 | + tables.set(table, t); |
| 68 | + return { id: data.id }; |
| 69 | + }, |
| 70 | + async update(table: string, patch: any, options?: any) { |
| 71 | + // Same binding as the sibling double in `service-job`: the fake refuses |
| 72 | + // exactly what `ObjectQLEngine.update` refuses, so the audit writes this |
| 73 | + // test asserts are writes a real server would have accepted. |
| 74 | + assertEngineUpdateDispatch(patch, options); |
| 75 | + const t = tables.get(table) ?? []; |
| 76 | + const r = t.find((x) => x.id === patch.id); |
| 77 | + if (!r) throw new Error(`row ${patch.id} not in ${table}`); |
| 78 | + Object.assign(r, patch); |
| 79 | + return r; |
| 80 | + }, |
| 81 | + }; |
| 82 | +} |
| 83 | + |
| 84 | +const waitFlow = (waitConfig: Record<string, unknown>) => ({ |
| 85 | + name: 'wait_flow', |
| 86 | + label: 'Wait Flow', |
| 87 | + type: 'autolaunched', |
| 88 | + nodes: [ |
| 89 | + { id: 'start', type: 'start', label: 'Start' }, |
| 90 | + { id: 'pause', type: 'wait', label: 'Wait', waitEventConfig: waitConfig }, |
| 91 | + { id: 'after', type: 'mark', label: 'After' }, |
| 92 | + { id: 'end', type: 'end', label: 'End' }, |
| 93 | + ], |
| 94 | + edges: [ |
| 95 | + { id: 'e1', source: 'start', target: 'pause' }, |
| 96 | + { id: 'e2', source: 'pause', target: 'after' }, |
| 97 | + { id: 'e3', source: 'after', target: 'end' }, |
| 98 | + ], |
| 99 | +}); |
| 100 | + |
| 101 | +const config = { eventType: 'timer', timerDuration: 'P1D' }; |
| 102 | + |
| 103 | +/** A store whose resume-time `load` is unreachable; everything else works. */ |
| 104 | +function storeWithUnreadableLoad(inner: InMemorySuspendedRunStore) { |
| 105 | + return { |
| 106 | + inner, |
| 107 | + async save(run: any) { return inner.save(run); }, |
| 108 | + async load(_runId: string): Promise<any> { throw new Error('connection refused'); }, |
| 109 | + async delete(runId: string) { return inner.delete(runId); }, |
| 110 | + async list() { return inner.list(); }, |
| 111 | + }; |
| 112 | +} |
| 113 | + |
| 114 | +/** |
| 115 | + * Park a run in "process 1", then cold-boot "process 2" whose durable read is |
| 116 | + * broken and whose job service is a real `DbJobAdapter`. Returns the adapter, |
| 117 | + * the fake ObjectQL tables, and the wake-up job's name. |
| 118 | + */ |
| 119 | +async function coldBootOntoDbJobAdapter(broken: boolean) { |
| 120 | + const inner = new InMemorySuspendedRunStore(); |
| 121 | + const boot1 = fakeJobCtx(); |
| 122 | + const e1 = new AutomationEngine(silentLogger()); |
| 123 | + e1.registerNodeExecutor(markerExecutor([])); |
| 124 | + registerWaitNode(e1, boot1.ctx); |
| 125 | + e1.setSuspendedRunStore(inner); |
| 126 | + e1.registerFlow('wait_flow', waitFlow(config)); |
| 127 | + const paused = await e1.execute('wait_flow'); |
| 128 | + expect(paused.status).toBe('paused'); |
| 129 | + |
| 130 | + const store = broken ? (storeWithUnreadableLoad(inner) as any) : inner; |
| 131 | + const ran: string[] = []; |
| 132 | + const objectql = makeFakeEngine(); |
| 133 | + const jobService = new DbJobAdapter({ engine: objectql }); |
| 134 | + const e2 = new AutomationEngine(silentLogger()); |
| 135 | + e2.registerNodeExecutor(markerExecutor(ran)); |
| 136 | + // The wait node is registered against the REAL adapter, so the teardown that |
| 137 | + // fires when the run leaves the node (#5512) goes through it too. |
| 138 | + registerWaitNode(e2, { |
| 139 | + logger: silentLogger(), |
| 140 | + getService: (id: string) => (id === 'job' ? jobService : undefined), |
| 141 | + } as any); |
| 142 | + e2.setSuspendedRunStore(store); |
| 143 | + e2.registerFlow('wait_flow', waitFlow(config)); |
| 144 | + // The re-arm pass registers the one-shot on the real adapter — from here on |
| 145 | + // every run of that job goes through `DbJobAdapter.wrap`. |
| 146 | + expect(await rearmSuspendedWaitTimers(e2, store, jobService, silentLogger())).toBe(1); |
| 147 | + |
| 148 | + return { paused, inner, ran, objectql, jobService, jobName: `flow-wait:${paused.runId}:pause` }; |
| 149 | +} |
| 150 | + |
| 151 | +describe('#5548 — the #5529 wait wake-up that consumed nothing is audited as degraded, not success', () => { |
| 152 | + it('the wake-up into an unreachable store lands sys_job_run.status = "degraded"', async () => { |
| 153 | + const { inner, ran, objectql, jobService, jobName, paused } = await coldBootOntoDbJobAdapter(true); |
| 154 | + |
| 155 | + // Fire the wake-up the way an operator or the timer would. |
| 156 | + await jobService.trigger(jobName); |
| 157 | + |
| 158 | + // The pause really was not consumed — the run is still parked, its row still |
| 159 | + // there. This is the premise the audit row has to reflect. |
| 160 | + expect(ran).toEqual([]); |
| 161 | + expect((await inner.list()).map((r) => r.runId)).toEqual([paused.runId]); |
| 162 | + |
| 163 | + const runs = objectql.tables.get('sys_job_run') ?? []; |
| 164 | + expect(runs).toHaveLength(1); |
| 165 | + expect(runs[0].job_name).toBe(jobName); |
| 166 | + // The cell this card exists for. Before the wiring it read 'success'. |
| 167 | + expect(runs[0].status).toBe('degraded'); |
| 168 | + expect(runs[0].error).toBe('STORE_UNAVAILABLE'); |
| 169 | + |
| 170 | + await jobService.destroy(); |
| 171 | + }); |
| 172 | + |
| 173 | + it('the job row mirrors it, stays active, and does NOT count as a failure', async () => { |
| 174 | + const { objectql, jobService, jobName } = await coldBootOntoDbJobAdapter(true); |
| 175 | + await jobService.trigger(jobName); |
| 176 | + |
| 177 | + const [job] = objectql.tables.get('sys_job') ?? []; |
| 178 | + expect(job.last_status).toBe('degraded'); |
| 179 | + expect(job.last_error).toBe('STORE_UNAVAILABLE'); |
| 180 | + // #5529's half is untouched: the one-shot is kept ARMED so the stuck run |
| 181 | + // stays visible and the wake-up re-firable. |
| 182 | + expect(job.active).toBe(true); |
| 183 | + // `degraded` is not a failure: the retry/alerting signal does not move. |
| 184 | + expect(job.failure_count).toBe(0); |
| 185 | + expect(job.run_count).toBe(1); |
| 186 | + |
| 187 | + await jobService.destroy(); |
| 188 | + }); |
| 189 | + |
| 190 | + it('a wake-up that DOES resume the run is still audited as success (control)', async () => { |
| 191 | + const { ran, objectql, jobService, jobName } = await coldBootOntoDbJobAdapter(false); |
| 192 | + await jobService.trigger(jobName); |
| 193 | + |
| 194 | + // The pause was consumed and traversal continued… |
| 195 | + expect(ran).toEqual(['after']); |
| 196 | + const runs = objectql.tables.get('sys_job_run') ?? []; |
| 197 | + expect(runs).toHaveLength(1); |
| 198 | + // …so the row says success, exactly as before this change. |
| 199 | + expect(runs[0].status).toBe('success'); |
| 200 | + expect(runs[0].error).toBeNull(); |
| 201 | + const [job] = objectql.tables.get('sys_job') ?? []; |
| 202 | + expect(job.last_status).toBe('success'); |
| 203 | + // The one-shot had its shot and settled the pause, so it disarms — the |
| 204 | + // `sys_job` row goes inactive, which is the OPPOSITE of the degraded case. |
| 205 | + expect(job.active).toBe(false); |
| 206 | + |
| 207 | + await jobService.destroy(); |
| 208 | + }); |
| 209 | +}); |
0 commit comments