Fix deleting Actors and log timestamps (#30)

Co-authored-by: Claude <noreply@anthropic.com>

Author: Pijukatel

requirements/api.md

@@ -34,6 +34,10 @@
     - `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.
+- Log content matches the Apify platform's log format: every log line starts with an ISO-8601 UTC
+  timestamp with millisecond precision followed by a space (`2026-08-31T09:13:25.123Z `), exactly one
+  timestamp per line regardless of how the output was chunked when produced. Apify clients' log
+  redirection (e.g. `Actor.call` in the SDKs) relies on this prefix to recognize log messages.
 
 # Actor id encoding
 

src/api/routes/actors.ts

@@ -94,7 +94,10 @@ export function mountActors(router: Router, deps: ApiServerDeps): void {
 				req.params.actorId as string,
 				requireUser(req).username,
 			);
-			if (actor) await deleteActor(actor.id);
+			// Matches the real platform: DELETE of a missing Actor 404s the same as GET (api.md's
+			// "applies uniformly to every DELETE").
+			if (!actor) throw recordNotFound();
+			await deleteActor(actor.id);
 			res.status(204).end();
 		}),
 	);
@@ -182,12 +185,13 @@ export function mountActors(router: Router, deps: ApiServerDeps): void {
 				req.params.actorId as string,
 				requireUser(req).username,
 			);
-			if (actor) {
-				await updateActor(actor.id, (current) => ({
-					...current,
-					versions: current.versions.filter((v) => v.versionNumber !== req.params.versionNumber),
-				}));
-			}
+			// Matches the real platform: a missing Actor and a missing version both 404, never a silent 204.
+			if (!actor) throw recordNotFound();
+			if (!findVersion(actor, req.params.versionNumber as string)) throw recordNotFound();
+			await updateActor(actor.id, (current) => ({
+				...current,
+				versions: current.versions.filter((v) => v.versionNumber !== req.params.versionNumber),
+			}));
 			res.status(204).end();
 		}),
 	);

src/services/logs.ts

@@ -9,6 +9,9 @@ interface LiveLog {
 	buffer: string[];
 	subscribers: Set<(chunk: string) => void>;
 	terminal: boolean;
+	/** Whether the next appended character starts a new log line - lets `appendLog` stamp exactly one
+	 * timestamp per line even when chunk boundaries fall mid-line. */
+	atLineStart: boolean;
 }
 
 const live = new Map<string, LiveLog>();
@@ -26,17 +29,41 @@ const flushMutex = new KeyedMutex();
 function getOrCreate(id: string): LiveLog {
 	let state = live.get(id);
 	if (!state) {
-		state = { buffer: [], subscribers: new Set(), terminal: false };
+		state = { buffer: [], subscribers: new Set(), terminal: false, atLineStart: true };
 		live.set(id, state);
 	}
 	return state;
 }
 
+/**
+ * Prefixes every log *line* in `chunk` with an ingestion timestamp (`2026-08-31T09:13:25.123Z `), the
+ * platform's log format (api.md). Apify clients' log redirection recognizes messages by this prefix,
+ * so unstamped lines would never be redirected.
+ */
+function stampLines(state: LiveLog, chunk: string): string {
+	let out = '';
+	let from = 0;
+	while (from < chunk.length) {
+		if (state.atLineStart) out += `${new Date().toISOString()} `;
+		const newlineAt = chunk.indexOf('\n', from);
+		if (newlineAt === -1) {
+			out += chunk.slice(from);
+			state.atLineStart = false;
+			break;
+		}
+		out += chunk.slice(from, newlineAt + 1);
+		state.atLineStart = true;
+		from = newlineAt + 1;
+	}
+	return out;
+}
+
 export function appendLog(id: string, chunk: string): void {
 	if (!chunk) return;
 	const state = getOrCreate(id);
-	state.buffer.push(chunk);
-	for (const subscriber of state.subscribers) subscriber(chunk);
+	const stamped = stampLines(state, chunk);
+	state.buffer.push(stamped);
+	for (const subscriber of state.subscribers) subscriber(stamped);
 }
 
 /** Returns an unsubscribe function. */

test/integration/actors-builds-runs.test.ts

@@ -310,6 +310,59 @@ describe('actors / versions / builds / runs (via real apify-client)', () => {
 		expect(body.error.type).toBe('record-not-found');
 	});
 
+	it('deletes an Actor for real, and a missing Actor 404s on DELETE the same as on GET (record-not-found)', async () => {
+		const actor = await server.client.actors().create({ name: 'delete-actor' });
+		const user = await server.client.user('me').get();
+
+		await server.client.actor(actor.id).delete();
+		expect(await server.client.actor(actor.id).get()).toBeUndefined();
+
+		const list = await server.client.actors().list();
+		expect(list.items.some((a) => a.id === actor.id)).toBe(false);
+
+		// apify-client-js's own `.delete()` swallows a `record-not-found` 404 to stay idempotent from the
+		// caller's perspective, so hit the HTTP endpoint directly to observe the real status/envelope.
+		for (const missingRef of ['nonexistent12345b', `${user.username}~does-not-exist`]) {
+			const res = await fetch(`${server.baseUrl}/v2/actors/${missingRef}`, {
+				method: 'DELETE',
+				headers: { Authorization: `Bearer ${server.token}` },
+			});
+			expect(res.status).toBe(404);
+			const body = (await res.json()) as { error: { type: string } };
+			expect(body.error.type).toBe('record-not-found');
+		}
+	});
+
+	it('deletes an Actor version for real, and a missing version or Actor 404s on DELETE (record-not-found)', async () => {
+		const actor = await server.client.actors().create({ name: 'delete-version-actor' });
+		await server.client
+			.actor(actor.id)
+			.versions()
+			.create({
+				versionNumber: '0.0',
+				buildTag: 'latest',
+				sourceType: 'SOURCE_FILES' as never,
+				sourceFiles: [],
+			} as never);
+
+		await server.client.actor(actor.id).version('0.0').delete();
+		expect(await server.client.actor(actor.id).version('0.0').get()).toBeUndefined();
+
+		const missingVersion = await fetch(`${server.baseUrl}/v2/actors/${actor.id}/versions/9.9`, {
+			method: 'DELETE',
+			headers: { Authorization: `Bearer ${server.token}` },
+		});
+		expect(missingVersion.status).toBe(404);
+		expect(((await missingVersion.json()) as { error: { type: string } }).error.type).toBe('record-not-found');
+
+		const missingActor = await fetch(`${server.baseUrl}/v2/actors/nonexistent12345c/versions/0.0`, {
+			method: 'DELETE',
+			headers: { Authorization: `Bearer ${server.token}` },
+		});
+		expect(missingActor.status).toBe(404);
+		expect(((await missingActor.json()) as { error: { type: string } }).error.type).toBe('record-not-found');
+	});
+
 	it('DELETE rejects a non-terminal build instead of deleting it (matches the real platform: reject, not abort-then-delete)', async () => {
 		const actor = await server.client.actors().create({ name: 'delete-running-build-actor' });
 		const { builds } = getRegistries();

test/integration/log-drain-race.test.ts

@@ -147,6 +147,7 @@ describe('end-to-end: a run never reads terminal via HTTP before its log has ful
 		// A fresh, non-stream log read the instant status is observed terminal must contain the full
 		// output - not just what had arrived before the container "exited".
 		const log = await server.client.log(run.id).get();
-		expect(log).toBe('final line\n');
+		// Strip the per-line timestamp - this test is about the drain race, not the log format.
+		expect(log?.replace(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z /, '')).toBe('final line\n');
 	});
 });

test/integration/logs.test.ts

@@ -12,6 +12,16 @@ import {
 import { getRegistries } from '../../src/storage/registries.js';
 import { getOrCreateUserForToken } from '../../src/services/users.js';
 
+/** The per-line timestamp prefix from the platform log format (api.md). */
+const LINE_STAMP = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z /;
+
+function stripStamps(log: string): string {
+	return log
+		.split('\n')
+		.map((line) => line.replace(LINE_STAMP, ''))
+		.join('\n');
+}
+
 /** Polls `check()` until it returns true or `timeoutMs` elapses, rather than a fixed sleep - keeps the
  * disconnect test fast on the happy path and still deterministic if cleanup is ever slow. */
 async function waitUntil(check: () => boolean, timeoutMs = 2000): Promise<void> {
@@ -195,7 +205,7 @@ describe('log streaming', () => {
 		await second;
 		spy.mockRestore();
 
-		expect(await getFullLog(jobId)).toBe('line1\nline2\n');
+		expect(stripStamps(await getFullLog(jobId))).toBe('line1\nline2\n');
 	});
 
 	it('getFullLog never returns a torn read while a flush for the same id is in flight (regression for the reader-vs-writer race)', async () => {
@@ -232,7 +242,7 @@ describe('log streaming', () => {
 		const result = await read;
 		spy.mockRestore();
 
-		expect(result).toBe('first chunk\n');
+		expect(stripStamps(result)).toBe('first chunk\n');
 	});
 
 	it("flushAllLogs persists every live log's buffered content (regression for a lost trailing chunk on graceful shutdown)", async () => {
@@ -248,8 +258,8 @@ describe('log streaming', () => {
 
 		await flushAllLogs();
 
-		expect(await getRegistries().logs.getValue(jobId1)).toBe('alpha\n');
-		expect(await getRegistries().logs.getValue(jobId2)).toBe('beta\n');
+		expect(stripStamps((await getRegistries().logs.getValue(jobId1)) ?? '')).toBe('alpha\n');
+		expect(stripStamps((await getRegistries().logs.getValue(jobId2)) ?? '')).toBe('beta\n');
 	});
 
 	it('a stream closes when the persisted record turns terminal even if markLogTerminal is never called (regression: aborted in the READY window)', async () => {
@@ -401,7 +411,7 @@ describe('log streaming', () => {
 
 		const reader = res.body!.getReader();
 		const { value } = await reader.read();
-		expect(Buffer.from(value!).toString()).toBe('line 1\n');
+		expect(stripStamps(Buffer.from(value!).toString())).toBe('line 1\n');
 
 		// The initial response is a single `res.status(200).send(soFar)`, never `res.write` + a
 		// held-open connection, so `subscribeLog` is never called - checked immediately after the first
@@ -444,10 +454,91 @@ describe('log streaming', () => {
 
 		const reader = res.body!.getReader();
 		const { value } = await reader.read();
-		expect(Buffer.from(value!).toString()).toBe('line 1\n');
+		expect(stripStamps(Buffer.from(value!).toString())).toBe('line 1\n');
 		expect(getSubscriberCount(jobId)).toBe(0);
 
 		const { done } = await reader.read();
 		expect(done).toBe(true);
 	});
+
+	it('stamps every log line with a platform-style ISO timestamp, exactly once per line even when chunk boundaries fall mid-line', async () => {
+		const jobId = 'stampedLinesJobId12';
+		// Docker output is not line-aligned: one line arrives split over two appends, and one append
+		// carries two lines. Both must come out with exactly one stamp per *line*.
+		appendLog(jobId, '[apify] INFO first line\n[apify] WARN second li');
+		appendLog(jobId, 'ne, same stamp\n');
+		appendLog(jobId, '[apify] INFO third line\n');
+
+		const log = await getFullLog(jobId);
+		const lines = log.split('\n').filter((line) => line.length > 0);
+		expect(lines).toHaveLength(3);
+		for (const line of lines) expect(line).toMatch(LINE_STAMP);
+		// The continuation of the split line must NOT have been stamped mid-line.
+		expect(lines[1]).toMatch(/second line, same stamp$/);
+		expect(stripStamps(log)).toBe(
+			'[apify] INFO first line\n[apify] WARN second line, same stamp\n[apify] INFO third line\n',
+		);
+	});
+
+	it('apify-client log redirection recovers every message from ?stream=true&raw=true (regression: Actor.call redirected nothing)', async () => {
+		const jobId = 'redirectedLogJobId1';
+		appendLog(jobId, '[apify] INFO Initializing Actor...\n');
+
+		const user = await getOrCreateUserForToken(server.token);
+		await getRegistries().runs.set(jobId, {
+			id: jobId,
+			userId: user.id,
+			actorId: 'x',
+			buildId: 'y',
+			buildNumber: '0.0.1',
+			status: 'RUNNING',
+			startedAt: new Date().toISOString(),
+			defaultDatasetId: 'd',
+			defaultKeyValueStoreId: 'k',
+			defaultRequestQueueId: 'r',
+			options: { memoryMbytes: 1024, timeoutSecs: 300 },
+			meta: { origin: 'API' },
+		});
+
+		// The exact request apify-client's log redirection makes.
+		const res = await fetch(`${server.baseUrl}/v2/actor-runs/${jobId}/log?stream=true&raw=true`, {
+			headers: { Authorization: `Bearer ${server.token}` },
+		});
+		expect(res.status).toBe(200);
+
+		setTimeout(() => appendLog(jobId, '[apify] INFO doing work\n[apify] WARN partial li'), 50);
+		setTimeout(() => appendLog(jobId, 'ne finished\n'), 100);
+		setTimeout(() => markLogTerminal(jobId), 150);
+
+		// Replicates the client's redirect parsing: buffer chunks, split on the timestamp marker, emit
+		// only marker-delimited messages (a possibly incomplete trailing part waits in the buffer until
+		// the final flush). Without the per-line timestamps this recovers zero messages.
+		const splitMarker = /(?:\n|^)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)/;
+		let streamBuffer = '';
+		const messages: string[] = [];
+		const flushBuffer = (includeLastPart: boolean) => {
+			const allParts = streamBuffer.split(splitMarker).slice(1);
+			const complete = includeLastPart ? allParts : allParts.slice(0, -2);
+			streamBuffer = includeLastPart ? '' : allParts.slice(-2).join('');
+			for (let i = 0; i + 1 < complete.length; i += 2) {
+				messages.push(`${complete[i]}${complete[i + 1]}`.trim());
+			}
+		};
+
+		const reader = res.body!.getReader();
+		for (;;) {
+			const { value, done } = await reader.read();
+			if (done) break;
+			streamBuffer += Buffer.from(value!).toString();
+			if (splitMarker.test(streamBuffer)) flushBuffer(false);
+		}
+		flushBuffer(true);
+
+		const contentsOnly = messages.map((message) => message.replace(LINE_STAMP, ''));
+		expect(contentsOnly).toEqual([
+			'[apify] INFO Initializing Actor...',
+			'[apify] INFO doing work',
+			'[apify] WARN partial line finished',
+		]);
+	});
 });