feat: complete TASK-WP0-08 logging
This commit is contained in:
@@ -69,6 +69,13 @@ interface InstanceConfiguration {
|
||||
schema_version: 1;
|
||||
}
|
||||
|
||||
export function readConfiguredLocalDataRoot(configFile = defaultInstanceConfigPath()) {
|
||||
const configuration = JSON.parse(readFileSync(configFile, "utf8")) as Record<string, unknown>;
|
||||
const candidate = configuration.data_root ?? configuration.local_data_root;
|
||||
if (typeof candidate !== "string" || !isAbsolute(candidate)) throw new Error("data_root_configuration_invalid");
|
||||
return resolve(candidate);
|
||||
}
|
||||
|
||||
export interface ValidatedReadOnlyAssetRoot {
|
||||
absolute_root: string;
|
||||
ok: true;
|
||||
|
||||
+27
-2
@@ -1,7 +1,10 @@
|
||||
import { resolve } from "node:path";
|
||||
import { join, resolve } from "node:path";
|
||||
|
||||
import { createApp } from "./app.js";
|
||||
import { readBrowserSupportRelease } from "./browser-support.js";
|
||||
import { readConfiguredLocalDataRoot } from "./local-data-root.js";
|
||||
import { ManagedStorage } from "./managed-storage.js";
|
||||
import { StructuredJsonlLogger } from "./structured-log.js";
|
||||
import { attachApiSupervisorControl, initializeApiCredentialClients, receiveApiCredentials } from "./supervisor-channel.js";
|
||||
|
||||
const credentialChannelEnabled = process.argv.includes("--dada-credential-stdin");
|
||||
@@ -21,5 +24,27 @@ const controlPipeIndex = process.argv.indexOf("--dada-control-pipe");
|
||||
if (controlPipeIndex >= 0) {
|
||||
const controlPipe = process.argv[controlPipeIndex + 1];
|
||||
if (!controlPipe) throw new Error("Supervisor control pipe name is required.");
|
||||
attachApiSupervisorControl(controlPipe, () => app.close());
|
||||
let storage: ManagedStorage | undefined;
|
||||
const control = attachApiSupervisorControl(controlPipe, async () => {
|
||||
await app.close();
|
||||
storage?.close();
|
||||
});
|
||||
try {
|
||||
const dataRoot = readConfiguredLocalDataRoot();
|
||||
storage = new ManagedStorage({ dataRoot, databasePath: join(dataRoot, "db", "dada.sqlite3") });
|
||||
const logger = new StructuredJsonlLogger({
|
||||
component: "api",
|
||||
directory: join(dataRoot, "logs", "api"),
|
||||
onWriteFailure: () => {
|
||||
try {
|
||||
storage?.setLogAvailability(false);
|
||||
} finally {
|
||||
control.reportStatus("storage_unavailable");
|
||||
}
|
||||
},
|
||||
});
|
||||
logger.write({ error_category: "none", status_category: "ready" });
|
||||
} catch {
|
||||
control.reportStatus("storage_unavailable");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -134,6 +134,7 @@ export class ManagedStorage {
|
||||
private readonly database: BetterSqlite3.Database;
|
||||
private dataRootWritable = true;
|
||||
private diskSpaceAvailable = true;
|
||||
private logWritable = true;
|
||||
private sqliteWritable = true;
|
||||
private measurementBaselineBytes = 0;
|
||||
private recoveryRequiresRemeasure = false;
|
||||
@@ -241,7 +242,7 @@ export class ManagedStorage {
|
||||
getState(): StorageStateRow {
|
||||
const state = this.database.prepare("SELECT * FROM local_backend_storage_state WHERE singleton = 1").get() as Omit<StorageStateRow, "active_storage_reservations_bytes">;
|
||||
const withReservations = { ...state, active_storage_reservations_bytes: this.activeReservationBytes() };
|
||||
if (!this.dataRootWritable || !this.diskSpaceAvailable || !this.sqliteWritable) {
|
||||
if (!this.dataRootWritable || !this.diskSpaceAvailable || !this.sqliteWritable || !this.logWritable) {
|
||||
return { ...withReservations, storage_status: "unavailable" };
|
||||
}
|
||||
return withReservations;
|
||||
@@ -260,7 +261,7 @@ export class ManagedStorage {
|
||||
if (!this.sqliteWritable) return;
|
||||
const state = this.database.prepare("SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1").get() as { managed_content_bytes: number };
|
||||
const classification = classifyCapacity(state.managed_content_bytes, this.activeReservationBytes());
|
||||
const storageStatus = !this.dataRootWritable || !this.diskSpaceAvailable || this.recoveryRequiresRemeasure
|
||||
const storageStatus = !this.dataRootWritable || !this.diskSpaceAvailable || !this.logWritable || this.recoveryRequiresRemeasure
|
||||
? "unavailable"
|
||||
: classification.storage_status;
|
||||
this.database.prepare(`
|
||||
@@ -291,6 +292,12 @@ export class ManagedStorage {
|
||||
this.refreshState();
|
||||
}
|
||||
|
||||
setLogAvailability(writable: boolean) {
|
||||
this.logWritable = writable;
|
||||
if (!writable) this.recoveryRequiresRemeasure = true;
|
||||
this.refreshState();
|
||||
}
|
||||
|
||||
private physicalManagedBytes() {
|
||||
return ["content", "managed-assets", "derived-assets"]
|
||||
.flatMap((root) => listFiles(resolvePathWithinRoot(this.dataRoot, root)))
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
import {
|
||||
appendFileSync,
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
readdirSync,
|
||||
renameSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
} from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
export const LOG_SIZE_LIMIT_BYTES = 10 * 1024 * 1024;
|
||||
export const LOG_FILE_LIMIT = 10;
|
||||
export const LOG_RETENTION_DAYS = 30;
|
||||
|
||||
const statusCategories = new Set([
|
||||
"starting", "ready", "completed", "failed", "unavailable", "degraded",
|
||||
"blocked", "stopping", "stopped",
|
||||
]);
|
||||
const errorCategories = new Set([
|
||||
"none", "invalid_input", "storage_unavailable", "log_write_failed",
|
||||
"service_unavailable", "timeout", "internal_error",
|
||||
]);
|
||||
const identifierPattern = /^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$/;
|
||||
const codePattern = /^[a-z][a-z0-9_]{0,63}$/;
|
||||
|
||||
export interface StructuredLogInput {
|
||||
correlation_id?: string;
|
||||
duration_ms?: number;
|
||||
error_category?: string;
|
||||
object_id?: string;
|
||||
status_category: string;
|
||||
}
|
||||
|
||||
export interface StructuredJsonlLoggerOptions {
|
||||
component: "api";
|
||||
directory: string;
|
||||
now?: () => Date;
|
||||
onWriteFailure?: () => void;
|
||||
}
|
||||
|
||||
export class LogWriteError extends Error {
|
||||
constructor(options?: ErrorOptions) {
|
||||
super("log_write_failed", options);
|
||||
}
|
||||
}
|
||||
|
||||
export class StructuredJsonlLogger {
|
||||
private readonly component: "api";
|
||||
private readonly directory: string;
|
||||
private readonly now: () => Date;
|
||||
private readonly onWriteFailure: () => void;
|
||||
private initialized = false;
|
||||
private currentSize = 0;
|
||||
|
||||
constructor(options: StructuredJsonlLoggerOptions) {
|
||||
this.component = options.component;
|
||||
this.directory = options.directory;
|
||||
this.now = options.now ?? (() => new Date());
|
||||
this.onWriteFailure = options.onWriteFailure ?? (() => undefined);
|
||||
}
|
||||
|
||||
write(input: StructuredLogInput | Record<string, unknown>) {
|
||||
try {
|
||||
this.initialize();
|
||||
const line = `${JSON.stringify(sanitizeLogEntry(this.component, input, this.now()))}\n`;
|
||||
const bytes = Buffer.byteLength(line);
|
||||
if (bytes > LOG_SIZE_LIMIT_BYTES) throw new Error("log_entry_too_large");
|
||||
if (this.currentSize > 0 && this.currentSize + bytes > LOG_SIZE_LIMIT_BYTES) this.rotate();
|
||||
appendFileSync(this.activePath(), line, { encoding: "utf8" });
|
||||
this.currentSize += bytes;
|
||||
} catch (error) {
|
||||
if (error instanceof LogWriteError) throw error;
|
||||
this.fail(error);
|
||||
}
|
||||
}
|
||||
|
||||
maintain() {
|
||||
try {
|
||||
mkdirSync(this.directory, { recursive: true });
|
||||
this.maintainFiles();
|
||||
this.currentSize = existsSync(this.activePath()) ? statSync(this.activePath()).size : 0;
|
||||
this.initialized = true;
|
||||
} catch (error) {
|
||||
if (error instanceof LogWriteError) throw error;
|
||||
this.fail(error);
|
||||
}
|
||||
}
|
||||
|
||||
private initialize() {
|
||||
if (!this.initialized) this.maintain();
|
||||
}
|
||||
|
||||
private maintainFiles() {
|
||||
const cutoff = this.now().getTime() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000;
|
||||
for (const file of this.files()) {
|
||||
const path = join(this.directory, file.name);
|
||||
if (file.index > LOG_FILE_LIMIT - 1 || statSync(path).mtimeMs < cutoff) rmSync(path, { force: true });
|
||||
}
|
||||
const remaining = this.files();
|
||||
for (const file of remaining.slice(LOG_FILE_LIMIT)) rmSync(join(this.directory, file.name), { force: true });
|
||||
}
|
||||
|
||||
private rotate() {
|
||||
const last = join(this.directory, `${this.component}.${LOG_FILE_LIMIT - 1}.jsonl`);
|
||||
rmSync(last, { force: true });
|
||||
for (let index = LOG_FILE_LIMIT - 2; index >= 1; index -= 1) {
|
||||
const source = join(this.directory, `${this.component}.${index}.jsonl`);
|
||||
if (existsSync(source)) renameSync(source, join(this.directory, `${this.component}.${index + 1}.jsonl`));
|
||||
}
|
||||
if (existsSync(this.activePath())) renameSync(this.activePath(), join(this.directory, `${this.component}.1.jsonl`));
|
||||
this.currentSize = 0;
|
||||
}
|
||||
|
||||
private files() {
|
||||
if (!existsSync(this.directory)) return [];
|
||||
const pattern = new RegExp(`^${this.component}(?:\\.(\\d+))?\\.jsonl$`);
|
||||
return readdirSync(this.directory)
|
||||
.map((name) => ({ match: pattern.exec(name), name }))
|
||||
.filter((entry): entry is { match: RegExpExecArray; name: string } => entry.match !== null)
|
||||
.map(({ match, name }) => ({ index: match[1] === undefined ? 0 : Number(match[1]), name }))
|
||||
.sort((left, right) => left.index - right.index);
|
||||
}
|
||||
|
||||
private activePath() {
|
||||
return join(this.directory, `${this.component}.jsonl`);
|
||||
}
|
||||
|
||||
private fail(error: unknown): never {
|
||||
try {
|
||||
this.onWriteFailure();
|
||||
} catch {
|
||||
// State persistence failure must not replace the stable redacted error.
|
||||
}
|
||||
throw new LogWriteError({ cause: error });
|
||||
}
|
||||
}
|
||||
|
||||
export function sanitizeLogEntry(component: "api", input: StructuredLogInput | Record<string, unknown>, now: Date) {
|
||||
const value = input as Record<string, unknown>;
|
||||
const entry: Record<string, unknown> = {
|
||||
component,
|
||||
schema_version: "1.0",
|
||||
status_category: statusCategories.has(String(value.status_category)) ? value.status_category : "failed",
|
||||
timestamp: now.toISOString(),
|
||||
};
|
||||
if (typeof value.correlation_id === "string" && identifierPattern.test(value.correlation_id)) entry.correlation_id = value.correlation_id;
|
||||
if (typeof value.object_id === "string" && identifierPattern.test(value.object_id)) entry.object_id = value.object_id;
|
||||
if (typeof value.duration_ms === "number" && Number.isSafeInteger(value.duration_ms) && value.duration_ms >= 0) entry.duration_ms = value.duration_ms;
|
||||
if (typeof value.error_category === "string") {
|
||||
entry.error_category = errorCategories.has(value.error_category) ? value.error_category : "internal_error";
|
||||
}
|
||||
return entry;
|
||||
}
|
||||
|
||||
export function sanitizeDiagnosticRecord(input: Record<string, unknown>) {
|
||||
return {
|
||||
app_version: typeof input.app_version === "string" && /^[0-9]+(?:\.[0-9]+){2,3}$/.test(input.app_version) ? input.app_version : "0.0.0",
|
||||
checked_at: typeof input.checked_at === "string" && !Number.isNaN(Date.parse(input.checked_at)) ? input.checked_at : new Date(0).toISOString(),
|
||||
component: input.component === "api" || input.component === "worker" || input.component === "supervisor" ? input.component : "supervisor",
|
||||
message_key: typeof input.message_key === "string" && codePattern.test(input.message_key) ? input.message_key : "diagnostic_redacted",
|
||||
result: input.result === "pass" || input.result === "warning" || input.result === "fail" ? input.result : "fail",
|
||||
stable_check_code: typeof input.stable_check_code === "string" && codePattern.test(input.stable_check_code) ? input.stable_check_code : "invalid_check",
|
||||
};
|
||||
}
|
||||
@@ -33,8 +33,14 @@ export function initializeApiCredentialClients(credentials: Record<(typeof API_C
|
||||
export function attachApiSupervisorControl(pipeName: string, shutdown: () => Promise<void>) {
|
||||
const socket = createConnection(`\\\\.\\pipe\\${pipeName}`);
|
||||
let pending = "";
|
||||
let connected = false;
|
||||
const pendingStatuses: string[] = [];
|
||||
socket.setEncoding("utf8");
|
||||
socket.on("connect", () => socket.write("ready\n"));
|
||||
socket.on("connect", () => {
|
||||
connected = true;
|
||||
socket.write("ready\n");
|
||||
for (const status of pendingStatuses.splice(0)) socket.write(`${status}\n`);
|
||||
});
|
||||
socket.on("data", (chunk) => {
|
||||
pending += chunk;
|
||||
if (!pending.includes("\n")) return;
|
||||
@@ -42,5 +48,11 @@ export function attachApiSupervisorControl(pipeName: string, shutdown: () => Pro
|
||||
pending = "";
|
||||
if (command === "shutdown") void shutdown().finally(() => socket.end());
|
||||
});
|
||||
return socket;
|
||||
return {
|
||||
reportStatus(status: "storage_unavailable") {
|
||||
if (socket.destroyed) return;
|
||||
if (connected) socket.write(`${status}\n`); else pendingStatuses.push(status);
|
||||
},
|
||||
socket,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
import type { StructuredJsonlLogger } from "./structured-log.js";
|
||||
|
||||
export class WorkerAiCallGate {
|
||||
private readonly getStorageStatus: () => "active" | "full" | "unavailable";
|
||||
private readonly logger: StructuredJsonlLogger;
|
||||
|
||||
constructor(input: {
|
||||
getStorageStatus: () => "active" | "full" | "unavailable";
|
||||
logger: StructuredJsonlLogger;
|
||||
}) {
|
||||
this.getStorageStatus = input.getStorageStatus;
|
||||
this.logger = input.logger;
|
||||
}
|
||||
|
||||
async execute<T>(correlationId: string, objectId: string, externalCall: () => Promise<T>) {
|
||||
const initialStatus = this.getStorageStatus();
|
||||
if (initialStatus === "full") return { reason: "storage_full" as const, status: "blocked" as const };
|
||||
if (initialStatus === "unavailable") return { reason: "storage_unavailable" as const, status: "blocked" as const };
|
||||
try {
|
||||
this.logger.write({
|
||||
correlation_id: correlationId,
|
||||
error_category: "none",
|
||||
object_id: objectId,
|
||||
status_category: "starting",
|
||||
});
|
||||
} catch {
|
||||
return { reason: "storage_unavailable" as const, status: "blocked" as const };
|
||||
}
|
||||
const statusAfterLog = this.getStorageStatus();
|
||||
if (statusAfterLog === "full") return { reason: "storage_full" as const, status: "blocked" as const };
|
||||
if (statusAfterLog === "unavailable") return { reason: "storage_unavailable" as const, status: "blocked" as const };
|
||||
const started = Date.now();
|
||||
const value = await externalCall();
|
||||
try {
|
||||
this.logger.write({
|
||||
correlation_id: correlationId,
|
||||
duration_ms: Date.now() - started,
|
||||
error_category: "none",
|
||||
object_id: objectId,
|
||||
status_category: "completed",
|
||||
});
|
||||
} catch {
|
||||
// The completed call cannot be undone; the write failure blocks every later call.
|
||||
}
|
||||
return { status: "completed" as const, value };
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
import { readFileSync } from "node:fs";
|
||||
import { isAbsolute, join, resolve } from "node:path";
|
||||
|
||||
export function readConfiguredLocalDataRoot(environment: NodeJS.ProcessEnv = process.env) {
|
||||
const localAppData = environment.LOCALAPPDATA;
|
||||
if (!localAppData || !isAbsolute(localAppData)) throw new Error("local_app_data_unavailable");
|
||||
const configFile = join(localAppData, "Dada", "P0A", "config", "instance.json");
|
||||
const configuration = JSON.parse(readFileSync(configFile, "utf8")) as Record<string, unknown>;
|
||||
const candidate = configuration.data_root ?? configuration.local_data_root;
|
||||
if (typeof candidate !== "string" || !isAbsolute(candidate)) throw new Error("data_root_configuration_invalid");
|
||||
return resolve(candidate);
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
import { createRequire } from "node:module";
|
||||
|
||||
import type BetterSqlite3 from "better-sqlite3";
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
const Database = require("better-sqlite3") as typeof BetterSqlite3;
|
||||
|
||||
export class WorkerStorageStatus {
|
||||
private readonly database: BetterSqlite3.Database;
|
||||
|
||||
constructor(databasePath: string) {
|
||||
this.database = new Database(databasePath);
|
||||
this.database.pragma("busy_timeout = 5000");
|
||||
}
|
||||
|
||||
getStatus(): "active" | "full" | "unavailable" {
|
||||
const row = this.database.prepare("SELECT storage_status FROM local_backend_storage_state WHERE singleton = 1").get() as { storage_status: "active" | "full" | "unavailable" };
|
||||
return row.storage_status;
|
||||
}
|
||||
|
||||
markLogUnavailable() {
|
||||
this.database.prepare(`
|
||||
UPDATE local_backend_storage_state
|
||||
SET storage_status = 'unavailable', measured_at = ?, version = version + 1
|
||||
WHERE singleton = 1
|
||||
`).run(new Date().toISOString());
|
||||
}
|
||||
|
||||
close() {
|
||||
this.database.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
import {
|
||||
appendFileSync,
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
readdirSync,
|
||||
renameSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
} from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
export const LOG_SIZE_LIMIT_BYTES = 10 * 1024 * 1024;
|
||||
export const LOG_FILE_LIMIT = 10;
|
||||
export const LOG_RETENTION_DAYS = 30;
|
||||
|
||||
const statusCategories = new Set(["starting", "ready", "completed", "failed", "unavailable", "degraded", "blocked", "stopping", "stopped"]);
|
||||
const errorCategories = new Set(["none", "invalid_input", "storage_unavailable", "log_write_failed", "service_unavailable", "timeout", "internal_error"]);
|
||||
const identifierPattern = /^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$/;
|
||||
|
||||
export interface StructuredLogInput {
|
||||
correlation_id?: string;
|
||||
duration_ms?: number;
|
||||
error_category?: string;
|
||||
object_id?: string;
|
||||
status_category: string;
|
||||
}
|
||||
|
||||
export class LogWriteError extends Error {
|
||||
constructor(options?: ErrorOptions) {
|
||||
super("log_write_failed", options);
|
||||
}
|
||||
}
|
||||
|
||||
export class StructuredJsonlLogger {
|
||||
private readonly directory: string;
|
||||
private readonly now: () => Date;
|
||||
private readonly onWriteFailure: () => void;
|
||||
private initialized = false;
|
||||
private currentSize = 0;
|
||||
|
||||
constructor(options: { component: "worker"; directory: string; now?: () => Date; onWriteFailure?: () => void }) {
|
||||
this.directory = options.directory;
|
||||
this.now = options.now ?? (() => new Date());
|
||||
this.onWriteFailure = options.onWriteFailure ?? (() => undefined);
|
||||
}
|
||||
|
||||
write(input: StructuredLogInput | Record<string, unknown>) {
|
||||
try {
|
||||
this.initialize();
|
||||
const line = `${JSON.stringify(sanitizeWorkerLogEntry(input, this.now()))}\n`;
|
||||
const bytes = Buffer.byteLength(line);
|
||||
if (bytes > LOG_SIZE_LIMIT_BYTES) throw new Error("log_entry_too_large");
|
||||
if (this.currentSize > 0 && this.currentSize + bytes > LOG_SIZE_LIMIT_BYTES) this.rotate();
|
||||
appendFileSync(this.activePath(), line, { encoding: "utf8" });
|
||||
this.currentSize += bytes;
|
||||
} catch (error) {
|
||||
if (error instanceof LogWriteError) throw error;
|
||||
this.fail(error);
|
||||
}
|
||||
}
|
||||
|
||||
maintain() {
|
||||
try {
|
||||
mkdirSync(this.directory, { recursive: true });
|
||||
const cutoff = this.now().getTime() - LOG_RETENTION_DAYS * 24 * 60 * 60 * 1000;
|
||||
for (const file of this.files()) {
|
||||
const path = join(this.directory, file.name);
|
||||
if (file.index > LOG_FILE_LIMIT - 1 || statSync(path).mtimeMs < cutoff) rmSync(path, { force: true });
|
||||
}
|
||||
for (const file of this.files().slice(LOG_FILE_LIMIT)) rmSync(join(this.directory, file.name), { force: true });
|
||||
this.currentSize = existsSync(this.activePath()) ? statSync(this.activePath()).size : 0;
|
||||
this.initialized = true;
|
||||
} catch (error) {
|
||||
if (error instanceof LogWriteError) throw error;
|
||||
this.fail(error);
|
||||
}
|
||||
}
|
||||
|
||||
private initialize() {
|
||||
if (!this.initialized) this.maintain();
|
||||
}
|
||||
|
||||
private rotate() {
|
||||
rmSync(join(this.directory, `worker.${LOG_FILE_LIMIT - 1}.jsonl`), { force: true });
|
||||
for (let index = LOG_FILE_LIMIT - 2; index >= 1; index -= 1) {
|
||||
const source = join(this.directory, `worker.${index}.jsonl`);
|
||||
if (existsSync(source)) renameSync(source, join(this.directory, `worker.${index + 1}.jsonl`));
|
||||
}
|
||||
if (existsSync(this.activePath())) renameSync(this.activePath(), join(this.directory, "worker.1.jsonl"));
|
||||
this.currentSize = 0;
|
||||
}
|
||||
|
||||
private files() {
|
||||
if (!existsSync(this.directory)) return [];
|
||||
return readdirSync(this.directory)
|
||||
.map((name) => ({ match: /^worker(?:\.(\d+))?\.jsonl$/.exec(name), name }))
|
||||
.filter((entry): entry is { match: RegExpExecArray; name: string } => entry.match !== null)
|
||||
.map(({ match, name }) => ({ index: match[1] === undefined ? 0 : Number(match[1]), name }))
|
||||
.sort((left, right) => left.index - right.index);
|
||||
}
|
||||
|
||||
private activePath() {
|
||||
return join(this.directory, "worker.jsonl");
|
||||
}
|
||||
|
||||
private fail(error: unknown): never {
|
||||
try {
|
||||
this.onWriteFailure();
|
||||
} catch {
|
||||
// State persistence failure must not replace the stable redacted error.
|
||||
}
|
||||
throw new LogWriteError({ cause: error });
|
||||
}
|
||||
}
|
||||
|
||||
function sanitizeWorkerLogEntry(input: StructuredLogInput | Record<string, unknown>, now: Date) {
|
||||
const value = input as Record<string, unknown>;
|
||||
const entry: Record<string, unknown> = {
|
||||
component: "worker",
|
||||
schema_version: "1.0",
|
||||
status_category: statusCategories.has(String(value.status_category)) ? value.status_category : "failed",
|
||||
timestamp: now.toISOString(),
|
||||
};
|
||||
if (typeof value.correlation_id === "string" && identifierPattern.test(value.correlation_id)) entry.correlation_id = value.correlation_id;
|
||||
if (typeof value.object_id === "string" && identifierPattern.test(value.object_id)) entry.object_id = value.object_id;
|
||||
if (typeof value.duration_ms === "number" && Number.isSafeInteger(value.duration_ms) && value.duration_ms >= 0) entry.duration_ms = value.duration_ms;
|
||||
if (typeof value.error_category === "string") entry.error_category = errorCategories.has(value.error_category) ? value.error_category : "internal_error";
|
||||
return entry;
|
||||
}
|
||||
@@ -33,8 +33,14 @@ export function initializeWorkerCredentialClient(credentials: Record<(typeof WOR
|
||||
export function attachWorkerSupervisorControl(pipeName: string, shutdown: () => Promise<void> | void) {
|
||||
const socket = createConnection(`\\\\.\\pipe\\${pipeName}`);
|
||||
let pending = "";
|
||||
let connected = false;
|
||||
const pendingStatuses: string[] = [];
|
||||
socket.setEncoding("utf8");
|
||||
socket.on("connect", () => socket.write("ready\n"));
|
||||
socket.on("connect", () => {
|
||||
connected = true;
|
||||
socket.write("ready\n");
|
||||
for (const status of pendingStatuses.splice(0)) socket.write(`${status}\n`);
|
||||
});
|
||||
socket.on("data", (chunk) => {
|
||||
pending += chunk;
|
||||
if (!pending.includes("\n")) return;
|
||||
@@ -42,5 +48,11 @@ export function attachWorkerSupervisorControl(pipeName: string, shutdown: () =>
|
||||
pending = "";
|
||||
if (command === "shutdown") void Promise.resolve(shutdown()).finally(() => socket.end());
|
||||
});
|
||||
return socket;
|
||||
return {
|
||||
reportStatus(status: "storage_unavailable") {
|
||||
if (socket.destroyed) return;
|
||||
if (connected) socket.write(`${status}\n`); else pendingStatuses.push(status);
|
||||
},
|
||||
socket,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
import { parentPort } from "node:worker_threads";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { WorkerAiCallGate } from "./ai-call-gate.js";
|
||||
import { readConfiguredLocalDataRoot } from "./runtime-config.js";
|
||||
import { StructuredJsonlLogger } from "./structured-log.js";
|
||||
import { WorkerStorageStatus } from "./storage-status.js";
|
||||
import { attachWorkerSupervisorControl, initializeWorkerCredentialClient, receiveWorkerCredentials } from "./supervisor-channel.js";
|
||||
|
||||
export function handleWorkerProbe(message: unknown) {
|
||||
@@ -20,5 +25,31 @@ if (!workerPort && process.argv.includes("--dada-credential-stdin")) {
|
||||
const controlPipe = process.argv[controlPipeIndex + 1];
|
||||
if (controlPipeIndex < 0 || !controlPipe) throw new Error("Supervisor control pipe name is required.");
|
||||
const keepAlive = setInterval(() => undefined, 30_000);
|
||||
attachWorkerSupervisorControl(controlPipe, () => clearInterval(keepAlive));
|
||||
let storage: WorkerStorageStatus | undefined;
|
||||
const control = attachWorkerSupervisorControl(controlPipe, () => {
|
||||
clearInterval(keepAlive);
|
||||
storage?.close();
|
||||
});
|
||||
let storageStatus: "active" | "unavailable" = "active";
|
||||
try {
|
||||
const dataRoot = readConfiguredLocalDataRoot();
|
||||
storage = new WorkerStorageStatus(join(dataRoot, "db", "dada.sqlite3"));
|
||||
const logger = new StructuredJsonlLogger({
|
||||
component: "worker",
|
||||
directory: join(dataRoot, "logs", "worker"),
|
||||
onWriteFailure: () => {
|
||||
storageStatus = "unavailable";
|
||||
try {
|
||||
storage?.markLogUnavailable();
|
||||
} finally {
|
||||
control.reportStatus("storage_unavailable");
|
||||
}
|
||||
},
|
||||
});
|
||||
logger.write({ error_category: "none", status_category: "ready" });
|
||||
new WorkerAiCallGate({ getStorageStatus: () => storageStatus === "unavailable" ? storageStatus : (storage?.getStatus() ?? "unavailable"), logger });
|
||||
} catch {
|
||||
storageStatus = "unavailable";
|
||||
control.reportStatus("storage_unavailable");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user