feat: add polling for run-task and URL scraper (#53)

* feat: add polling for run-task and URL scraper

* test: add test for run-task and URL scraper polling

* refactor: extract polling logic to a separate fn

Author: ngut37

nodes/Apify/__tests__/Apify.node.spec.ts

@@ -96,12 +96,12 @@ describe('Apify Node', () => {
 
 	describe('actor-tasks', () => {
 		describe('run-task', () => {
-			it('should run the run-task workflow', async () => {
-				const mockRunTask = fixtures.getRunTaskResult();
+			it('should run the run-task workflow (waitForFinish: false)', async () => {
+				const mockRunTask = fixtures.runActorResult();
 
 				const scope = nock('https://api.apify.com')
 					.post('/v2/actor-tasks/PwUDLcG3zMyT8E4vq/runs')
-					.query(true)
+					.query({ waitForFinish: 0, memory: 1024 })
 					.reply(200, mockRunTask);
 
 				const runTaskWorkflow = require('./workflows/actor-tasks/run-task.workflow.json');
@@ -121,6 +121,37 @@ describe('Apify Node', () => {
 
 				expect(scope.isDone()).toBe(true);
 			});
+
+			it('should run the run-task workflow and wait for finish (waitForFinish: true)', async () => {
+				const mockRunTask = fixtures.runActorResult();
+				const mockFinishedRun = fixtures.getRunTaskResult();
+
+				const scope = nock('https://api.apify.com')
+					.post('/v2/actor-tasks/PwUDLcG3zMyT8E4vq/runs')
+					.query({ waitForFinish: 0, memory: 1024 })
+					.reply(200, mockRunTask)
+					.get(`/v2/actor-runs/${mockRunTask.data.id}`)
+					.reply(200, mockFinishedRun);
+
+				const runTaskWorkflow = require('./workflows/actor-tasks/run-task-wait-for-finish.workflow.json');
+				const { waitPromise } = await executeWorkflow({
+					credentialsHelper,
+					workflow: runTaskWorkflow,
+				});
+				const result = await waitPromise.promise();
+
+				const nodeResults = getRunTaskDataByNodeName(result, 'Run task');
+				expect(nodeResults.length).toBe(1);
+				const [nodeResult] = nodeResults;
+				expect(nodeResult.executionStatus).toBe('success');
+
+				const data = getTaskData(nodeResult);
+				// expect polled terminal run as result
+				expect(data).not.toEqual(mockRunTask.data);
+				expect(data).toEqual(mockFinishedRun.data);
+
+				expect(scope.isDone()).toBe(true);
+			});
 		});
 	});
 
@@ -187,7 +218,7 @@ describe('Apify Node', () => {
 			it('should run the run-actor workflow and wait for finish', async () => {
 				const mockRunActor = fixtures.runActorResult();
 				const mockBuild = fixtures.getBuildResult();
-				const mockFinishedRun = fixtures.getRunResult();
+				const mockFinishedRun = fixtures.getSuccessRunResult();
 
 				const scope = nock('https://api.apify.com')
 					.get('/v2/acts/nFJndFXA5zjCTuudP')
@@ -222,15 +253,18 @@ describe('Apify Node', () => {
 		});
 		describe('scrape-single-url', () => {
 			it('should run the scrape-single-url workflow', async () => {
-				const mockRunResponse = fixtures.runScrapeSingleUrlActorResult();
+				const mockRunActor = fixtures.runActorResult();
+				const mockFinishedRun = fixtures.getSuccessRunResult();
 				const mockItems = fixtures.getScrapeSingleUrlItemsResult();
 
-				const datasetId = mockRunResponse.data.defaultDatasetId;
+				const datasetId = mockFinishedRun.data.defaultDatasetId;
 
 				const scope = nock('https://api.apify.com')
 					.post(`/v2/acts/${helpers.consts.WEB_CONTENT_SCRAPER_ACTOR_ID}/runs`)
-					.query({ waitForFinish: 60 })
-					.reply(200, mockRunResponse)
+					.query({ waitForFinish: 0 })
+					.reply(200, mockRunActor)
+					.get(`/v2/actor-runs/${mockRunActor.data.id}`)
+					.reply(200, mockFinishedRun)
 					.get(`/v2/datasets/${datasetId}/items`)
 					.query({ format: 'json' })
 					.reply(200, mockItems);

nodes/Apify/__tests__/utils/fixtures.ts

@@ -370,7 +370,7 @@ export const getSuccessRunResult = () => {
 			buildId: 'DgGC7ZxWmZ0cnuNIy',
 			exitCode: 0,
 			defaultKeyValueStoreId: '4aoPkdUdfoRpf9w7E',
-			defaultDatasetId: 'U0hEU6N57UfDgqI98',
+			defaultDatasetId: '63kMAihbWVgBvEAZ2',
 			defaultRequestQueueId: 'qNGmAgPf3Pb50PhqC',
 			pricingInfo: {
 				pricingModel: 'PAY_PER_EVENT',

nodes/Apify/__tests__/workflows/actor-tasks/run-task-wait-for-finish.workflow.json

@@ -0,0 +1,43 @@
+{
+	"name": "Run task and wait for finish workflow",
+	"nodes": [
+		{
+			"parameters": {
+				"resource": "Actor tasks",
+				"operation": "Run task",
+				"actorTaskId": {
+					"__rl": true,
+					"value": "PwUDLcG3zMyT8E4vq",
+					"mode": "list",
+					"cachedResultName": "Google Search Results Scraper (Task)",
+					"cachedResultUrl": "https://console.apify.com/actors/tasks/PwUDLcG3zMyT8E4vq/input"
+				},
+				"waitForFinish": true
+			},
+			"id": "2ada004c-bc23-435d-a65c-371ebcafb4f5",
+			"name": "Run task",
+			"type": "n8n-nodes-apify.apify",
+			"typeVersion": 1,
+			"position": [920, 460],
+			"credentials": {
+				"apifyApi": {
+					"id": "dJAynKkN2pRqy3Ko",
+					"name": "Apify account"
+				}
+			}
+		}
+	],
+	"pinData": {},
+	"connections": {},
+	"active": false,
+	"settings": {
+		"executionOrder": "v1"
+	},
+	"versionId": "29897fc5-8618-4467-9665-358072641546",
+	"meta": {
+		"templateCredsSetupCompleted": true,
+		"instanceId": "3f65ba173ae28613be507e94c0a98de4375527c55e31b7fc173a4edee4e2ded3"
+	},
+	"id": "OFLCq8UoIVInQv2Y",
+	"tags": []
+}

nodes/Apify/__tests__/workflows/actor-tasks/run-task.workflow.json

@@ -1,5 +1,5 @@
 {
-	"name": "Run task workflow",
+	"name": "Run task no wait workflow",
 	"nodes": [
 		{
 			"parameters": {
@@ -12,7 +12,7 @@
 					"cachedResultName": "Google Search Results Scraper (Task)",
 					"cachedResultUrl": "https://console.apify.com/actors/tasks/PwUDLcG3zMyT8E4vq/input"
 				},
-				"waitForFinish": 60
+				"waitForFinish": false
 			},
 			"id": "2ada004c-bc23-435d-a65c-371ebcafb4f5",
 			"name": "Run task",

nodes/Apify/resources/actor-tasks/run-task/execute.ts

@@ -4,14 +4,14 @@ import {
 	NodeApiError,
 	NodeOperationError,
 } from 'n8n-workflow';
-import { apiRequest } from '../../../resources/genericFunctions';
+import { apiRequest, pollRunStatus } from '../../../resources/genericFunctions';
 
 export async function runTask(this: IExecuteFunctions, i: number): Promise<INodeExecutionData> {
 	const actorTaskId = this.getNodeParameter('actorTaskId', i, undefined, {
 		extractValue: true,
 	}) as string;
 	const input = this.getNodeParameter('customBody', i, {}) as object;
-	const waitForFinish = this.getNodeParameter('waitForFinish', i, null) as number | null;
+	const waitForFinish = this.getNodeParameter('waitForFinish', i) as boolean;
 	const timeout = this.getNodeParameter('timeout', i, null) as number | null;
 	const memory = this.getNodeParameter('memory', i, null) as number | null;
 	const build = this.getNodeParameter('build', i, '') as string;
@@ -21,27 +21,29 @@ export async function runTask(this: IExecuteFunctions, i: number): Promise<INode
 	}
 
 	const qs: Record<string, any> = {};
-	if (waitForFinish != null) qs.waitForFinish = waitForFinish;
 	if (timeout != null) qs.timeout = timeout;
 	if (memory != null) qs.memory = memory;
 	if (build) qs.build = build;
+	qs.waitForFinish = 0; // always start run without waiting
 
-	try {
-		const apiResult = await apiRequest.call(this, {
-			method: 'POST',
-			uri: `/v2/actor-tasks/${actorTaskId}/runs`,
-			body: input,
-			qs,
-		});
+	const apiResult = await apiRequest.call(this, {
+		method: 'POST',
+		uri: `/v2/actor-tasks/${actorTaskId}/runs`,
+		body: input,
+		qs,
+	});
 
-		if (!apiResult) {
-			throw new NodeApiError(this.getNode(), {
-				message: `Task run for ${actorTaskId} not found`,
-			});
-		}
+	if (!apiResult?.data?.id) {
+		throw new NodeApiError(this.getNode(), {
+			message: `Run ID not found after running the task`,
+		});
+	}
 
+	if (!waitForFinish) {
 		return { json: { ...apiResult.data } };
-	} catch (error) {
-		throw new NodeApiError(this.getNode(), error);
 	}
+
+	const runId = apiResult.data.id;
+	const lastRunData = await pollRunStatus.call(this, runId);
+	return { json: { ...lastRunData } };
 }

nodes/Apify/resources/actor-tasks/run-task/properties.ts

@@ -47,12 +47,9 @@ export const properties: INodeProperties[] = [
 		displayName: 'Wait for Finish',
 		name: 'waitForFinish',
 		description:
-			"The maximum number of seconds the server waits for the run to finish. By default, the server doesn't wait for the run to finish and returns immediately. The maximum value is 60 seconds.",
-		default: null,
-		type: 'number',
-		typeOptions: {
-			maxValue: 60,
-		},
+			'Whether to wait for the run to finish. If true, the node will poll the run status until it reaches a terminal state (SUCCEEDED, FAILED, TIMED-OUT, ABORTED) or 5 minutes have passed. If false, the node will return immediately after starting the run.',
+		default: true,
+		type: 'boolean',
 		displayOptions: {
 			show: {
 				resource: ['Actor tasks'],

nodes/Apify/resources/actors/run-actor/execute.ts

@@ -4,8 +4,7 @@ import {
 	NodeApiError,
 	NodeOperationError,
 } from 'n8n-workflow';
-import { apiRequest } from '../../genericFunctions';
-import * as helpers from '../../../helpers';
+import { apiRequest, pollRunStatus } from '../../genericFunctions';
 
 export async function runActor(this: IExecuteFunctions, i: number): Promise<INodeExecutionData> {
 	const actorId = this.getNodeParameter('actorId', i, undefined, {
@@ -83,27 +82,7 @@ export async function runActor(this: IExecuteFunctions, i: number): Promise<INod
 	// This loop is infinite and will only stop when a terminal status is reached,
 	// or when the workflow maximum timeout is hit, as set in your n8n configuration.
 	const runId = run.data.id;
-	let lastRunData = run.data;
-	while (true) {
-		try {
-			const pollResult = await apiRequest.call(this, {
-				method: 'GET',
-				uri: `/v2/actor-runs/${runId}`,
-			});
-			const status = pollResult?.data?.status;
-			lastRunData = pollResult?.data;
-			if (helpers.consts.TERMINAL_RUN_STATUSES.includes(status)) {
-				break;
-			}
-		} catch (err) {
-			throw new NodeApiError(this.getNode(), {
-				message: `Error polling run status: ${err}`,
-			});
-		}
-		await new Promise((resolve) =>
-			setTimeout(resolve, helpers.consts.WAIT_FOR_FINISH_POLL_INTERVAL),
-		);
-	}
+	const lastRunData = await pollRunStatus.call(this, runId);
 	return {
 		json: { ...lastRunData },
 	};

nodes/Apify/resources/actors/scrape-single-url/execute.ts

@@ -1,5 +1,5 @@
 import { IExecuteFunctions, INodeExecutionData, NodeApiError } from 'n8n-workflow';
-import { apiRequest } from '../../../resources/genericFunctions';
+import { apiRequest, pollRunStatus } from '../../../resources/genericFunctions';
 import { consts } from '../../../helpers';
 
 export async function scrapeSingleUrl(
@@ -24,15 +24,27 @@ export async function scrapeSingleUrl(
 			saveMarkdown: true,
 		};
 
-		// Run the actor and wait for it to finish
+		// Run the actor and do not wait for finish
+
 		const run = await apiRequest.call(this, {
 			method: 'POST',
 			uri: `/v2/acts/${consts.WEB_CONTENT_SCRAPER_ACTOR_ID}/runs`,
 			body: input,
-			qs: { waitForFinish: 60 },
+			qs: { waitForFinish: 0 },
 		});
 
-		const defaultDatasetId = run?.data?.defaultDatasetId || run?.defaultDatasetId;
+		const runId = run?.data?.id || run?.id;
+
+		if (!runId) {
+			throw new NodeApiError(this.getNode(), {
+				message: 'No run ID returned from actor run',
+			});
+		}
+
+		// Poll for terminal status
+		const lastRunData = await pollRunStatus.call(this, runId);
+
+		const defaultDatasetId = lastRunData?.defaultDatasetId;
 
 		if (!defaultDatasetId) {
 			throw new NodeApiError(this.getNode(), {

nodes/Apify/resources/genericFunctions.ts

@@ -118,6 +118,34 @@ export async function apiRequestAllItems(
 	return combinedData;
 }
 
+export async function pollRunStatus(
+	this: IHookFunctions | IExecuteFunctions | ILoadOptionsFunctions,
+	runId: string,
+): Promise<any> {
+	let lastRunData: any;
+	while (true) {
+		try {
+			const pollResult = await apiRequest.call(this, {
+				method: 'GET',
+				uri: `/v2/actor-runs/${runId}`,
+			});
+			const status = pollResult?.data?.status;
+			lastRunData = pollResult?.data;
+			if (['SUCCEEDED', 'FAILED', 'TIMED-OUT', 'ABORTED'].includes(status)) {
+				break;
+			}
+		} catch (err) {
+			throw new NodeApiError(this.getNode(), {
+				message: `Error polling run status: ${err}`,
+			});
+		}
+		await new Promise(
+			(resolve) => setTimeout(resolve, 1000), // 1 second polling interval
+		);
+	}
+	return lastRunData;
+}
+
 export function getActorOrTaskId(this: IHookFunctions): string {
 	const resource = this.getNodeParameter('resource', '') as string;
 	const actorId = this.getNodeParameter('actorId', '') as { value: string };