Apify Run Queue

# configurations
.idea

# crawlee storage folder
storage

# installed files
node_modules

storage
node_modules
.venv

FROM apify/actor-node:latest AS builder

WORKDIR /app

COPY package.json yarn.lock ./

# install dependencies
RUN yarn install --production=false

# copy the source code
COPY . ./

RUN yarn build

# runner image

FROM apify/actor-node:latest

WORKDIR /app

COPY package.json yarn.lock ./

RUN yarn install --production

COPY --from=builder /app/dist ./dist

COPY . .

CMD ["yarn", "start"]

{
  "name": "apify-run-queue",
  "version": "1.0.0",
  "type": "module",
  "description": "",
  "dependencies": {
    "apify": "^3.2.6",
    "apify-client": "^2.12.2",
    "zod": "^3.24.3"
  },
  "devDependencies": {
    "@apify/tsconfig": "^0.1.0",
    "@types/node": "^20.0.0",
    "tsx": "^4.4.0",
    "typescript": "~5.6.0"
  },
  "scripts": {
    "start": "node dist/main.js",
    "dev": "tsx src/main.ts",
    "build": "tsc",
    "test": "echo \"Error: oops, the actor has no tests yet, sad!\" && exit 1"
  },
  "author": "",
  "license": "ISC"
}

{
  "extends": "@apify/tsconfig",
  "compilerOptions": {
    "module": "ESNext",
    "target": "ESNext",
    "moduleResolution": "node",
    "outDir": "dist",
    "lib": ["ESNext", "DOM"],
    "strict": true,
    "esModuleInterop": true,
    "resolveJsonModule": true,
    "skipLibCheck": true
  },
  "include": ["./src/**/*"]
}

{
  "actorSpecification": 1,
  "name": "apify-run-queue",
  "version": "0.0",
  "buildTag": "latest",
  "environmentVariables": {},
  "input": "./input.json",
  "defaultMemory": 1024,
  "storages": {
    "dataset": "./output.json"
  }
}

{
  "title": "Input schema for Apify Run Queue",
  "type": "object",
  "schemaVersion": 1,
  "properties": {
    "actorCalls": {
      "title": "Actor Calls",
      "type": "array",
      "description": "The actor calls to run",
      "editor": "json",
      "prefill": [
        {
          "actorId": "apify/website-content-crawler",
          "input": {
            "startUrls": [
              {
                "url": "https://www.apify.com"
              }
            ]
          }
        }
      ]
    }
  },
  "required": ["actorCalls"]
}

{
  "actorSpecification": 1,
  "views": {
    "overview": {
      "title": "Overview",
      "transformation": {
        "fields": [
          "success",
          "actorId",
          "invocationId",
          "runId",
          "retries",
          "error"
        ]
      },
      "display": {
        "component": "table",
        "properties": {
          "success": {
            "label": "Success",
            "format": "boolean"
          },
          "actorId": {
            "label": "Actor Call ID",
            "format": "text"
          },
          "invocationId": {
            "label": "Invocation ID",
            "format": "text"
          },
          "runId": {
            "label": "Run ID",
            "format": "text"
          },
          "retries": {
            "label": "Retries",
            "format": "number"
          },
          "error": {
            "label": "Error",
            "format": "text"
          }
        }
      }
    }
  }
}

1import { Actor, ApifyClient } from 'apify';
2import { z } from 'zod';
3
4const inputSchema = z.object({
5  actorCalls: z.array(
6    z.object({
7      invocationId: z.string().optional().nullable().default(null),
8      actorId: z.string(),
9      input: z.record(z.any()),
10      timeoutSecs: z.number().optional(),
11      webhooks: z.array(z.any()).optional().default([]),
12      maxItems: z.number().optional(),
13      memory: z.number().optional(),
14      build: z.string().optional(),
15    })
16  ),
17  throttleNotificationWebhook: z.string().optional(),
18  retryDelaySecs: z.number().optional().default(30),
19  webhooks: z.array(z.any()).optional().default([]),
20});
21
22const MEMORY_LIMIT_MESSAGE =
23  'By launching this job you will exceed the memory limit of';
24
25Actor.main(async () => {
26  const input = await Actor.getInput();
27  const parsedInput = inputSchema.parse(input);
28  const apify = new ApifyClient({
29    token: process.env.APIFY_TOKEN,
30  });
31
32  const actorCalls = parsedInput.actorCalls;
33  const throttleNotificationWebhook = parsedInput.throttleNotificationWebhook;
34
35  let retries = 0;
36  for (let i = 0; i < actorCalls.length; ) {
37    const actorCall = actorCalls[i];
38    const actor = await apify.actor(actorCall.actorId);
39    const actorInfo = await actor.get();
40
41    if (!actorInfo) {
42      throw new Error(`Actor ${actorCall.actorId} not found`);
43    }
44
45    try {
46      const run = await actor.start(actorCall.input, {
47        timeout: actorCall.timeoutSecs,
48        maxItems: actorCall.maxItems,
49        memory: actorCall.memory,
50        build: actorCall.build,
51        webhooks: [...parsedInput.webhooks, ...actorCall.webhooks],
52      });
53
54      await Actor.pushData({
55        invocationId: actorCall.invocationId,
56        actorId: actorCall.actorId,
57        success: true,
58        runId: run.id,
59        retries,
60        error: null,
61      });
62
63      i += 1;
64      retries = 0;
65    } catch (error) {
66      if (
67        error instanceof Error &&
68        error.message.includes(MEMORY_LIMIT_MESSAGE)
69      ) {
70        const errorMessage =
71          retries > 0
72            ? `Memory limit exceeded when calling ${actorCall.actorId}, retries: ${retries}`
73            : `Memory limit exceeded when calling ${actorCall.actorId}`;
74
75        console.log(errorMessage);
76        if (throttleNotificationWebhook) {
77          await sendThrottleNotification({
78            webhookUrl: throttleNotificationWebhook,
79            actorId: actorInfo.id,
80            invocationId: actorCall.invocationId,
81          });
82        }
83
84        await new Promise((resolve) =>
85          setTimeout(resolve, parsedInput.retryDelaySecs * 1000)
86        );
87        retries += 1;
88      } else {
89        await Actor.pushData({
90          invocationId: actorCall.invocationId,
91          actorId: actorCall.actorId,
92          success: false,
93          runId: null,
94          retries,
95          error: error instanceof Error ? error.message : 'Unknown error',
96        });
97        i += 1;
98        retries = 0;
99      }
100    }
101  }
102});
103
104type ThrottleNotificationParams = {
105  webhookUrl: string;
106  actorId: string;
107  invocationId: string | null;
108};
109
110function sendThrottleNotification({
111  webhookUrl,
112  actorId,
113  invocationId,
114}: ThrottleNotificationParams) {
115  console.log(`Sending throttle notification for ${actorId}`);
116  fetch(webhookUrl, {
117    method: 'POST',
118    body: JSON.stringify({
119      createdAt: new Date().toISOString(),
120      eventType: 'ACTOR.RUN.THROTTLED',
121      eventData: {
122        actorId,
123      },
124      resource: {
125        invocationId,
126        actId: actorId,
127      },
128      status: 'THROTTLED',
129    }),
130  }).catch((err) => {
131    console.error(
132      `Failed to send throttle notification webhook for ${actorId}: ${err}`
133    );
134  });
135}