diff --git a/packages/workflow-executor/src/adapters/server-types.ts b/packages/workflow-executor/src/adapters/server-types.ts index aece818126..52350418a9 100644 --- a/packages/workflow-executor/src/adapters/server-types.ts +++ b/packages/workflow-executor/src/adapters/server-types.ts @@ -370,9 +370,23 @@ export const ServerAutomatedInboxAssignmentsResponseSchema = z.object({ assignments: z.array(ServerAutomatedInboxAssignmentSchema), }); +export const SERVER_AUTOMATED_INBOX_READ_FAILURE_REASONS = [ + 'agent-forbidden', + 'agent-unreachable', + 'segment-read-failed', +] as const; +export type ServerAutomatedInboxReadFailureReason = + (typeof SERVER_AUTOMATED_INBOX_READ_FAILURE_REASONS)[number]; + +export interface ServerAutomatedInboxReadFailure { + reason: ServerAutomatedInboxReadFailureReason; + httpStatus?: number; +} + export interface ServerAutomatedInboxSyncRequest { closed: { recordId: string; stillInSegment: boolean }[]; candidates: string[]; + readFailure?: ServerAutomatedInboxReadFailure; } export const SERVER_AUTOMATED_INBOX_SYNC_OUTCOMES = [ diff --git a/packages/workflow-executor/src/automation-poller.ts b/packages/workflow-executor/src/automation-poller.ts index b8e5232e56..1fa0162a6d 100644 --- a/packages/workflow-executor/src/automation-poller.ts +++ b/packages/workflow-executor/src/automation-poller.ts @@ -1,16 +1,18 @@ import type { ServerAutomatedInboxAssignment, ServerAutomatedInboxConfig, + ServerAutomatedInboxReadFailure, } from './adapters/server-types'; import type { AutomationPort } from './ports/automation-port'; import type { Logger } from './ports/logger-port'; import type { SegmentReaderPort } from './ports/segment-reader-port'; +import { AgentHttpError } from '@forestadmin/agent-client'; import { IANAZone } from 'luxon'; import createConsoleLogger from './adapters/console-logger'; import { DEFAULT_STOP_TIMEOUT_S } from './defaults'; -import { AutomatedInboxGoneError, extractErrorMessage } from './errors'; +import { AgentPortError, AutomatedInboxGoneError, extractErrorMessage } from './errors'; import InFlightRunRegistry from './in-flight-run-registry'; // One membership question per chunk, small enough that a `pk In (...)` stays a query an agent will @@ -32,7 +34,7 @@ const MAX_EXCLUDED_RECORDS = 150; // Every inbox reads the customer's agent several times. Sweeping them all at once piles those reads // onto the customer's database, and agent-client's ten-second timeout turns the pile-up into inboxes -// that skip their sync. +// whose reads time out. const MAX_CONCURRENT_INBOX_POLLS = 5; // The orchestrator's poller lease lives three of these, so a dead holder is replaced within a @@ -72,6 +74,53 @@ const isTerminalRun = (runState: string | null | undefined): boolean => const isLiveRun = (runState: string | null | undefined): boolean => runState != null && LIVE_RUN_STATES.has(runState); +const FORBIDDEN_AGENT_STATUSES: ReadonlySet = new Set([401, 403]); + +const UNREACHABLE_AGENT_STATUSES: ReadonlySet = new Set([502, 503, 504]); + +// Superagent's own timeout is ECONNABORTED. Any other failure without an HTTP answer (a JWT that +// cannot be signed, a malformed agent URL) is a fault on our side, not an agent to go and restart. +const UNREACHABLE_AGENT_ERROR_CODES: ReadonlySet = new Set([ + 'ECONNABORTED', + 'ECONNREFUSED', + 'ECONNRESET', + 'EAI_AGAIN', + 'EHOSTUNREACH', + 'ENETUNREACH', + 'ENOTFOUND', + 'EPIPE', + 'ETIMEDOUT', +]); + +// agent-client reports status 0 when a response carries none. 0 is not a real HTTP status, so a +// read that lands there reached no answer worth classifying by code: treat it as unreachable. +const isHttpStatus = (status: number): boolean => status >= 100 && status <= 599; + +function classifyReadFailure(error: unknown): ServerAutomatedInboxReadFailure { + if (!(error instanceof AgentPortError)) return { reason: 'segment-read-failed' }; + + const { cause } = error; + + if (!(cause instanceof AgentHttpError)) { + const code = (cause as { code?: unknown })?.code; + + return typeof code === 'string' && UNREACHABLE_AGENT_ERROR_CODES.has(code) + ? { reason: 'agent-unreachable' } + : { reason: 'segment-read-failed' }; + } + + const httpStatus = cause.status; + + if (!isHttpStatus(httpStatus)) return { reason: 'agent-unreachable' }; + if (FORBIDDEN_AGENT_STATUSES.has(httpStatus)) return { reason: 'agent-forbidden', httpStatus }; + + if (UNREACHABLE_AGENT_STATUSES.has(httpStatus)) { + return { reason: 'agent-unreachable', httpStatus }; + } + + return { reason: 'segment-read-failed', httpStatus }; +} + export type AutomationPollerState = 'idle' | 'running' | 'draining' | 'stopped'; type PaddedPageReason = @@ -82,17 +131,12 @@ type PaddedPageReason = | 'field-without-not-in' | 'not-in-refused'; -/** - * `skipped` is a read that had nothing to ask and so never reached the agent. It is not a - * failure, and it is not proof the agent answers either — which is the distinction the sync - * decision rests on. - */ interface SegmentRead { - outcome: 'ok' | 'failed' | 'skipped'; items: T[]; paddedPageReason?: PaddedPageReason; requestedPageSize?: number; pagesRead?: number; + failure?: ServerAutomatedInboxReadFailure; } export interface AutomationPollerConfig { @@ -354,22 +398,16 @@ export default class AutomationPoller { ), ]); - // Reaching the agent at all is what the sync attests to. Reporting an empty poll when every - // read failed would tell the orchestrator this inbox is being swept while nothing is, which - // is the one thing a "no sync received" alert must never be lied to about. - if (closed.outcome !== 'ok' && candidates.outcome !== 'ok') { - this.logger('Error', 'Could not reach the agent, reporting nothing for this inbox', { - ...logContext, - }); - - return; - } + // The candidate read is the one that starts runs, so its failure is the one worth fixing. + const readFailure = candidates.failure ?? closed.failure; - // Sent even when both lists are empty: this call is what tells the orchestrator the inbox is - // still being polled, and an inbox with nothing to do must not read as an inbox nobody polls. + // Sent even when both lists are empty or both reads failed: the orchestrator keeps the last + // read failure until a sync comes without one, so a skipped sync would leave the settings + // panel with a stale warning, or none at all. const results = await this.config.automationPort.sync(config.inboxId, { closed: closed.items, candidates: candidates.items, + readFailure, }); this.logger('Info', 'Automated inbox polled', { @@ -380,6 +418,7 @@ export default class AutomationPoller { candidatePageSize: candidates.requestedPageSize, candidatePagesRead: candidates.pagesRead, paddedPageReason: candidates.paddedPageReason, + readFailure: readFailure?.reason, outcomes: results.reduce>( (counts, { outcome }) => ({ ...counts, [outcome]: (counts[outcome] ?? 0) + 1 }), {}, @@ -486,7 +525,7 @@ export default class AutomationPoller { return false; }); - if (readable.length === 0) return { outcome: 'skipped', items: [] }; + if (readable.length === 0) return { items: [] }; const stillInSegment = new Set(); @@ -502,7 +541,6 @@ export default class AutomationPoller { } return { - outcome: 'ok', items: readable.map(recordId => ({ recordId, stillInSegment: stillInSegment.has(recordId), @@ -528,7 +566,7 @@ export default class AutomationPoller { error: extractErrorMessage(error), }); - return { outcome: 'failed', items: [] }; + return { items: [], failure: classifyReadFailure(error) }; } } @@ -556,7 +594,6 @@ export default class AutomationPoller { // An agent that ignores an operator it does not know would hand known records back. return { - outcome: 'ok', items: page.filter(recordId => !knownSet.has(recordId)), requestedPageSize: config.maxConcurrentRuns, }; @@ -651,7 +688,6 @@ export default class AutomationPoller { } return { - outcome: 'ok', items: [...candidates], paddedPageReason, requestedPageSize, diff --git a/packages/workflow-executor/test/adapters/agent-client-segment-reader.test.ts b/packages/workflow-executor/test/adapters/agent-client-segment-reader.test.ts index c2a7778213..7ee61d8946 100644 --- a/packages/workflow-executor/test/adapters/agent-client-segment-reader.test.ts +++ b/packages/workflow-executor/test/adapters/agent-client-segment-reader.test.ts @@ -438,6 +438,25 @@ describe('AgentClientSegmentReader', () => { await expect(reader.listRecordIds(makeQuery())).rejects.toThrow(AgentPortError); }); + + it('should keep the HTTP status of the agent answer on the port error', async () => { + nock(AGENT_URL).get('/forest/orders').query(true).reply(403, {}); + + await expect(reader.listRecordIds(makeQuery())).rejects.toMatchObject({ + cause: expect.objectContaining({ name: 'AgentHttpError', status: 403 }), + }); + }); + + it('should keep the network error code when the agent does not answer', async () => { + nock(AGENT_URL) + .get('/forest/orders') + .query(true) + .replyWithError({ code: 'ECONNREFUSED', message: 'connect ECONNREFUSED' }); + + await expect(reader.listRecordIds(makeQuery())).rejects.toMatchObject({ + cause: expect.objectContaining({ code: 'ECONNREFUSED' }), + }); + }); }); describe('field operators', () => { diff --git a/packages/workflow-executor/test/automation-poller.test.ts b/packages/workflow-executor/test/automation-poller.test.ts index dce5e200ab..bab00a4783 100644 --- a/packages/workflow-executor/test/automation-poller.test.ts +++ b/packages/workflow-executor/test/automation-poller.test.ts @@ -8,11 +8,25 @@ import type { SegmentReaderPort, } from '../src/ports/segment-reader-port'; +import { AgentHttpError } from '@forestadmin/agent-client'; + import AutomationPoller from '../src/automation-poller'; -import { AutomatedInboxGoneError } from '../src/errors'; +import { + AgentPortError, + AutomatedInboxGoneError, + CompositeRecordIdMismatchError, + SegmentRecordIdMissingError, +} from '../src/errors'; const POLL_INTERVAL_S = 300; +function agentUnreachable(code: string): AgentPortError { + return new AgentPortError( + 'listSegmentRecordIds', + Object.assign(new Error(`agent unreachable: ${code}`), { code }), + ); +} + function makeConfig( overrides: Partial = {}, ): ServerAutomatedInboxConfig { @@ -783,7 +797,11 @@ describe('AutomationPoller', () => { await runOneCycle(makePoller(context)); - expect(context.automationPort.sync).not.toHaveBeenCalled(); + expect(context.automationPort.sync).toHaveBeenCalledWith('inbox-1', { + closed: [], + candidates: [], + readFailure: { reason: 'segment-read-failed' }, + }); expect(context.logger).toHaveBeenCalledWith( 'Error', 'Could not read new candidates of an automated inbox', @@ -1071,6 +1089,10 @@ describe('AutomationPoller', () => { 'Could not read new candidates of an automated inbox', expect.objectContaining({ error: 'agent down' }), ); + expect(context.automationPort.sync).toHaveBeenCalledWith( + 'inbox-1', + expect.objectContaining({ readFailure: { reason: 'segment-read-failed' } }), + ); }); it('should report the candidate read failed when a later page fails before any candidate', async () => { @@ -1512,7 +1534,7 @@ describe('AutomationPoller', () => { }); context.segmentReaderPort.listRecordIds.mockImplementation( async ({ recordIds }: ListSegmentRecordIdsQuery) => { - if (recordIds === undefined) throw new Error('segment page timed out'); + if (recordIds === undefined) throw agentUnreachable('ECONNABORTED'); return []; }, @@ -1525,6 +1547,7 @@ describe('AutomationPoller', () => { expect(context.automationPort.sync).toHaveBeenCalledWith('inbox-1', { closed: [{ recordId: 'treated', stillInSegment: false }], candidates: [], + readFailure: { reason: 'agent-unreachable' }, }); expect(context.logger).toHaveBeenCalledWith( 'Error', @@ -1550,36 +1573,165 @@ describe('AutomationPoller', () => { expect(context.automationPort.sync).toHaveBeenCalledWith('inbox-1', { closed: [], candidates: ['fresh'], + readFailure: { reason: 'segment-read-failed' }, }); }); - it('should report nothing when it could not reach the agent at all', async () => { + it('should still sync, with the failure, when it could not reach the agent at all', async () => { const context = makeContext({ assignments: [makeAssignment({ recordId: 'treated' })] }); - context.segmentReaderPort.listRecordIds.mockRejectedValue(new Error('agent unreachable')); + context.segmentReaderPort.listRecordIds.mockRejectedValue(agentUnreachable('ECONNREFUSED')); await runOneCycle(makePoller(context)); - // A sync here would tell the orchestrator this inbox is being swept while nothing is, which - // is exactly what a "no sync received" alert must never be lied to about. - expect(context.automationPort.sync).not.toHaveBeenCalled(); - expect(context.logger).toHaveBeenCalledWith( - 'Error', - 'Could not reach the agent, reporting nothing for this inbox', - expect.objectContaining({ inboxId: 'inbox-1' }), - ); + expect(context.automationPort.sync).toHaveBeenCalledWith('inbox-1', { + closed: [], + candidates: [], + readFailure: { reason: 'agent-unreachable' }, + }); }); - it('should still report when only the read that had nothing to ask was skipped', async () => { - // Nothing to reconcile, so that read never reaches the agent — but the candidate read did, - // and a successful poll must not be mistaken for an unreachable agent. - const context = makeContext(); - context.segmentReaderPort.listRecordIds.mockResolvedValue(['fresh']); + it('should send no read failure when every read succeeded', async () => { + const context = makeContext({ assignments: [makeAssignment({ recordId: 'treated' })] }); + context.segmentReaderPort.listRecordIds.mockResolvedValue([]); await runOneCycle(makePoller(context)); expect(context.automationPort.sync).toHaveBeenCalledWith('inbox-1', { - closed: [], - candidates: ['fresh'], + closed: [{ recordId: 'treated', stillInSegment: false }], + candidates: [], + }); + }); + + describe('read failure reported with the sync', () => { + const agentAnswered = (status: number) => + new AgentPortError('listSegmentRecordIds', new AgentHttpError(status, {}, '')); + + it.each([ + ['a 401', agentAnswered(401), { reason: 'agent-forbidden', httpStatus: 401 }], + ['a 403', agentAnswered(403), { reason: 'agent-forbidden', httpStatus: 403 }], + ['a 502', agentAnswered(502), { reason: 'agent-unreachable', httpStatus: 502 }], + ['a 503', agentAnswered(503), { reason: 'agent-unreachable', httpStatus: 503 }], + ['a 504', agentAnswered(504), { reason: 'agent-unreachable', httpStatus: 504 }], + ['a 500', agentAnswered(500), { reason: 'segment-read-failed', httpStatus: 500 }], + ['a 400', agentAnswered(400), { reason: 'segment-read-failed', httpStatus: 400 }], + ['a timeout', agentUnreachable('ECONNABORTED'), { reason: 'agent-unreachable' }], + ['a refused connection', agentUnreachable('ECONNREFUSED'), { reason: 'agent-unreachable' }], + ['an unknown host', agentUnreachable('ENOTFOUND'), { reason: 'agent-unreachable' }], + ['a reset connection', agentUnreachable('ECONNRESET'), { reason: 'agent-unreachable' }], + ['a temporary DNS failure', agentUnreachable('EAI_AGAIN'), { reason: 'agent-unreachable' }], + ['an unreachable host', agentUnreachable('EHOSTUNREACH'), { reason: 'agent-unreachable' }], + [ + 'an unreachable network', + agentUnreachable('ENETUNREACH'), + { reason: 'agent-unreachable' }, + ], + ['a broken pipe', agentUnreachable('EPIPE'), { reason: 'agent-unreachable' }], + ['a socket timeout', agentUnreachable('ETIMEDOUT'), { reason: 'agent-unreachable' }], + ['a response without a status', agentAnswered(0), { reason: 'agent-unreachable' }], + [ + 'an error of its own before any answer', + new AgentPortError( + 'listSegmentRecordIds', + new Error('secretOrPrivateKey must have a value'), + ), + { reason: 'segment-read-failed' }, + ], + [ + 'a record without an id', + new SegmentRecordIdMissingError('orders'), + { reason: 'segment-read-failed' }, + ], + [ + 'a record id it cannot split', + new CompositeRecordIdMismatchError('1', 2), + { reason: 'segment-read-failed' }, + ], + ])('should name the failure of a candidate read that got %s', async (_, error, expected) => { + const context = makeContext(); + context.segmentReaderPort.listRecordIds.mockRejectedValue(error); + + await runOneCycle(makePoller(context)); + + expect(context.automationPort.sync).toHaveBeenCalledWith('inbox-1', { + closed: [], + candidates: [], + readFailure: expected, + }); + }); + + it('should report the candidate read failure over the membership one', async () => { + const context = makeContext({ assignments: [makeAssignment({ recordId: 'treated' })] }); + context.segmentReaderPort.listRecordIds.mockImplementation( + async ({ recordIds }: ListSegmentRecordIdsQuery) => { + throw agentAnswered(recordIds === undefined ? 500 : 403); + }, + ); + + await runOneCycle(makePoller(context)); + + expect(context.automationPort.sync).toHaveBeenCalledWith( + 'inbox-1', + expect.objectContaining({ + readFailure: { reason: 'segment-read-failed', httpStatus: 500 }, + }), + ); + }); + + it('should report a membership read failure when the candidate read worked', async () => { + const context = makeContext({ assignments: [makeAssignment({ recordId: 'treated' })] }); + context.segmentReaderPort.listRecordIds.mockImplementation( + async ({ recordIds }: ListSegmentRecordIdsQuery) => { + if (recordIds !== undefined) throw agentAnswered(403); + + return []; + }, + ); + + await runOneCycle(makePoller(context)); + + expect(context.automationPort.sync).toHaveBeenCalledWith( + 'inbox-1', + expect.objectContaining({ + readFailure: { reason: 'agent-forbidden', httpStatus: 403 }, + }), + ); + }); + + it('should fail the whole membership read when a later chunk fails', async () => { + const recordIds = Array.from({ length: 51 }, (_, index) => `r${index}`); + const context = makeContext({ + assignments: recordIds.map(recordId => makeAssignment({ recordId })), + }); + context.segmentReaderPort.listRecordIds.mockImplementation( + async ({ recordIds: batch }: ListSegmentRecordIdsQuery) => { + if (batch === undefined) return []; + if (batch.includes('r50')) throw agentAnswered(503); + + return []; + }, + ); + + await runOneCycle(makePoller(context)); + + // All-or-nothing: the first chunk's results are dropped, and the read reports the failure. + expect(context.automationPort.sync).toHaveBeenCalledWith('inbox-1', { + closed: [], + candidates: [], + readFailure: { reason: 'agent-unreachable', httpStatus: 503 }, + }); + }); + + it('should name the failure in the poll log line', async () => { + const context = makeContext(); + context.segmentReaderPort.listRecordIds.mockRejectedValue(agentAnswered(403)); + + await runOneCycle(makePoller(context)); + + expect(context.logger).toHaveBeenCalledWith( + 'Info', + 'Automated inbox polled', + expect.objectContaining({ inboxId: 'inbox-1', readFailure: 'agent-forbidden' }), + ); }); });