josh/electric-cherry
1
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases Wiki Activity

Deduplicate artifact bytes and deliver files by authenticated hash

Browse Source
This commit is contained in:
josh
2026-10-06 14:35:13 -04:00
parent 18283c63a7
commit 62df14b180
45 changed files with 510 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/hosts/index.ts
+18 -7
View File
@@ -1,3 +1,4 @@
import { packFileJson, unpackFileJson, referenceBlobs } from '../db/file-blobs.js';
import { pruneHistory } from '../retention.js';
import { recordAgentReceipt } from '../provisioning/agent-events.js';
import { EventEmitter } from 'node:events';
@@ -396,7 +397,7 @@ export type HostOperationStatus = Omit<HostOperation, 'payload'>;
const operationStatusColumns = 'o.id,o.host_id,o.kind,o.instance_id,o.generation,o.state,o.result,o.error,o.created_at,o.updated_at';
const operationPayloadSql = 'COALESCE((SELECT p.payload FROM ec_host_operation_payloads p WHERE p.operation_id=o.id),o.payload)';
function operationStatusFromRow(row: Omit<OperationRow, 'payload'>): HostOperationStatus { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; }
function operationFromRow(row: OperationRow): HostOperation { return { ...operationStatusFromRow(row), payload: JSON.parse(row.payload) }; }
function operationFromRow(row: OperationRow): HostOperation { const status = operationStatusFromRow(row); return { ...status, result: unpackFileJson(status.result), payload: unpackFileJson(JSON.parse(row.payload)) }; }
export function getHostOperation(id: string): HostOperation | null {
const row = database().prepare(`SELECT ${operationStatusColumns},${operationPayloadSql} AS payload FROM ec_host_operations o WHERE o.id=?`)
.get(id) as OperationRow | undefined;
@@ -474,20 +475,24 @@ function enqueueHostOperation(hostId: string, kind: string, payload: string | Bu
FROM ec_host_operations o WHERE o.host_id=? AND o.idempotency_key=?`).get(payload, hostId, options.idempotencyKey) as
{ id: string; kind: string; instance_id: string | null; generation: number | null; history_expired_at: string | null; request_hash: string | null; same_payload: number } | undefined;
if (existing) {
const samePayload = existing.history_expired_at ? existing.request_hash === createHash('sha256').update(payload).digest('hex') : !!existing.same_payload;
const samePayload = existing.request_hash ? existing.request_hash === createHash('sha256').update(payload).digest('hex') : !!existing.same_payload;
if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || !samePayload) throw new Error('Operation idempotency key is already associated with a different request.');
return getHostOperationStatus(existing.id)!;
}
}
if (['instance.create','create'].includes(kind)) requireReadyHost(hostId);
if (host.mode === 'disabled' && kind !== 'instance.cancel-create' && !(host.enablingSince && ['runtime.pull','library.download','host.inspect'].includes(kind))) throw new Error('Host is disabled. Enable it before requesting operations.');
const requestHash = createHash('sha256').update(payload).digest('hex');
const packed = packFileJson(JSON.parse(payload.toString()));
payload = JSON.stringify(packed.value);
const id = options.id ?? randomUUID(); const timestamp = now();
const serverName = options.instanceId ? (database().prepare('SELECT server FROM native_instances WHERE id=?')
.get(options.instanceId) as { server: string } | undefined)?.server ?? null : null;
database().prepare(`INSERT INTO ec_host_operations(id,host_id,kind,instance_id,generation,idempotency_key,payload,created_at,updated_at,source,server_name)
VALUES(?,?,?,?,?,?,CAST(? AS TEXT),?,?,?,?)`)
database().prepare(`INSERT INTO ec_host_operations(id,host_id,kind,instance_id,generation,idempotency_key,payload,created_at,updated_at,source,server_name,request_hash)
VALUES(?,?,?,?,?,?,CAST(? AS TEXT),?,?,?,?,?)`)
.run(id, hostId, kind, options.instanceId ?? null, options.generation ?? null, options.idempotencyKey ?? null,
routingPayload ?? payload, timestamp, timestamp, (currentOperationSource() ?? 'Controller').slice(0, 200), serverName);
routingPayload ?? payload, timestamp, timestamp, (currentOperationSource() ?? 'Controller').slice(0, 200), serverName, requestHash);
referenceBlobs('operation', id, packed.hashes);
if (routingPayload !== undefined) database().prepare('INSERT INTO ec_host_operation_payloads(operation_id,payload) VALUES(?,CAST(? AS TEXT))').run(id, payload);
database().prepare(`UPDATE ec_host_operations AS o SET routine=CASE WHEN ${routineObservationSql} THEN 1 ELSE 0 END,
target_path=${operationTargetSql} WHERE o.id=?`).run(id);
@@ -562,7 +567,8 @@ export function getHostOperationPayload(hostId: string, id: string): Buffer {
const row = database().prepare(`SELECT CAST(${operationPayloadSql} AS BLOB) AS payload FROM ec_host_operations o WHERE o.id=? AND o.host_id=?`)
.get(id, hostId) as { payload: Buffer } | undefined;
if (!row) throw new Error('Host operation was not found for this host.');
return row.payload;
if (!row.payload.includes('"contentSha256"') || getHost(hostId)?.inventory.capabilities?.includes('files.blobs')) return row.payload;
return Buffer.from(JSON.stringify(unpackFileJson(JSON.parse(row.payload.toString()))));
}
export function reportHostOperationStatus(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperationStatus {
const operation = database().transaction(() => {
@@ -575,8 +581,13 @@ export function reportHostOperationStatus(hostId: string, id: string, input: { s
const state = input.state as HostOperationState;
if (operation.state === 'running' && state === 'accepted') return operation;
const error = input.error === undefined || input.error === null ? null : String(input.error).slice(0, 4000);
const result = input.result === undefined ? null : JSON.stringify(input.result);
let result = input.result === undefined ? null : JSON.stringify(input.result);
if (result !== null && Buffer.byteLength(result) > 128 * 1024 * 1024) throw new Error('Operation result exceeds the limit.');
if (result !== null) {
const packed = packFileJson(input.result);
result = JSON.stringify(packed.value);
referenceBlobs('operation-result', id, packed.hashes);
}
const observation = state === 'completed' && !error && successfulObservation(operation.kind,
operation.kind === 'rcon.execute' ? JSON.parse(getHostOperationPayload(hostId, id).toString('utf8')) : {}, input.result);
database().prepare('UPDATE ec_host_operations SET state=?,result=?,error=?,updated_at=?,observation_success=? WHERE id=?')
src/lib/server/hosts/schema.ts
+35
View File
@@ -1,3 +1,5 @@
import { createHash } from 'node:crypto';
import { ensureFileBlobSchema, migrationRows, packFileJson, referenceBlobs } from '../db/file-blobs.js';
import type Database from 'better-sqlite3';
import { successfulObservationSql, routineObservationSql, operationTargetSql, importantOperationSql } from './observations.js';
@@ -6,6 +8,7 @@ export function ensureHostSchema(handle: Database.Database): void {
if (initialized.has(handle)) return;
// Background readers open only after controller migrations have completed.
if (handle.readonly) { initialized.add(handle); return; }
ensureFileBlobSchema(handle);
handle.exec(`
CREATE TABLE IF NOT EXISTS ec_hosts (
id TEXT PRIMARY KEY, name TEXT NOT NULL, hostname TEXT NOT NULL DEFAULT '',
@@ -150,5 +153,37 @@ export function ensureHostSchema(handle: Database.Database): void {
handle.exec('UPDATE ec_resource_allowances SET cpu_cores=cpu_cores/1.25, memory_bytes=CAST(memory_bytes/1.25 AS INTEGER)');
})();
}
if (!handle.prepare('SELECT 1 FROM ec_host_schema_migrations WHERE name=?').get('operation-file-blobs')) {
handle.transaction(() => {
for (const row of migrationRows<{ id: string; payload: string; result: string | null }>(handle,
`SELECT o.rowid AS rid,o.id,COALESCE(p.payload,o.payload) payload,o.result
FROM ec_host_operations o LEFT JOIN ec_host_operation_payloads p ON p.operation_id=o.id
WHERE o.rowid>? AND o.kind IN ('files.stage','instance.create','instance.rebuild','instance.restart','instance.start','files.read')
ORDER BY o.rowid LIMIT 1`)) {
const packed = packFileJson(JSON.parse(row.payload), handle);
handle.prepare('UPDATE ec_host_operations SET request_hash=COALESCE(request_hash,?) WHERE id=?')
.run(createHash('sha256').update(row.payload).digest('hex'), row.id);
if (packed.hashes.size) {
handle.prepare(`INSERT INTO ec_host_operation_payloads(operation_id,payload) VALUES(?,?)
ON CONFLICT(operation_id) DO UPDATE SET payload=excluded.payload`).run(row.id, JSON.stringify(packed.value));
handle.prepare("UPDATE ec_host_operations SET payload=json_remove(payload,'$.files') WHERE id=?").run(row.id);
referenceBlobs('operation', row.id, packed.hashes, handle);
}
if (row.result) {
const result = packFileJson(JSON.parse(row.result), handle);
if (result.hashes.size) {
handle.prepare('UPDATE ec_host_operations SET result=? WHERE id=?').run(JSON.stringify(result.value), row.id);
referenceBlobs('operation-result', row.id, result.hashes, handle);
}
}
}
handle.prepare('INSERT INTO ec_host_schema_migrations(name) VALUES(?)').run('operation-file-blobs');
})();
}
handle.exec(`CREATE TRIGGER IF NOT EXISTS release_operation_blobs AFTER DELETE ON ec_host_operations BEGIN
DELETE FROM file_blob_references WHERE owner_type IN ('operation','operation-result') AND owner_id=OLD.id; END;
CREATE TRIGGER IF NOT EXISTS expire_operation_blobs AFTER UPDATE OF history_expired_at ON ec_host_operations
WHEN NEW.history_expired_at IS NOT NULL BEGIN
DELETE FROM file_blob_references WHERE owner_type IN ('operation','operation-result') AND owner_id=NEW.id; END;`);
initialized.add(handle);
}