fix(deploy): address cross-model review findings for Slice A

- Skip recording on replicated executions: peer nodes receiving deploy_component via
  replicateOperation already have req._deploymentId set by origin, so they no longer
  spin up a fresh recorder + UUID + row. Prevents N duplicate rows per N-node cluster.
- Cap payload at 200 MiB while Slice A buffers in memory. Throws a clear ClientError
  pointing users at the package-identifier path or Slice B's streaming variant.
- Register list_deployments and get_deployment in utility/operation_authorization.ts.
  Pattern matches get_components: requires_su=true with the operation enum as the
  named exception so a role can be granted it without SU rights (per the design's
  permission model).
- Add "audit": true to hdb_deployment in systemSchema.json so fresh installs match
  the audit setting the 5-2-0 upgrade directive applies.
- Drop two now-unused imports (Transform from recorder, existsSync from test).
- Auto-format pass via npm run format:write.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: kriszyp

components/deploymentOperations.ts

@@ -47,15 +47,18 @@ export async function handleListDeployments(req: ListRequest = {}): Promise<{ de
 	const conditions: any[] = [];
 	if (req.project) conditions.push({ attribute: 'project', value: req.project });
 	if (req.status) conditions.push({ attribute: 'status', value: req.status });
-	if (req.since != null) conditions.push({ attribute: 'started_at', value: req.since, comparator: 'greater_than_equal' });
+	if (req.since != null)
+		conditions.push({ attribute: 'started_at', value: req.since, comparator: 'greater_than_equal' });
 	if (req.until != null) conditions.push({ attribute: 'started_at', value: req.until, comparator: 'less_than_equal' });
 
 	const collected: any[] = [];
 	for await (const row of table.search(conditions)) {
 		collected.push(stripBlob(row));
 	}
 	// Newest first by started_at; ties broken by deployment_id for stability.
-	collected.sort((a, b) => (b.started_at ?? 0) - (a.started_at ?? 0) || String(a.deployment_id).localeCompare(b.deployment_id));
+	collected.sort(
+		(a, b) => (b.started_at ?? 0) - (a.started_at ?? 0) || String(a.deployment_id).localeCompare(b.deployment_id)
+	);
 
 	const total = collected.length;
 	const offset = Math.max(0, req.offset ?? 0);

components/deploymentRecorder.ts

@@ -10,14 +10,29 @@
 
 import { randomUUID } from 'node:crypto';
 import { createHash, Hash } from 'node:crypto';
-import { Transform } from 'node:stream';
 import type { Readable } from 'node:stream';
 import { databases } from '../resources/databases.ts';
 import { createBlob } from '../resources/blob.ts';
 import * as terms from '../utility/hdbTerms.ts';
+import { ClientError } from '../utility/errors/hdbError.ts';
 import { hostname } from 'node:os';
 
-type DeploymentStatus = 'pending' | 'extracting' | 'installing' | 'loading' | 'replicating' | 'restarting' | 'success' | 'failed' | 'rolled_back';
+// Slice A buffers the entire payload in memory before computing the hash and persisting.
+// This cap prevents an OOM on accidentally-huge uploads while Slice B is in flight. Slice B
+// replaces the buffer with a streaming hash + Blob-source pattern that lifts this limit
+// back to whatever the replication path supports.
+const SLICE_A_PAYLOAD_LIMIT_BYTES = 200 * 1024 * 1024;
+
+type DeploymentStatus =
+	| 'pending'
+	| 'extracting'
+	| 'installing'
+	| 'loading'
+	| 'replicating'
+	| 'restarting'
+	| 'success'
+	| 'failed'
+	| 'rolled_back';
 
 interface CreateOptions {
 	project?: string;
@@ -88,11 +103,27 @@ export class DeploymentRecorder {
 			buffer = Buffer.from(source, 'base64');
 		} else {
 			const chunks: Buffer[] = [];
+			let collected = 0;
 			for await (const chunk of source as AsyncIterable<Buffer | string>) {
-				chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as any));
+				const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as any);
+				collected += buf.length;
+				if (collected > SLICE_A_PAYLOAD_LIMIT_BYTES) {
+					(source as Readable).destroy?.();
+					throw new ClientError(
+						`Deploy payload exceeds Slice A's interim ${SLICE_A_PAYLOAD_LIMIT_BYTES} byte cap. ` +
+							`Use a package identifier (npm:/file:/git:) or wait for Slice B's streaming path.`
+					);
+				}
+				chunks.push(buf);
 			}
 			buffer = Buffer.concat(chunks);
 		}
