Enforce CPU+memory limits on Actor containers and stream systemInfo events to runs

package.json

@@ -31,7 +31,8 @@
 		"dockerode": "^4.0.5",
 		"express": "^5.1.0",
 		"json5": "^2.2.3",
-		"tar-stream": "^3.1.7"
+		"tar-stream": "^3.1.7",
+		"ws": "^8.21.3"
 	},
 	"devDependencies": {
 		"@eslint/js": "^9.18.0",
@@ -40,6 +41,7 @@
 		"@types/express-serve-static-core": "^5.1.3",
 		"@types/node": "^22.13.0",
 		"@types/tar-stream": "^3.1.3",
+		"@types/ws": "^8.18.1",
 		"apify-client": "^2.13.0",
 		"axios": "^1.7.9",
 		"eslint": "^9.18.0",

requirements/actor-driver.md

@@ -101,6 +101,34 @@
 - Actor run details are saved in `__RUNS__` internal storage
 - Actor run log is saved in `__LOGS__` internal storage
 
+## Resource limits
+
+- Every run's container has a hard memory limit and a hard CPU limit. Memory is the run's `memoryMbytes`;
+  CPU is derived from it at the platform's ratio of one core per 4096 MB, so a 1024 MB run gets 0.25 core.
+- A derived CPU limit below what Docker accepts is raised to that minimum.
+- Limits are applied exactly as requested, even when they exceed the host's own capacity. Such a run is
+  warned about in its own log, naming the requested and the host figures; the limits still apply. When the
+  host's capacity cannot be determined, no warning is produced.
+- Disk is not limited. `diskMbytes` is reported but never enforced.
+
+## Run resource telemetry
+
+- While a run's container is up, its CPU and memory usage are measured once a second and published as
+  `systemInfo` events on the run's events channel (`api.md`).
+- Every event carries all eight fields - `memAvgBytes`, `memCurrentBytes`, `memMaxBytes`, `cpuAvgUsage`,
+  `cpuMaxUsage`, `cpuCurrentUsage`, `isCpuOverloaded`, `createdAt` - or is not published at all. A
+  measurement that cannot be read completely is skipped, leaving the run's running figures unaffected.
+    - `cpuCurrentUsage` is percent of one CPU core, not of the run's own grant.
+    - `memCurrentBytes` and `memAvgBytes` exclude reclaimable page cache, matching what `docker stats`
+      reports for the same container.
+    - `memMaxBytes` is the run's configured memory limit, constant for its lifetime.
+    - `isCpuOverloaded` is true when used cores exceed 95% of the run's granted cores.
+    - `memAvgBytes`, `cpuAvgUsage` and `cpuMaxUsage` cover every sample published for that run so far.
+- Measurement lasts exactly as long as the container: it starts once the container is running and stops
+  before the container is removed, and it never delays a run from reaching a terminal state.
+- The bundled sample Actors log their granted resources and each `systemInfo` event they receive, so the
+  contract is observable from a single `apify call`.
+
 # Users
 
 - Users are created adhoc by the runtime for each new token used in the API call (`cli.md`'s User bootstrap).
@@ -132,3 +160,10 @@
   resolved against it (`cli.md`'s User bootstrap). If neither source has a value, the key is absent entirely — never a
   placeholder value. One host-level password (source 1)
   or one harvested-per-account password (source 2) used specifically for each user.
+- `ACTOR_EVENTS_WEBSOCKET_URL` / `APIFY_ACTOR_EVENTS_WS_URL` — the run's own events channel
+  (`api.md`), carrying no credential.
+- `ACTOR_MEMORY_MBYTES` / `APIFY_MEMORY_MBYTES` — the run's requested `memoryMbytes`.
+- `APIFY_DEDICATED_CPUS` — the run's granted CPU cores. No `ACTOR_`-prefixed counterpart; only the
+  Python SDK reads it.
+- Every `ACTOR_*`/`APIFY_*` pair above is set to an identical value: the two SDKs disagree on which name
+  wins, so a divergent pair would size the same run differently depending on which one reads it.

requirements/api.md

@@ -25,12 +25,14 @@
   This also protects the runtime's own driver invariant: deleting the record first would permanently
   strand a running Docker container, since its only stop path (`POST .../abort`) requires the record to
   still resolve, and startup reconciliation only ever considers _existing_ run records.
-- Two endpoints are exceptions to the `{data}` envelope:
+- Three endpoints are exceptions to the `{data}` envelope:
     - `GET /v2/logs/:buildOrRunId` (and its `actor-builds`/`actor-runs` aliases): the body is plain text,
       never `{data}`-wrapped, matching apify-client-js's `log().get()`.
     - `GET /v2/datasets/:datasetId/items` (and its `actor-runs/:runId/dataset/items` alias): the body is
       a bare JSON array of items, never `{data}`-wrapped, with pagination metadata carried in
       `x-apify-pagination-*` response headers, matching apify-client-js's `_createPaginationList`.
+    - `GET /actor-runtime/events/:runId`: a websocket upgrade, not a JSON response at all - see "Actor
+      runtime API" below.
 - `*At` timestamp fields are ISO-8601 strings.
 
 # Actor id encoding
@@ -149,6 +151,30 @@
   console-local, unauthenticated route on the console's own port. Both routes funnel into the same
   underlying validate-and-persist path, so the two surfaces can never drift apart in behavior, only in
   how they are reached.
+- **`GET /actor-runtime/events/:runId`** - a websocket upgrade, reachable at exactly this one path on
+  the fixed API port (`system.md`). It carries the run's platform events: `systemInfo` once a second
+  (`actor-driver.md`), plus a one-off `aborting` frame under `?gracefully=` (below). Each frame is a
+  single text message, `{"name": "...", "data": {...}}`.
+    - The endpoint has no authentication. The run id in the path is the only thing it scopes on, and a
+      connection only ever receives that run's own frames; one run never sees another's.
+    - An unknown or already-terminal run id gets a completed upgrade followed immediately by a `1008`
+      close with a reason, never a non-101 HTTP status - the Python SDK treats a refused first connection
+      as fatal to the Actor.
+    - A connection to a live run stays open until the run ends, when the server closes it with `1000`. It
+      is never dropped while healthy, except that a graceful runtime shutdown terminates every open
+      connection along with the rest of the server.
+    - `persistState` is never sent over this channel; both SDKs generate it themselves.
+
+## Graceful abort (`?gracefully=`)
+
+- `POST /v2/actor-runs/:runId/abort` accepts an optional `?gracefully=` boolean.
+- Omitted or `false`: the run aborts immediately, exactly as before this parameter existed.
+- `true` on a running run: the record moves to `ABORTING` at once, an `aborting` frame with an empty
+  payload is published on the run's events channel, and the container is stopped 30 seconds later. The
+  request stays open until then, so it is the longest-held response in this API.
+- `true` on a run with no container (still `READY`, or already terminal): behaves as if omitted.
+- A second abort arriving during an open window: another `?gracefully=true` joins that window and neither
+  restarts it nor stops the container early; a non-graceful one escalates and stops the container at once.
 
 ## Upstream fallback (opt-in, off by default, all HTTP methods)
 

requirements/system.md

@@ -28,9 +28,7 @@
   status message, but every other endpoint - storages, actor/build/run records, console - still works).
 - The API port (3333) and the console frontend port (3000) are fixed values and
   are not configurable; they are the same on every start of the container.
-
-# Running the container
-
+- Port 3333 also serves the per-run events websocket (`api.md`); no additional port is published for it.
 - Required `docker run` flags: mount the host Docker socket read-write
   (`-v /var/run/docker.sock:/var/run/docker.sock`) so the runtime can build and run Actor containers,
   and mount a persistent data directory (`-v <host-dir>:/data`, e.g. `-v "$(pwd)/data:/data"`) so
@@ -63,10 +61,11 @@
   reuse the already-pulled base image and every other push/call/log-stream/storage-access needs no
   outbound network access at all (see `cli.md`'s offline-capability note) - unless the opt-in upstream
   API fallback (`api.md`'s "Upstream fallback" section) has been switched on, in which case a local miss
-  that is eligible for it makes one outbound request per such call.
-- The bundled sample Actors (`sample_actor_ts`, `sample_actor_py`) are not offline: they crawl a live
-  site (`https://crawlee.dev/` by default). Running them needs outbound network access from the Actor
-  container, unlike operating the runtime around them (see `test.md`).
+  that is eligible for it makes one outbound request per such call. Measuring a run's CPU and memory
+  needs no network access of its own.
+- The sample Actors are not offline: they crawl a live site (`https://crawlee.dev/` by default). Running
+  them needs outbound network access from the Actor container, unlike operating the runtime around them
+  (see `test.md`).
 - The runtime's one-time, per-token, real-console identity check (`cli.md`'s User bootstrap) is
   best-effort online with a silent offline fallback.
 
@@ -89,6 +88,9 @@ The intended scale of the system is:
 
 The intended scale is not enforced, but the system operating above the intended scale can experience performance or functional issues.
 
+The "less than 5 running actors at the same time" budget also bounds the per-run resource measurement and
+the open events connections - each running Actor adds at most one of each.
+
 # Tests
 
 The system is tested according to the requirements in `test.md`

sample_actor_py/src/main.py

@@ -7,13 +7,27 @@
 
 from __future__ import annotations
 
-from apify import Actor
+from apify import Actor, Event, EventSystemInfoData
 from crawlee import ConcurrencySettings
 from crawlee.crawlers import ParselCrawler, ParselCrawlingContext
 
 
 async def main() -> None:
     async with Actor:
+        config = Actor.configuration
+        Actor.log.info(
+            f'Resources granted to this run: {config.memory_mbytes} MB memory, {config.dedicated_cpus} CPU core(s).'
+        )
+
+        # The runtime measures this container and relays a systemInfo event once a second.
+        async def log_resource_usage(event_data: EventSystemInfoData) -> None:
+            Actor.log.info(
+                f'Resource usage: CPU {event_data.cpu_info.used_ratio:.1%} of the grant, '
+                f'memory {event_data.memory_info.current_size.to_mb():.1f} MB'
+            )
+
+        Actor.on(Event.SYSTEM_INFO, log_resource_usage)
+
         actor_input = await Actor.get_input() or {}
         start_url = actor_input.get('startUrl', 'https://crawlee.dev/')
         max_pages = int(actor_input.get('maxPages', 2))

sample_actor_ts/src/main.ts

@@ -11,6 +11,20 @@ interface Input {
 	maxPages?: number;
 }
 
+const { memoryMbytes } = Actor.getEnv();
+log.info(
+	`Resources granted to this run: ${memoryMbytes} MB memory, ${process.env.APIFY_DEDICATED_CPUS ?? 'unknown'} CPU core(s).`,
+);
+
+// The runtime measures this container and relays a systemInfo event once a second.
+Actor.on('systemInfo', (info: { cpuCurrentUsage?: number; memCurrentBytes?: number; isCpuOverloaded?: boolean }) => {
+	const memoryMb = info.memCurrentBytes !== undefined ? (info.memCurrentBytes / 1024 / 1024).toFixed(1) : 'unknown';
+	log.info(
+		`Resource usage: CPU ${info.cpuCurrentUsage?.toFixed(1)}% of one core, memory ${memoryMb} MB, ` +
+			`CPU overloaded: ${info.isCpuOverloaded}`,
+	);
+});
+
 const input = await Actor.getInput<Input>();
 const startUrl = input?.startUrl ?? 'https://crawlee.dev/';
 const maxPages = input?.maxPages ?? 2;

src/api/events-ws.ts

@@ -0,0 +1,102 @@
+/**
+ * The events websocket: `GET /actor-runtime/events/:runId`, upgraded on the API server directly
+ * (Express does not handle `upgrade`). Unauthenticated by decision - cross-run isolation is structural,
+ * since the path's `:runId` is the only thing this handler scopes on and there is no broadcast listener.
+ * Rejections complete the upgrade and then close `1008`: refusing the handshake would abort a Python
+ * Actor at `Actor.init()`.
+ */
+import type { IncomingMessage, Server } from 'node:http';
+import type { Duplex } from 'node:stream';
+import { WebSocketServer, type WebSocket } from 'ws';
+
+import { getRunById } from '../services/runs.js';
+import { isTerminalJobStatus } from '../services/job-status.js';
+import { isEventsTerminal, subscribeEvents } from '../services/events-channel.js';
+import { pollUntilTerminal } from './poll-until-terminal.js';
+
+// A trailing slash is a different path, not a second spelling of this one.
+const EVENTS_PATH_PATTERN = /^\/actor-runtime\/events\/([^/]+)$/;
+
+/** Mirrors `api/routes/logs.ts`'s poll cadence. */
+const TERMINAL_POLL_INTERVAL_MS = 250;
+
+export interface EventsWebSocketServer {
+	/**
+	 * Stops accepting upgrades and terminates every open connection. Not redundant with
+	 * `closeServer(apiServer)`: neither Node's `closeAllConnections()` nor `wss.close()` ends an
+	 * already-upgraded socket, so without this a connected Actor hangs graceful shutdown indefinitely.
+	 */
+	close(): void;
+}
+
+function extractRunId(pathname: string): string | undefined {
+	return EVENTS_PATH_PATTERN.exec(pathname)?.[1];
+}
+
+/**
+ * An unknown or already-terminal run is closed `1008`; a live run is subscribed to its own frames and
+ * closed `1000` once it ends. A healthy connection is never closed for any other reason.
+ */
+async function handleConnection(ws: WebSocket, runId: string): Promise<void> {
+	// Must come first: an `'error'` with no listener crashes the process, and any client can provoke one
+	// on this unauthenticated socket. `ws` still emits `'close'` after it, so teardown below is unaffected.
+	ws.on('error', () => undefined);
+
+	const run = await getRunById(runId);
+	if (!run) {
+		ws.close(1008, `Unknown run id: ${runId}`);
+		return;
+	}
+	if (isTerminalJobStatus(run.status)) {
+		ws.close(1008, `Run ${runId} has already ended`);
+		return;
+	}
+
+	const unsubscribe = subscribeEvents(runId, (frame) => {
+		if (ws.readyState === ws.OPEN) ws.send(frame);
+	});
+
+	const poller = pollUntilTerminal({
+		intervalMs: TERMINAL_POLL_INTERVAL_MS,
+		isTerminal: () => isEventsTerminal(runId),
+		refetch: () => getRunById(runId),
+		onTerminal: () => {
+			unsubscribe();
+			if (ws.readyState === ws.OPEN || ws.readyState === ws.CONNECTING) ws.close(1000, `Run ${runId} has ended`);
+		},
+	});
+
+	ws.on('close', () => {
+		poller.stop();
+		unsubscribe();
+	});
+}
+
+/** Registers the upgrade handler on the API server and returns a handle shutdown can close. */
+export function attachEventsWebSocket(server: Server): EventsWebSocketServer {
+	const wss = new WebSocketServer({ noServer: true });
+
+	server.on('upgrade', (req: IncomingMessage, socket: Duplex, head: Buffer) => {
+		const pathname = req.url ? new URL(req.url, 'http://localhost').pathname : undefined;
+		const runId = pathname ? extractRunId(pathname) : undefined;
+		if (!runId) {
+			// Not this endpoint's path, and this is the server's only 'upgrade' listener.
+			socket.destroy();
+			return;
+		}
+
+		wss.handleUpgrade(req, socket, head, (ws) => {
+			// Contains an unexpected rejection to this one connection; unhandled, it would kill the process.
+			void handleConnection(ws, runId).catch(() => {
+				ws.terminate();
+			});
+		});
+	});
+
+	return {
+		close() {
+			for (const client of wss.clients) client.terminate();
+			wss.close();
+		},
+	};
+}