diff --git a/packages/nodes-base/nodes/Databricks/test/pipelineUpdateEvents.test.ts b/packages/nodes-base/nodes/Databricks/test/pipelineUpdateEvents.test.ts new file mode 100644 index 00000000000..b5ffaddf135 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/test/pipelineUpdateEvents.test.ts @@ -0,0 +1,840 @@ +import type { + IDataObject, + INode, + IPollFunctions, + JsonObject, + NodeParameterValueType, +} from 'n8n-workflow'; +import { NodeApiError, NodeOperationError } from 'n8n-workflow'; +import { mock, mockDeep } from 'vitest-mock-extended'; + +import { DEFAULT_MAX_PAGES, PIPELINE_EVENTS_MAX_PAGE_SIZE, type PipelineEvent } from '../transport'; +import { + MANUAL_PAGE_SIZE, + MAX_IN_FLIGHT_UPDATES, + pollPipelineUpdateEvents, +} from '../trigger/pipelineUpdateEvents'; +import { OVERLAP_MS, toIso } from '../trigger/shared'; + +const HOST = 'https://adb-example.cloud.databricks.com'; +const PIPELINE_ID = '8199cd89-e2f5-4169-a6aa-656a24c8886d'; +const PIPELINE_NAME = 'n8n-spike-pipeline'; +const PIPELINE_URL = `${HOST}/pipelines/${PIPELINE_ID}`; +const UPDATE_ID = '01ee1dae-da54-415a-aba8-0c8b0de503f1'; +const FAILED_UPDATE_ID = '4518bfc6-f9d6-4a17-8038-1ad43f74c6da'; +const OTHER_UPDATE_ID = '7b2c9f10-3d4e-4f5a-8b6c-0d1e2f3a4b5c'; +const RUNNING_AT = '2026-09-01T14:20:31.066Z'; +const COMPLETED_AT = '2026-09-01T14:20:35.738Z'; +const FAILED_AT = '2026-09-01T14:21:11.671Z'; +const RUNNING_MS = Date.parse(RUNNING_AT); +const COMPLETED_MS = Date.parse(COMPLETED_AT); +const FAILED_MS = Date.parse(FAILED_AT); +const CURSOR_MS = RUNNING_MS - 1000; +const FLOOR_MS = CURSOR_MS - OVERLAP_MS; +const NOW = FAILED_MS + 10 * 60 * 1000; +const POLL_BUDGET_MS = 36_000; +const FAILURE_MESSAGE = + "Update 4518bf has failed. Failed to analyze flow 'workspace.n8n_spike.n8n_spike_table'."; +const TRACEBACK = + 'Traceback (most recent call last):\n File ".../n8n-spike-dlt-notebook", cell 1, line 7, in n8n_spike_table\n raise Exception("intentional pipeline failure (n8n spike)") ...'; +const ALL_EVENTS = ['updateCompleted', 'updateFailed', 'updateStarted']; +const IN_FLIGHT_STATES = [ + 'QUEUED', + 'CREATED', + 'WAITING_FOR_RESOURCES', + 'INITIALIZING', + 'RESETTING', + 'SETTING_UP_TABLES', + 'RUNNING', + 'STOPPING', +]; +const LEVELS_FILTER = "level in ('INFO', 'WARN', 'ERROR')"; + +const node = mock({ name: 'Databricks Trigger', typeVersion: 1 }); + +const progressEvent = ( + state: string, + timestamp: string, + updateId = UPDATE_ID, + overrides: Partial = {}, +): PipelineEvent => ({ + id: `${updateId}:${state}`, + event_type: 'update_progress', + level: 'INFO', + message: `Update ${updateId.slice(0, 6)} is ${state}.`, + timestamp, + origin: { pipeline_id: PIPELINE_ID, pipeline_name: PIPELINE_NAME, update_id: updateId }, + details: { update_progress: { state } }, + ...overrides, +}); + +const runningEvent = progressEvent('RUNNING', RUNNING_AT); +const completedEvent = progressEvent('COMPLETED', COMPLETED_AT); +const failedEvent = progressEvent('FAILED', FAILED_AT, FAILED_UPDATE_ID, { + level: 'ERROR', + message: FAILURE_MESSAGE, + error: { + fatal: true, + exceptions: [ + { + class_name: 'Exception', + error_class: 'PYTHON.EXCEPTION', + sql_state: 'P0001', + message: TRACEBACK, + stack: [ + { + declaring_class: 'n8n_spike_table', + method_name: 'n8n_spike_table', + file_name: 'n8n-spike-dlt-notebook', + line_number: 7, + }, + ], + }, + ], + }, +}); +const flowEvent = (timestamp: string): PipelineEvent => ({ + id: `flow:${timestamp}`, + event_type: 'flow_progress', + level: 'INFO', + message: "Flow 'n8n_spike_table' is RUNNING.", + timestamp, + origin: { + pipeline_id: PIPELINE_ID, + pipeline_name: PIPELINE_NAME, + update_id: UPDATE_ID, + flow_name: 'n8n_spike_table', + }, + details: { flow_progress: { status: 'RUNNING' } }, +}); +const mixedPage = [ + runningEvent, + progressEvent('RUNNING', toIso(RUNNING_MS + 500), FAILED_UPDATE_ID), + completedEvent, + failedEvent, +]; + +const simplifiedItem = (event: string, body: IDataObject, updateId = UPDATE_ID) => ({ + json: { + event, + pipeline: { id: PIPELINE_ID, name: PIPELINE_NAME, url: PIPELINE_URL }, + update: { id: updateId, url: `${PIPELINE_URL}/updates/${updateId}` }, + ...body, + }, +}); +const startedItem = simplifiedItem('updateStarted', { + timing: { startedAt: RUNNING_AT, runningAt: RUNNING_AT }, +}); +const completedResult = { state: 'COMPLETED', message: 'Update 01ee1d is COMPLETED.' }; +const completedItem = simplifiedItem('updateCompleted', { + result: completedResult, + timing: { + startedAt: RUNNING_AT, + runningAt: RUNNING_AT, + endedAt: COMPLETED_AT, + durationMs: COMPLETED_MS - RUNNING_MS, + }, +}); +const failedResult = { + state: 'FAILED', + message: FAILURE_MESSAGE, + errors: [ + { + type: 'Exception', + code: 'PYTHON.EXCEPTION', + sqlState: 'P0001', + message: TRACEBACK, + stack: [ + { + class: 'n8n_spike_table', + method: 'n8n_spike_table', + file: 'n8n-spike-dlt-notebook', + line: 7, + }, + ], + }, + ], +}; +const failedItem = simplifiedItem( + 'updateFailed', + { result: failedResult, timing: { endedAt: FAILED_AT } }, + FAILED_UPDATE_ID, +); + +const emitted = (event: string, updateId = UPDATE_ID) => ({ + event, + update: expect.objectContaining({ id: updateId }), +}); + +const tracked = (startedMs?: number, endedMs?: number, runningMs?: number) => ({ + startedMs, + runningMs, + endedMs, +}); +const running = (startedMs: number, endedMs?: number) => tracked(startedMs, endedMs, startedMs); + +const watchingState = (updates: IDataObject = {}, cursorMs = CURSOR_MS): IDataObject => ({ + pipelineId: PIPELINE_ID, + cursorMs, + floorMs: cursorMs - OVERLAP_MS, + updates, +}); + +const freshState = (pipelineId = PIPELINE_ID): IDataObject => ({ + pipelineId, + cursorMs: NOW, + floorMs: NOW, + updates: {}, +}); + +const apiErrorFromBody = (status: number, data: JsonObject) => + new NodeApiError(node, { + message: `Request failed with status code ${status}`, + response: { status, data }, + }); + +type ContextOptions = { + events?: NodeParameterValueType; + simplify?: boolean; + pipelineId?: string; + mode?: 'trigger' | 'manual'; + pollBudgetMs?: number; + staticData?: IDataObject; +}; + +const createContext = (options: ContextOptions = {}) => { + const context = mockDeep(); + const staticData = options.staticData ?? {}; + context.getNode.mockReturnValue(node); + context.getMode.mockReturnValue(options.mode ?? 'trigger'); + context.getPollBudgetMs.mockReturnValue(options.pollBudgetMs ?? POLL_BUDGET_MS); + context.getWorkflowStaticData.mockReturnValue(staticData); + context.getCredentials.mockResolvedValue({ host: HOST }); + context.getNodeParameter.mockImplementation((name, fallback) => { + switch (name) { + case 'authentication': + return 'accessToken'; + case 'events': + return options.events ?? ['updateCompleted', 'updateFailed']; + case 'simplify': + return options.simplify ?? true; + case 'pipelineId': + return options.pipelineId ?? PIPELINE_ID; + default: + return fallback; + } + }); + const api = context.helpers.httpRequestWithAuthentication; + const requestQuery = (call = 0) => api.mock.calls[call][1].qs; + const poll = async () => await pollPipelineUpdateEvents.call(context); + const pollWith = async (events: unknown[], nextPageToken?: string) => { + api.mockResolvedValueOnce({ events, next_page_token: nextPageToken }); + return await poll(); + }; + const eventsWith = async (events: unknown[], nextPageToken?: string) => + ((await pollWith(events, nextPageToken)) ?? [[]])[0].map((item) => ({ + event: item.json.event, + update: item.json.update, + })); + return { context, staticData, api, requestQuery, poll, pollWith, eventsWith }; +}; + +describe('pollPipelineUpdateEvents', () => { + beforeEach(() => { + vi.spyOn(Date, 'now').mockReturnValue(NOW); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + describe('first poll', () => { + it('starts watching from now without calling the API', async () => { + const { staticData, api, poll } = createContext(); + + await expect(poll()).resolves.toBeNull(); + + expect(staticData).toEqual(freshState()); + expect(api).not.toHaveBeenCalled(); + }); + + it.each([ + [ + 'state of another pipeline', + { ...watchingState({ [OTHER_UPDATE_ID]: tracked(5) }, 5), pipelineId: FAILED_UPDATE_ID }, + ], + [ + 'state of another pipeline with a leftover key', + { ...watchingState(), pipelineId: FAILED_UPDATE_ID, seen: 3 }, + ], + [ + 'state with a broken shape', + { pipelineId: PIPELINE_ID, cursorMs: 'yesterday', updates: [] }, + ], + ['state without a floor', { pipelineId: PIPELINE_ID, cursorMs: CURSOR_MS, updates: {} }], + ['state with a broken update entry', watchingState({ [UPDATE_ID]: { startedMs: 'soon' } })], + ])('resets %s in place', async (_label, stale) => { + const staticData: IDataObject = { ...stale }; + const { api, poll } = createContext({ staticData }); + + await expect(poll()).resolves.toBeNull(); + + expect(staticData).toEqual(freshState()); + expect(api).not.toHaveBeenCalled(); + }); + + it('stores the pipeline ID in lower case', async () => { + const { staticData, poll } = createContext({ pipelineId: PIPELINE_ID.toUpperCase() }); + + await expect(poll()).resolves.toBeNull(); + + expect(staticData).toEqual(freshState()); + }); + + it('does not emit an event from before the activation time', async () => { + const { staticData, requestQuery, poll, pollWith } = createContext({ events: ALL_EVENTS }); + + await expect(poll()).resolves.toBeNull(); + await expect(pollWith([progressEvent('RUNNING', toIso(NOW - 1000))])).resolves.toBeNull(); + + expect(requestQuery()).toMatchObject({ + filter: `${LEVELS_FILTER} AND timestamp > '${toIso(NOW)}'`, + }); + expect(staticData).toEqual(freshState()); + }); + + it('keeps an event whose sub-millisecond timestamp falls right after the floor', async () => { + const justAfterNow = toIso(NOW).replace('Z', '4Z'); + const { staticData, eventsWith, poll } = createContext({ events: ALL_EVENTS }); + + await expect(poll()).resolves.toBeNull(); + await expect(eventsWith([progressEvent('RUNNING', justAfterNow)])).resolves.toEqual([ + emitted('updateStarted'), + ]); + expect(staticData).toEqual({ ...freshState(), updates: { [UPDATE_ID]: running(NOW) } }); + }); + }); + + describe('classification', () => { + it('classifies a completed update whose start it did not see', async () => { + const { staticData, pollWith } = createContext({ staticData: watchingState() }); + + await expect(pollWith([completedEvent])).resolves.toEqual([ + [ + simplifiedItem('updateCompleted', { + result: completedResult, + timing: { endedAt: COMPLETED_AT }, + }), + ], + ]); + expect(staticData).toEqual( + watchingState({ [UPDATE_ID]: tracked(undefined, COMPLETED_MS) }, COMPLETED_MS), + ); + }); + + it('classifies a failed update with its message and traceback', async () => { + const { staticData, pollWith } = createContext({ staticData: watchingState() }); + + await expect(pollWith([failedEvent])).resolves.toEqual([[failedItem]]); + expect(staticData).toEqual( + watchingState({ [FAILED_UPDATE_ID]: tracked(undefined, FAILED_MS) }, FAILED_MS), + ); + }); + + it('classifies a cancelled update as failed', async () => { + const { pollWith } = createContext({ staticData: watchingState() }); + + await expect(pollWith([progressEvent('CANCELED', FAILED_AT)])).resolves.toEqual([ + [ + simplifiedItem('updateFailed', { + result: { state: 'CANCELED', message: 'Update 01ee1d is CANCELED.' }, + timing: { endedAt: FAILED_AT }, + }), + ], + ]); + }); + + it.each([...IN_FLIGHT_STATES, 'SOME_FUTURE_STATE'])( + 'treats %s as a started update', + async (state) => { + const runningMs = state === 'RUNNING' ? RUNNING_MS : undefined; + const { staticData, pollWith } = createContext({ + events: ALL_EVENTS, + staticData: watchingState(), + }); + + await expect(pollWith([progressEvent(state, RUNNING_AT)])).resolves.toEqual([ + [ + simplifiedItem('updateStarted', { + timing: { + startedAt: RUNNING_AT, + runningAt: runningMs === undefined ? undefined : RUNNING_AT, + }, + }), + ], + ]); + expect(staticData).toEqual( + watchingState({ [UPDATE_ID]: tracked(RUNNING_MS, undefined, runningMs) }, RUNNING_MS), + ); + }, + ); + + it.each<[string, PipelineEvent['error'], IDataObject]>([ + [ + 'no exceptions', + { fatal: true, exceptions: [] }, + { state: 'FAILED', message: FAILURE_MESSAGE }, + ], + ['no error block', undefined, { state: 'FAILED', message: FAILURE_MESSAGE }], + [ + 'an exception with only a message', + { exceptions: [{ message: 'boom' }] }, + { state: 'FAILED', message: FAILURE_MESSAGE, errors: [{ message: 'boom' }] }, + ], + [ + 'a frame with only a line number', + { exceptions: [{ message: 'boom', stack: [{ line_number: 3 }] }] }, + { + state: 'FAILED', + message: FAILURE_MESSAGE, + errors: [{ message: 'boom', stack: [{ line: 3 }] }], + }, + ], + [ + 'an empty stack', + { exceptions: [{ message: 'boom', stack: [] }] }, + { state: 'FAILED', message: FAILURE_MESSAGE, errors: [{ message: 'boom' }] }, + ], + ])('simplifies the result of a failure with %s', async (_label, error, result) => { + const { pollWith } = createContext({ staticData: watchingState() }); + const event = progressEvent('FAILED', FAILED_AT, FAILED_UPDATE_ID, { + level: 'ERROR', + message: FAILURE_MESSAGE, + error, + }); + + await expect(pollWith([event])).resolves.toEqual([ + [ + simplifiedItem( + 'updateFailed', + { result, timing: { endedAt: FAILED_AT } }, + FAILED_UPDATE_ID, + ), + ], + ]); + }); + + it('omits the pipeline name when the event has none', async () => { + const { pollWith } = createContext({ staticData: watchingState() }); + const event = progressEvent('COMPLETED', COMPLETED_AT, UPDATE_ID, { + origin: { pipeline_id: PIPELINE_ID, update_id: UPDATE_ID }, + }); + + const output = await pollWith([event]); + + expect(output?.[0][0].json.pipeline).toEqual({ id: PIPELINE_ID, url: PIPELINE_URL }); + }); + + it('returns the raw event behind the event label when simplify is off', async () => { + const { pollWith } = createContext({ simplify: false, staticData: watchingState() }); + + await expect(pollWith([failedEvent])).resolves.toEqual([ + [{ json: { event: 'updateFailed', ...failedEvent } }], + ]); + }); + + it('skips events that are not update progress and still advances the cursor', async () => { + const { staticData, pollWith } = createContext({ + events: ALL_EVENTS, + staticData: watchingState(), + }); + + await expect( + pollWith([ + flowEvent(RUNNING_AT), + { ...runningEvent, origin: { pipeline_id: PIPELINE_ID } }, + { ...runningEvent, origin: { ...runningEvent.origin, update_id: '__proto__' } }, + { ...runningEvent, origin: { ...runningEvent.origin, update_id: 'update-1' } }, + { ...runningEvent, details: {} }, + { ...runningEvent, details: { update_progress: { state: 7 } } }, + { ...runningEvent, timestamp: undefined }, + { ...runningEvent, timestamp: 'yesterday' }, + ]), + ).resolves.toBeNull(); + + expect(staticData).toEqual(watchingState({}, RUNNING_MS)); + expect(Object.prototype).not.toHaveProperty('startedMs'); + }); + + it('rejects a page with a null entry in the events and keeps the state', async () => { + const { staticData, api, poll } = createContext({ staticData: watchingState() }); + api.mockResolvedValueOnce({ events: [null, runningEvent] }); + + await expect(poll()).rejects.toThrow( + 'Databricks did not return a JSON list of pipeline events', + ); + expect(staticData).toEqual(watchingState()); + }); + }); + + describe('update lifecycle across polls', () => { + it('tracks a started update without emitting when updateStarted is not subscribed', async () => { + const { staticData, pollWith } = createContext({ staticData: watchingState() }); + + await expect(pollWith([runningEvent])).resolves.toBeNull(); + expect(staticData).toEqual(watchingState({ [UPDATE_ID]: running(RUNNING_MS) }, RUNNING_MS)); + }); + + it('emits the start, then the completion with its duration, then nothing on a repeat', async () => { + const { staticData, pollWith, eventsWith } = createContext({ + events: ALL_EVENTS, + staticData: watchingState(), + }); + + await expect(eventsWith([runningEvent])).resolves.toEqual([emitted('updateStarted')]); + + await expect(pollWith([completedEvent])).resolves.toEqual([[completedItem]]); + expect(staticData).toEqual( + watchingState({ [UPDATE_ID]: running(RUNNING_MS, COMPLETED_MS) }, COMPLETED_MS), + ); + + await expect(pollWith([runningEvent, completedEvent])).resolves.toBeNull(); + }); + + it('emits the start and the end of an update that ran between polls', async () => { + const { pollWith } = createContext({ events: ALL_EVENTS, staticData: watchingState() }); + + await expect(pollWith([runningEvent, completedEvent])).resolves.toEqual([ + [startedItem, completedItem], + ]); + }); + + it('orders the events by timestamp when the API returns them out of order', async () => { + const { eventsWith } = createContext({ events: ALL_EVENTS, staticData: watchingState() }); + + await expect(eventsWith([completedEvent, runningEvent])).resolves.toEqual([ + emitted('updateStarted'), + emitted('updateCompleted'), + ]); + }); + + it('ignores a late start event of an update already reported as finished', async () => { + const { staticData, eventsWith } = createContext({ + events: ALL_EVENTS, + staticData: watchingState(), + }); + + await expect(eventsWith([completedEvent])).resolves.toEqual([emitted('updateCompleted')]); + await expect(eventsWith([runningEvent])).resolves.toEqual([]); + expect(staticData).toEqual( + watchingState({ [UPDATE_ID]: tracked(undefined, COMPLETED_MS) }, COMPLETED_MS), + ); + }); + + it('emits the start once, at the first in-flight event of an update', async () => { + const queuedAt = toIso(RUNNING_MS - 30_000); + const { staticData, pollWith } = createContext({ + events: ALL_EVENTS, + staticData: watchingState(), + }); + + await expect( + pollWith([progressEvent('WAITING_FOR_RESOURCES', queuedAt), runningEvent]), + ).resolves.toEqual([[simplifiedItem('updateStarted', { timing: { startedAt: queuedAt } })]]); + expect(staticData).toEqual( + watchingState( + { [UPDATE_ID]: tracked(RUNNING_MS - 30_000, undefined, RUNNING_MS) }, + RUNNING_MS, + ), + ); + }); + + it('measures the duration from the RUNNING event and keeps the earlier start', async () => { + const queuedAt = toIso(RUNNING_MS - 30_000); + const { pollWith } = createContext({ staticData: watchingState() }); + + await expect( + pollWith([progressEvent('WAITING_FOR_RESOURCES', queuedAt), runningEvent]), + ).resolves.toBeNull(); + await expect(pollWith([completedEvent])).resolves.toEqual([ + [ + simplifiedItem('updateCompleted', { + result: completedResult, + timing: { + startedAt: queuedAt, + runningAt: RUNNING_AT, + endedAt: COMPLETED_AT, + durationMs: COMPLETED_MS - RUNNING_MS, + }, + }), + ], + ]); + }); + + it('reports no duration when it never saw the RUNNING event', async () => { + const queuedMs = RUNNING_MS - 30_000; + const { pollWith } = createContext({ + staticData: watchingState({ [UPDATE_ID]: tracked(queuedMs) }), + }); + + await expect(pollWith([completedEvent])).resolves.toEqual([ + [ + simplifiedItem('updateCompleted', { + result: completedResult, + timing: { startedAt: toIso(queuedMs), endedAt: COMPLETED_AT }, + }), + ], + ]); + }); + + it('follows two interleaved updates in time order', async () => { + const { eventsWith } = createContext({ events: ALL_EVENTS, staticData: watchingState() }); + + await expect(eventsWith(mixedPage)).resolves.toEqual([ + emitted('updateStarted'), + emitted('updateStarted', FAILED_UPDATE_ID), + emitted('updateCompleted'), + emitted('updateFailed', FAILED_UPDATE_ID), + ]); + }); + + it.each([ + ['updateStarted', [emitted('updateStarted'), emitted('updateStarted', FAILED_UPDATE_ID)]], + ['updateCompleted', [emitted('updateCompleted')]], + ['updateFailed', [emitted('updateFailed', FAILED_UPDATE_ID)]], + ])('emits only %s when it is the single subscribed event', async (event, expected) => { + const { eventsWith } = createContext({ events: [event], staticData: watchingState() }); + + await expect(eventsWith(mixedPage)).resolves.toEqual(expected); + }); + }); + + describe('cursor', () => { + it('lists the events after the floor at the INFO, WARN and ERROR levels', async () => { + const { context, api, requestQuery, pollWith } = createContext({ + staticData: watchingState(), + }); + + await pollWith([completedEvent]); + + expect(context.getNodeParameter).toHaveBeenCalledWith('pipelineId', '', { + extractValue: true, + }); + expect(api).toHaveBeenCalledTimes(1); + expect(requestQuery()).toEqual({ + max_results: PIPELINE_EVENTS_MAX_PAGE_SIZE, + filter: `${LEVELS_FILTER} AND timestamp > '${toIso(FLOOR_MS)}'`, + order_by: 'timestamp asc', + }); + }); + + it('advances the cursor to the newest fetched event, including events it does not report', async () => { + const laterMs = COMPLETED_MS + 60_000; + const { staticData, pollWith } = createContext({ + events: ALL_EVENTS, + staticData: watchingState(), + }); + + await expect(pollWith([runningEvent, flowEvent(toIso(laterMs))])).resolves.toHaveLength(1); + + expect(staticData).toEqual(watchingState({ [UPDATE_ID]: running(RUNNING_MS) }, laterMs)); + }); + + it('keeps the cursor when the poll lists no events', async () => { + const { staticData, pollWith } = createContext({ staticData: watchingState() }); + + await expect(pollWith([])).resolves.toBeNull(); + expect(staticData).toEqual(watchingState()); + }); + + it('drops finished updates the window can no longer list and keeps unfinished ones', async () => { + const t = CURSOR_MS + OVERLAP_MS + 60_000; + const oldStart = CURSOR_MS - 10 * OVERLAP_MS; + const { staticData, pollWith } = createContext({ + staticData: watchingState({ + [FAILED_UPDATE_ID]: tracked(CURSOR_MS - 5000, CURSOR_MS), + [OTHER_UPDATE_ID]: tracked(oldStart), + }), + }); + + await expect(pollWith([progressEvent('RUNNING', toIso(t))])).resolves.toBeNull(); + + expect(staticData).toEqual( + watchingState({ [OTHER_UPDATE_ID]: tracked(oldStart), [UPDATE_ID]: running(t) }, t), + ); + }); + + it('keeps a finished update whose end is still inside the window', async () => { + const { staticData, pollWith } = createContext({ + staticData: watchingState({ [FAILED_UPDATE_ID]: tracked(undefined, CURSOR_MS) }), + }); + + await expect(pollWith([runningEvent])).resolves.toBeNull(); + + expect(staticData).toEqual( + watchingState( + { [FAILED_UPDATE_ID]: tracked(undefined, CURSOR_MS), [UPDATE_ID]: running(RUNNING_MS) }, + RUNNING_MS, + ), + ); + }); + + it('caps the tracked unfinished updates and drops the oldest first', async () => { + const updates = Object.fromEntries( + Array.from({ length: MAX_IN_FLIGHT_UPDATES }, (_, index) => [ + `00000000-0000-4000-8000-${String(index).padStart(12, '0')}`, + tracked(CURSOR_MS - MAX_IN_FLIGHT_UPDATES + index), + ]), + ); + const kept = Object.fromEntries(Object.entries(updates).slice(1)); + const { staticData, pollWith } = createContext({ staticData: watchingState(updates) }); + + await expect(pollWith([runningEvent])).resolves.toBeNull(); + + expect(staticData).toEqual( + watchingState({ ...kept, [UPDATE_ID]: running(RUNNING_MS) }, RUNNING_MS), + ); + }); + + describe('truncated listing', () => { + it('reports what it fetched, continues from there on the next poll and logs it once', async () => { + const { context, staticData, api, requestQuery, poll, pollWith } = createContext({ + events: ALL_EVENTS, + staticData: watchingState(), + }); + api.mockResolvedValue({ events: [runningEvent], next_page_token: 'more' }); + + await expect(poll()).resolves.toEqual([[startedItem]]); + + expect(api).toHaveBeenCalledTimes(DEFAULT_MAX_PAGES); + expect(staticData).toEqual(watchingState({ [UPDATE_ID]: running(RUNNING_MS) }, RUNNING_MS)); + expect(context.logger.info).toHaveBeenCalledTimes(1); + expect(context.logger.info).toHaveBeenCalledWith( + expect.stringContaining(`pipeline ${PIPELINE_ID} since ${toIso(FLOOR_MS)}`), + ); + expect(context.logger.info).toHaveBeenCalledWith( + expect.stringContaining(`continues from ${RUNNING_AT}`), + ); + expect(context.logger.warn).not.toHaveBeenCalled(); + + await expect(pollWith([completedEvent])).resolves.toEqual([[completedItem]]); + expect(requestQuery(DEFAULT_MAX_PAGES)).toMatchObject({ + filter: `${LEVELS_FILTER} AND timestamp > '${toIso(RUNNING_MS - OVERLAP_MS)}'`, + }); + expect(api).toHaveBeenCalledTimes(DEFAULT_MAX_PAGES + 1); + expect(context.logger.info).toHaveBeenCalledTimes(1); + }); + + it('stops after one page when the poll budget is already spent', async () => { + const { context, staticData, api, poll } = createContext({ + events: ALL_EVENTS, + pollBudgetMs: 0, + staticData: watchingState(), + }); + api.mockResolvedValueOnce({ events: [runningEvent], next_page_token: 'more' }); + + await expect(poll()).resolves.toEqual([[startedItem]]); + + expect(api).toHaveBeenCalledTimes(1); + expect(staticData).toEqual(watchingState({ [UPDATE_ID]: running(RUNNING_MS) }, RUNNING_MS)); + expect(context.logger.info).toHaveBeenCalledTimes(1); + }); + }); + }); + + describe('manual mode', () => { + it('lists the latest page, reports it in time order and leaves the static data alone', async () => { + const { context, staticData, api, requestQuery, eventsWith } = createContext({ + mode: 'manual', + events: ALL_EVENTS, + }); + + await expect( + eventsWith([failedEvent, completedEvent, runningEvent], 'ignored'), + ).resolves.toEqual([ + emitted('updateStarted'), + emitted('updateCompleted'), + emitted('updateFailed', FAILED_UPDATE_ID), + ]); + + expect(api).toHaveBeenCalledTimes(1); + expect(requestQuery()).toEqual({ + max_results: MANUAL_PAGE_SIZE, + filter: LEVELS_FILTER, + order_by: 'timestamp desc', + }); + expect(context.getWorkflowStaticData).not.toHaveBeenCalled(); + expect(staticData).toEqual({}); + }); + + it('computes the duration when the page holds both ends of an update', async () => { + const { pollWith } = createContext({ mode: 'manual', events: ['updateCompleted'] }); + + await expect(pollWith([completedEvent, runningEvent])).resolves.toEqual([[completedItem]]); + }); + + it('returns null when the page has no update events', async () => { + const { pollWith } = createContext({ mode: 'manual' }); + + await expect(pollWith([flowEvent(RUNNING_AT)])).resolves.toBeNull(); + }); + }); + + describe('errors', () => { + it('explains a PERMISSION_DENIED error with the Can View hint', async () => { + const { api, poll } = createContext({ staticData: watchingState() }); + api.mockRejectedValue( + apiErrorFromBody(403, { + error_code: 'PERMISSION_DENIED', + message: `User does not have Can View permission on pipeline ${PIPELINE_ID}.`, + }), + ); + + const error = await poll().catch((thrown: unknown) => thrown); + + expect(error).toBeInstanceOf(NodeApiError); + expect(error).toMatchObject({ + message: `User does not have Can View permission on pipeline ${PIPELINE_ID}.`, + description: + 'Grant Can View on the pipeline to the user or service principal of the credential, then retry.', + }); + }); + + it('rethrows other API errors unchanged', async () => { + const { api, poll } = createContext({ staticData: watchingState() }); + const apiError = apiErrorFromBody(500, { error_code: 'INTERNAL_ERROR', message: 'boom' }); + api.mockRejectedValue(apiError); + + await expect(poll()).rejects.toBe(apiError); + }); + + it.each([ + ['an empty pipeline ID', ''], + ['a pipeline ID without dashes', '8199cd89e2f54169a6aa656a24c8886d'], + ['a path', '../updates'], + ['a job ID', '281874479417551'], + ])('rejects %s before any request', async (_label, pipelineId) => { + const { staticData, api, poll } = createContext({ + pipelineId, + staticData: watchingState(), + }); + + const error = await poll().catch((thrown: unknown) => thrown); + + expect(error).toBeInstanceOf(NodeOperationError); + expect(error).toMatchObject({ message: 'Pipeline ID must be a UUID' }); + expect(api).not.toHaveBeenCalled(); + expect(staticData).toEqual(watchingState()); + }); + + it.each([ + ['a string', 'updateFailed'], + ['an unknown event', ['updateFailed', 'updateCancelled']], + ])('rejects %s as the events parameter', async (_label, events) => { + const { api, poll } = createContext({ events, staticData: watchingState() }); + + await expect(poll()).rejects.toThrow('Events must be a list of update events'); + expect(api).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/nodes-base/nodes/Databricks/trigger/pipelineUpdateEvents.ts b/packages/nodes-base/nodes/Databricks/trigger/pipelineUpdateEvents.ts new file mode 100644 index 00000000000..1d9822f162c --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/trigger/pipelineUpdateEvents.ts @@ -0,0 +1,323 @@ +import { isRecord } from '@n8n/utils/is-record'; +import { NodeOperationError } from 'n8n-workflow'; +import type { IDataObject, INodeExecutionData, IPollFunctions } from 'n8n-workflow'; + +import { getActiveCredentialType, getHost } from '../actions/helpers'; +import { + listAllPipelineEvents, + listPipelineEvents, + PIPELINE_EVENTS_MAX_PAGE_SIZE, + UUID_PATTERN, + type PipelineEvent, + type PipelineEventLevel, +} from '../transport'; +import { + OVERLAP_MS, + readEvents, + toIso, + toOutput, + withLegibleErrors, + withoutUndefined, +} from './shared'; + +export const PIPELINE_UPDATE_EVENTS = ['updateCompleted', 'updateFailed', 'updateStarted'] as const; +export const MANUAL_PAGE_SIZE = 100; +export const MAX_IN_FLIGHT_UPDATES = 100; + +const WATCHED_LEVELS: readonly PipelineEventLevel[] = ['INFO', 'WARN', 'ERROR']; +const PERMISSION_HINT = + 'Grant Can View on the pipeline to the user or service principal of the credential, then retry.'; + +type PipelineUpdateEvent = (typeof PIPELINE_UPDATE_EVENTS)[number]; + +type PipelineException = NonNullable['exceptions']>[number]; + +type StackFrame = NonNullable[number]; + +type UpdateProgress = { raw: PipelineEvent; updateId: string; state: string; timestampMs: number }; + +type TrackedUpdate = { startedMs?: number; runningMs?: number; endedMs?: number }; + +type TrackedUpdates = Record; + +type PipelineUpdateWatchState = { + pipelineId: string; + cursorMs: number; + floorMs: number; + updates: TrackedUpdates; +}; + +type ItemOptions = { + subscribed: PipelineUpdateEvent[]; + simplify: boolean; + pipelineId: string; + host: string; +}; + +function isUuid(value: unknown): value is string { + return typeof value === 'string' && UUID_PATTERN.test(value); +} + +function isOptionalNumber(value: unknown): value is number | undefined { + return value === undefined || typeof value === 'number'; +} + +function isTrackedUpdate(value: unknown): value is TrackedUpdate { + return ( + isRecord(value) && + isOptionalNumber(value.startedMs) && + isOptionalNumber(value.runningMs) && + isOptionalNumber(value.endedMs) + ); +} + +function isPipelineUpdateWatchState(value: unknown): value is PipelineUpdateWatchState { + return ( + isRecord(value) && + typeof value.pipelineId === 'string' && + typeof value.cursorMs === 'number' && + typeof value.floorMs === 'number' && + isRecord(value.updates) && + Object.values(value.updates).every(isTrackedUpdate) + ); +} + +function readPipelineId(context: IPollFunctions): string { + const pipelineId = String(context.getNodeParameter('pipelineId', '', { extractValue: true })); + if (!isUuid(pipelineId)) { + throw new NodeOperationError(context.getNode(), 'Pipeline ID must be a UUID', { + description: + 'Use the ID shown in the pipeline URL in Databricks, for example 8199cd89-e2f5-4169-a6aa-656a24c8886d.', + }); + } + return pipelineId.toLowerCase(); +} + +function parseTimestamp(value: string | undefined): number | undefined { + if (value === undefined) return undefined; + const ms = Date.parse(value); + return Number.isNaN(ms) ? undefined : ms; +} + +function toUpdateProgress(event: PipelineEvent): UpdateProgress | undefined { + const updateId = event.origin?.update_id; + const state = event.details?.update_progress?.state; + const timestampMs = parseTimestamp(event.timestamp); + if ( + event.event_type !== 'update_progress' || + !isUuid(updateId) || + typeof state !== 'string' || + timestampMs === undefined + ) { + return undefined; + } + return { raw: event, updateId, state, timestampMs }; +} + +function toChronological(events: PipelineEvent[]): UpdateProgress[] { + return events + .flatMap((event) => toUpdateProgress(event) ?? []) + .sort((a, b) => a.timestampMs - b.timestampMs); +} + +function terminalEventOf(state: string): PipelineUpdateEvent | undefined { + if (state === 'COMPLETED') return 'updateCompleted'; + if (state === 'FAILED' || state === 'CANCELED') return 'updateFailed'; + return undefined; +} + +function collectNewEvents( + updates: TrackedUpdates, + progress: UpdateProgress, +): PipelineUpdateEvent[] { + let entry = updates[progress.updateId]; + if (entry === undefined) { + entry = {}; + updates[progress.updateId] = entry; + } + if (entry.endedMs !== undefined) return []; + const terminal = terminalEventOf(progress.state); + if (terminal !== undefined) { + entry.endedMs = progress.timestampMs; + return [terminal]; + } + if (progress.state === 'RUNNING' && entry.runningMs === undefined) { + entry.runningMs = progress.timestampMs; + } + if (entry.startedMs !== undefined) return []; + entry.startedMs = progress.timestampMs; + return ['updateStarted']; +} + +function simplifyFrame(frame: StackFrame): IDataObject { + return withoutUndefined({ + class: frame.declaring_class, + method: frame.method_name, + file: frame.file_name, + line: frame.line_number, + }); +} + +function simplifyException(exception: PipelineException): IDataObject { + const stack = exception.stack ?? []; + return withoutUndefined({ + type: exception.class_name, + code: exception.error_class, + sqlState: exception.sql_state, + message: exception.message, + stack: stack.length > 0 ? stack.map(simplifyFrame) : undefined, + }); +} + +function simplifyResult(progress: UpdateProgress): IDataObject { + const exceptions = progress.raw.error?.exceptions ?? []; + return withoutUndefined({ + state: progress.state, + message: progress.raw.message, + errors: exceptions.length > 0 ? exceptions.map(simplifyException) : undefined, + }); +} + +function simplifyTiming( + event: PipelineUpdateEvent, + progress: UpdateProgress, + entry: TrackedUpdate, +): IDataObject { + const { startedMs, runningMs } = entry; + const started = withoutUndefined({ + startedAt: startedMs === undefined ? undefined : toIso(startedMs), + runningAt: runningMs === undefined ? undefined : toIso(runningMs), + }); + if (event === 'updateStarted') return started; + return withoutUndefined({ + ...started, + endedAt: toIso(progress.timestampMs), + durationMs: runningMs === undefined ? undefined : progress.timestampMs - runningMs, + }); +} + +function simplifyUpdate( + event: PipelineUpdateEvent, + progress: UpdateProgress, + entry: TrackedUpdate, + options: ItemOptions, +): IDataObject { + const pipelineUrl = `${options.host}/pipelines/${options.pipelineId}`; + return { + event, + pipeline: withoutUndefined({ + id: options.pipelineId, + name: progress.raw.origin?.pipeline_name, + url: pipelineUrl, + }), + update: { id: progress.updateId, url: `${pipelineUrl}/updates/${progress.updateId}` }, + ...(event !== 'updateStarted' && { result: simplifyResult(progress) }), + timing: simplifyTiming(event, progress, entry), + }; +} + +function toItem( + event: PipelineUpdateEvent, + progress: UpdateProgress, + entry: TrackedUpdate, + options: ItemOptions, +): INodeExecutionData { + return { + json: options.simplify + ? simplifyUpdate(event, progress, entry, options) + : { event, ...progress.raw }, + }; +} + +function collectItems( + updates: TrackedUpdates, + progress: UpdateProgress[], + options: ItemOptions, +): INodeExecutionData[] { + return progress.flatMap((update) => + collectNewEvents(updates, update) + .filter((event) => options.subscribed.includes(event)) + .map((event) => toItem(event, update, updates[update.updateId], options)), + ); +} + +function startWatching(staticData: IDataObject, pipelineId: string): void { + for (const key of Object.keys(staticData)) delete staticData[key]; + const now = Date.now(); + staticData.pipelineId = pipelineId; + staticData.cursorMs = now; + staticData.floorMs = now; + staticData.updates = {}; +} + +function advanceCursor(state: PipelineUpdateWatchState, fetched: PipelineEvent[]): void { + state.cursorMs = fetched.reduce( + (max, event) => Math.max(max, parseTimestamp(event.timestamp) ?? max), + state.cursorMs, + ); + state.floorMs = Math.max(state.floorMs, state.cursorMs - OVERLAP_MS); + for (const [key, entry] of Object.entries(state.updates)) { + if (entry.endedMs !== undefined && entry.endedMs < state.floorMs) delete state.updates[key]; + } + const inFlight = Object.entries(state.updates) + .filter(([, entry]) => entry.endedMs === undefined) + .sort(([, a], [, b]) => (a.startedMs ?? 0) - (b.startedMs ?? 0)); + for (const [key] of inFlight.slice(0, -MAX_IN_FLIGHT_UPDATES)) delete state.updates[key]; +} + +export async function pollPipelineUpdateEvents( + this: IPollFunctions, +): Promise { + const subscribed = readEvents(this, PIPELINE_UPDATE_EVENTS, 'update'); + const simplify = this.getNodeParameter('simplify', true) === true; + const pipelineId = readPipelineId(this); + const credentialType = getActiveCredentialType(this); + const host = await getHost(this, credentialType); + const options: ItemOptions = { subscribed, simplify, pipelineId, host }; + + if (this.getMode() === 'manual') { + const page = await withLegibleErrors( + async () => + await listPipelineEvents(this, credentialType, { + pipelineId, + levels: WATCHED_LEVELS, + order: 'desc', + pageSize: MANUAL_PAGE_SIZE, + }), + PERMISSION_HINT, + ); + return toOutput(collectItems({}, toChronological(page.items), options)); + } + + const staticData = this.getWorkflowStaticData('node'); + if (!isPipelineUpdateWatchState(staticData) || staticData.pipelineId !== pipelineId) { + startWatching(staticData, pipelineId); + return null; + } + + const deadlineEpochMs = Date.now() + this.getPollBudgetMs(); + const after = toIso(staticData.floorMs); + const page = await withLegibleErrors( + async () => + await listAllPipelineEvents( + this, + credentialType, + { pipelineId, after, levels: WATCHED_LEVELS, pageSize: PIPELINE_EVENTS_MAX_PAGE_SIZE }, + { deadlineEpochMs }, + ), + PERMISSION_HINT, + ); + const progress = toChronological(page.items).filter( + (update) => update.timestampMs >= staticData.floorMs, + ); + const items = collectItems(staticData.updates, progress, options); + + advanceCursor(staticData, page.items); + if (page.nextPageToken !== undefined) { + this.logger.info( + `Databricks Trigger could not list every event of pipeline ${pipelineId} since ${after} in one poll. The next poll continues from ${toIso(staticData.cursorMs)}.`, + ); + } + + return toOutput(items); +}