+		if (buffer.length > SLICE_A_PAYLOAD_LIMIT_BYTES) {
+			throw new ClientError(
+				`Deploy payload (${buffer.length} bytes) exceeds Slice A's interim ${SLICE_A_PAYLOAD_LIMIT_BYTES} byte cap. ` +
+					`Use a package identifier (npm:/file:/git:) or wait for Slice B's streaming path.`
+			);
+		}
 		hash.update(buffer);
 		byteCount = buffer.length;
 		this.record.payload_blob = createBlob(buffer, { type: 'application/gzip' });

components/operations.js

@@ -390,21 +390,29 @@ async function deployComponent(req) {
 	// observable and auditable even if the CLI disconnects. The row also holds the payload
 	// in a Blob attribute — Slice B will use that for peer delivery; for now it's the
 	// audit record and the rollback source.
-	const recorder = await DeploymentRecorder.create({
-		project: req.project,
-		package_identifier: req.package ?? null,
-		user: req.hdb_user?.username,
-		restart_mode: req.restart === 'rolling' ? 'rolling' : req.restart ? 'immediate' : null,
-	});
-	req._deploymentId = recorder.deploymentId;
+	//
+	// Only the origin node records — peers receiving a replicated deploy_component skip
+	// recording so we don't accumulate one row per node for the same deploy. The row will
+	// reach peers via the table's replication once Slice B has them consume it.
+	const isReplicatedExecution = typeof req._deploymentId === 'string';
+	const recorder = isReplicatedExecution
+		? null
+		: await DeploymentRecorder.create({
+				project: req.project,
+				package_identifier: req.package ?? null,
+				user: req.hdb_user?.username,
+				restart_mode: req.restart === 'rolling' ? 'rolling' : req.restart ? 'immediate' : null,
+			});
+	if (recorder) req._deploymentId = recorder.deploymentId;
 
 	let extractionPayload = req.payload;
 	try {
-		// If a tarball came in (Buffer or Readable from the multipart parser), tee it through
-		// a hash-and-size tap into the row's payload_blob, then re-source extraction from the
-		// persisted blob. This means we read the upload exactly once into local storage; the
-		// blob is the staging area and (in Slice B) the channel peers will replicate from.
-		if (req.payload != null) {
+		// On the origin, tee the tarball (Buffer or Readable from the multipart parser)
+		// through a hash-and-size tap into the row's payload_blob, then re-source extraction
+		// from the persisted blob. This is the staging area and (in Slice B) the channel
+		// peers will replicate from. On peer nodes we skip recording entirely and use the
+		// raw payload as-is.
+		if (recorder && req.payload != null) {
 			await recorder.ingestPayload(req.payload);
 			extractionPayload = recorder.row.payload_blob.stream();
 		}
@@ -462,11 +470,13 @@ async function deployComponent(req) {
 			response.message = `Successfully deployed: ${application.name}, restarting Harper`;
 		} else response.message = `Successfully deployed: ${application.name}`;
 
-		response.deployment_id = recorder.deploymentId;
-		await recorder.finish('success');
+		if (recorder) {
+			response.deployment_id = recorder.deploymentId;
+			await recorder.finish('success');
+		}
 		return response;
 	} catch (err) {
-		await recorder.finish('failed', err);
+		if (recorder) await recorder.finish('failed', err);
 		throw err;
 	}
 }

integrationTests/deploy/deploy-tracking.test.ts

@@ -9,7 +9,7 @@
 import { suite, test, before, after } from 'node:test';
 import { ok, strictEqual } from 'node:assert/strict';
 import { join } from 'node:path';
-import { existsSync, mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'node:fs';
+import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { setTimeout as sleep } from 'node:timers/promises';
 import { request } from 'node:http';
@@ -37,7 +37,7 @@ function postMultipart(
 				headers: {
 					'Content-Type': contentType,
 					'Transfer-Encoding': 'chunked',
-					Authorization: 'Basic ' + Buffer.from(`${auth.username}:${auth.password}`).toString('base64'),
+					'Authorization': 'Basic ' + Buffer.from(`${auth.username}:${auth.password}`).toString('base64'),
 				},
 			},
 			(res) => {
@@ -58,11 +58,10 @@ async function callOperation(
 	op: Record<string, unknown>
 ): Promise<{ status: number; body: any }> {
 	const url = new URL(ctx.harper.operationsAPIURL);
-	const auth =
-		'Basic ' + Buffer.from(`${ctx.harper.admin.username}:${ctx.harper.admin.password}`).toString('base64');
+	const auth = 'Basic ' + Buffer.from(`${ctx.harper.admin.username}:${ctx.harper.admin.password}`).toString('base64');
 	const res = await fetch(url, {
 		method: 'POST',
-		headers: { 'Content-Type': 'application/json', Authorization: auth },
+		headers: { 'Content-Type': 'application/json', 'Authorization': auth },
 		body: JSON.stringify(op),
 	});
 	const text = await res.text();
@@ -133,10 +132,16 @@ suite('Deployment tracking', (ctx: ContextWithHarper) => {
 		strictEqual(row.project, project);
 		strictEqual(row.status, 'success');
 		strictEqual(row.payload_blob_present, true, 'payload_blob should have been persisted');
-		ok(typeof row.payload_hash === 'string' && /^[0-9a-f]{64}$/i.test(row.payload_hash), 'payload_hash should be a sha256 hex string');
+		ok(
+			typeof row.payload_hash === 'string' && /^[0-9a-f]{64}$/i.test(row.payload_hash),
+			'payload_hash should be a sha256 hex string'
+		);
 		ok(typeof row.payload_size === 'number' && row.payload_size > 0, 'payload_size should be a positive integer');
 		ok(typeof row.started_at === 'number' && row.started_at > 0, 'started_at should be set');
-		ok(typeof row.completed_at === 'number' && row.completed_at >= row.started_at, 'completed_at should be >= started_at');
+		ok(
+			typeof row.completed_at === 'number' && row.completed_at >= row.started_at,
+			'completed_at should be >= started_at'
+		);
 	});
 
 	test('list_deployments surfaces the row, supports project filter', async () => {
@@ -148,7 +153,10 @@ suite('Deployment tracking', (ctx: ContextWithHarper) => {
 		ok(deploymentId && ids.includes(deploymentId), `listed deployments should include ${deploymentId}`);
 		// blob bytes must NOT travel back in the list response — only the presence boolean.
 		ok(!('payload_blob' in listed.body.deployments[0]), 'list_deployments must not include payload_blob bytes');
-		ok('payload_blob_present' in listed.body.deployments[0], 'list_deployments should include payload_blob_present flag');
+		ok(
+			'payload_blob_present' in listed.body.deployments[0],
+			'list_deployments should include payload_blob_present flag'
+		);
 	});
 
 	test('a failed deploy is recorded with status=failed and error.message', async () => {
@@ -190,7 +198,10 @@ suite('Deployment tracking', (ctx: ContextWithHarper) => {
 			const failed = listed.body.deployments.find((d: any) => d.project === project);
 			ok(failed, `expected to find a deployment for ${project} in list`);
 			strictEqual(failed.status, 'failed');
-			ok(failed.error && typeof failed.error.message === 'string' && failed.error.message.length > 0, 'failed deployment should have error.message');
+			ok(
+				failed.error && typeof failed.error.message === 'string' && failed.error.message.length > 0,
+				'failed deployment should have error.message'
+			);
 		} finally {
 			try {
 				rmSync(brokenDir, { recursive: true, force: true });

json/systemSchema.json

@@ -374,6 +374,7 @@
 		"hash_attribute": "deployment_id",
 		"name": "hdb_deployment",
 		"schema": "system",
+		"audit": true,
 		"attributes": [
 			{
 				"attribute": "deployment_id"

utility/operation_authorization.ts

@@ -40,6 +40,7 @@ import PermissionResponseObject from '../security/data_objects/PermissionRespons
 import { handleHDBError, hdbErrors } from '../utility/errors/hdbError.ts';
 
 import * as regDeprecated from '../resources/registrationDeprecated.ts';
+import * as deploymentOperations from '../components/deploymentOperations.ts';
 
 const requiredPermissions = new Map();
 const DELETE_PERM = 'delete';
@@ -241,6 +242,14 @@ requiredPermissions.set(functionsOperations.addComponent.name, new (permission a
 requiredPermissions.set(functionsOperations.dropCustomFunctionProject.name, new (permission as any)(true, []));
 requiredPermissions.set(functionsOperations.packageComponent.name, new (permission as any)(true, []));
 requiredPermissions.set(functionsOperations.deployComponent.name, new (permission as any)(true, []));
+requiredPermissions.set(
+	deploymentOperations.handleListDeployments.name,
+	new (permission as any)(true, [], terms.OPERATIONS_ENUM.LIST_DEPLOYMENTS)
+);
+requiredPermissions.set(
+	deploymentOperations.handleGetDeployment.name,
+	new (permission as any)(true, [], terms.OPERATIONS_ENUM.GET_DEPLOYMENT)
+);
 
 //Below are functions that are currently open to all roles
 requiredPermissions.set(regDeprecated.getRegistrationInfo.name, new (permission as any)(false, []));