diff --git a/config/unidesk-host-k8s.yaml b/config/unidesk-host-k8s.yaml index 42638e68..f7d28024 100644 --- a/config/unidesk-host-k8s.yaml +++ b/config/unidesk-host-k8s.yaml @@ -229,7 +229,7 @@ services: cpu: 50m memory: 96Mi limits: - memory: 512Mi + memory: 768Mi runtimeConfig: heartBeatTimeoutMs: "30000" diff --git a/src/components/microservices/todo-note/src/git-storage.ts b/src/components/microservices/todo-note/src/git-storage.ts index cf380087..aa2dffe0 100644 --- a/src/components/microservices/todo-note/src/git-storage.ts +++ b/src/components/microservices/todo-note/src/git-storage.ts @@ -1,9 +1,9 @@ -import { createHash } from "node:crypto"; +import { createHash, type Hash } from "node:crypto"; import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { dirname, join } from "node:path"; import { spawnSync } from "node:child_process"; -import { normalizeTodoSnapshot, type TodoSnapshot } from "./repository"; +import { normalizeTodoReminderNotification, normalizeTodoSnapshotEntry, type TodoSnapshot } from "./repository"; interface GitResult { ok: boolean; @@ -61,17 +61,47 @@ export function gitStorageConfigFromEnv(): TodoGitStorageConfig { }; } -function stableValue(value: unknown): unknown { - if (Array.isArray(value)) return value.map(stableValue); - if (typeof value === "object" && value !== null) { - return Object.fromEntries(Object.entries(value as Record).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => [key, stableValue(item)])); +function updateStableJson(hash: Hash, input: unknown, arrayValue = false): void { + let value = input; + if (typeof value === "object" && value !== null && typeof (value as { toJSON?: unknown }).toJSON === "function") { + value = (value as { toJSON: () => unknown }).toJSON(); } - return value; + if (Array.isArray(value)) { + hash.update("["); + value.forEach((item, index) => { + if (index > 0) hash.update(","); + updateStableJson(hash, item, true); + }); + hash.update("]"); + return; + } + if (typeof value === "object" && value !== null) { + hash.update("{"); + const entries = Object.entries(value as Record) + .filter(([, item]) => item !== undefined && typeof item !== "function" && typeof item !== "symbol") + .sort(([left], [right]) => left.localeCompare(right)); + entries.forEach(([key, item], index) => { + if (index > 0) hash.update(","); + hash.update(JSON.stringify(key)); + hash.update(":"); + updateStableJson(hash, item); + }); + hash.update("}"); + return; + } + const encoded = JSON.stringify(value); + if (encoded === undefined) { + if (arrayValue) hash.update("null"); + return; + } + hash.update(encoded); } export function snapshotHash(snapshot: TodoSnapshot): string { const comparable = { schemaVersion: snapshot.schemaVersion, instances: snapshot.instances, reminderNotifications: snapshot.reminderNotifications }; - return createHash("sha256").update(JSON.stringify(stableValue(comparable))).digest("hex"); + const hash = createHash("sha256"); + updateStableJson(hash, comparable); + return hash.digest("hex"); } function safeId(value: string): string { @@ -110,6 +140,7 @@ export class TodoGitStorage { if (!existsSync(indexPath)) return null; const index = JSON.parse(readFileSync(indexPath, "utf8")) as { schemaVersion?: unknown; instances?: unknown; reminderNotificationsPath?: unknown }; if (index.schemaVersion !== 1 || !Array.isArray(index.instances)) throw new GitStorageError("Todo Git index has an unsupported schema"); + const seenInstanceIds = new Set(); const instances = index.instances.map((raw) => { if (typeof raw !== "object" || raw === null || Array.isArray(raw)) throw new GitStorageError("Todo Git index entry is invalid"); const entry = raw as { id?: unknown; instancePath?: unknown; historyPath?: unknown; source?: unknown }; @@ -128,32 +159,39 @@ export class TodoGitStorage { } const instance = JSON.parse(readFileSync(join(this.config.worktreePath, instancePath), "utf8")); const historyFile = JSON.parse(readFileSync(join(this.config.worktreePath, historyPath), "utf8")) as { history?: unknown }; - return { instance, history: historyFile.history, source: entry.source }; + const normalized = normalizeTodoSnapshotEntry({ instance, history: historyFile.history, source: entry.source }); + if (seenInstanceIds.has(normalized.instance.id)) throw new GitStorageError("Todo Git index contains a duplicate instance", { instanceId: normalized.instance.id }); + seenInstanceIds.add(normalized.instance.id); + return normalized; }); const reminderNotificationsPath = join(this.config.basePath, "reminders", "notifications.json"); - let reminderNotifications: unknown[] = []; + let reminderNotifications: TodoSnapshot["reminderNotifications"] = []; if (index.reminderNotificationsPath !== undefined) { if (index.reminderNotificationsPath !== reminderNotificationsPath) { throw new GitStorageError("Todo Git index points outside its managed reminder notification path", { reminderNotificationsPath: index.reminderNotificationsPath }); } const reminderFile = JSON.parse(readFileSync(join(this.config.worktreePath, reminderNotificationsPath), "utf8")) as { notifications?: unknown }; if (!Array.isArray(reminderFile.notifications)) throw new GitStorageError("Todo Git reminder notification file is invalid"); - reminderNotifications = reminderFile.notifications; + reminderNotifications = reminderFile.notifications.map(normalizeTodoReminderNotification); } - return normalizeTodoSnapshot({ + return { schemaVersion: 1, generatedAt: typeof (index as { generatedAt?: unknown }).generatedAt === "string" ? (index as { generatedAt: string }).generatedAt : new Date().toISOString(), instances, reminderNotifications, - }); + }; }); } async writeSnapshot(snapshot: TodoSnapshot, reason: string): Promise> { if (!this.enabled()) return { ok: true, changed: false, primary: "postgres" }; - const normalized = normalizeTodoSnapshot(snapshot); + if (snapshot.schemaVersion !== 1 || !Array.isArray(snapshot.instances) || !Array.isArray(snapshot.reminderNotifications)) { + throw new GitStorageError("Todo snapshot has an unsupported schema"); + } + const normalized = snapshot; return await this.enqueue(async () => { this.syncWorktree(); + const normalizedHash = snapshotHash(normalized); const root = this.path(); rmSync(root, { recursive: true, force: true }); mkdirSync(root, { recursive: true }); @@ -185,13 +223,13 @@ export class TodoGitStorage { this.writeJson(join(this.config.basePath, "indexes", "instances.json"), { schemaVersion: 1, generatedAt: normalized.generatedAt, - hash: snapshotHash(normalized), + hash: normalizedHash, reminderNotificationsPath, instances: indexEntries, }); this.runGit(["add", "--all", "--", this.config.basePath]); const diff = this.runGit(["diff", "--cached", "--quiet"], { allowFailure: true }); - if (diff.exitCode === 0) return { ok: true, changed: false, hash: snapshotHash(normalized), head: this.headSummary() }; + if (diff.exitCode === 0) return { ok: true, changed: false, hash: normalizedHash, head: this.headSummary() }; const commit = this.runGit(["commit", "-m", `${this.config.commitMessagePrefix} ${reason}`], { env: { GIT_AUTHOR_NAME: this.config.authorName, @@ -203,7 +241,7 @@ export class TodoGitStorage { if (!commit.ok) throw new GitStorageError("Todo Git commit failed", { stderr: commit.stderr.slice(-2000) }); for (let attempt = 1; attempt <= 3; attempt += 1) { const pushed = this.runGit(["push", "origin", this.config.branch], { allowFailure: true }); - if (pushed.ok) return { ok: true, changed: true, hash: snapshotHash(normalized), attempts: attempt, head: this.headSummary() }; + if (pushed.ok) return { ok: true, changed: true, hash: normalizedHash, attempts: attempt, head: this.headSummary() }; const fetched = this.runGit(["fetch", "origin", this.config.branch], { allowFailure: true }); if (!fetched.ok) continue; const rebased = this.runGit(["rebase", `origin/${this.config.branch}`], { allowFailure: true }); @@ -219,7 +257,7 @@ export class TodoGitStorage { async verify(databaseSnapshot: TodoSnapshot): Promise> { if (!this.enabled()) return { ok: true, primary: "postgres", databaseHash: snapshotHash(databaseSnapshot), gitHash: null }; const gitSnapshot = await this.readSnapshot(); - const databaseHash = snapshotHash(normalizeTodoSnapshot(databaseSnapshot)); + const databaseHash = snapshotHash(databaseSnapshot); const gitHash = gitSnapshot === null ? null : snapshotHash(gitSnapshot); return { ok: gitHash === databaseHash, @@ -234,7 +272,7 @@ export class TodoGitStorage { private async enqueue(operation: () => Promise): Promise { const next = this.queue.then(operation, operation); - this.queue = next.catch(() => undefined); + this.queue = next.then(() => undefined, () => undefined); return await next; } diff --git a/src/components/microservices/todo-note/src/index.ts b/src/components/microservices/todo-note/src/index.ts index 6dac1824..3781cdb5 100644 --- a/src/components/microservices/todo-note/src/index.ts +++ b/src/components/microservices/todo-note/src/index.ts @@ -1,5 +1,6 @@ import express from "express"; -import { gunzipSync } from "node:zlib"; +import { readFileSync } from "node:fs"; +import { gunzipSync, gzipSync } from "node:zlib"; import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../../shared/src/rotating-jsonl"; import { GitStorageError, TodoGitStorage, gitStorageConfigFromEnv, snapshotHash } from "./git-storage"; @@ -51,6 +52,52 @@ const logWriter = createHourlyJsonlWriter({ }); let storageState: Record = { ok: false, initialized: false }; let mutationQueue: Promise = Promise.resolve(); +let garbageCollectionScheduled = false; + +function cgroupMemoryBytes(path: string): number | null { + try { + const raw = readFileSync(path, "utf8").trim(); + if (raw === "max") return null; + const value = Number(raw); + return Number.isFinite(value) && value >= 0 ? value : null; + } catch { + return null; + } +} + +function memorySummary(): Record { + const usage = process.memoryUsage(); + return { + rssBytes: usage.rss, + heapUsedBytes: usage.heapUsed, + heapTotalBytes: usage.heapTotal, + externalBytes: usage.external, + arrayBuffersBytes: usage.arrayBuffers, + cgroup: { + currentBytes: cgroupMemoryBytes("/sys/fs/cgroup/memory.current"), + peakBytes: cgroupMemoryBytes("/sys/fs/cgroup/memory.peak"), + limitBytes: cgroupMemoryBytes("/sys/fs/cgroup/memory.max"), + }, + }; +} + +function scheduleGarbageCollection(): void { + if (garbageCollectionScheduled) return; + garbageCollectionScheduled = true; + setTimeout(() => { + garbageCollectionScheduled = false; + Bun.gc(true); + }, 0); +} + +async function captureRollbackSnapshot(): Promise { + const snapshot = await repository.exportSnapshot(); + return gzipSync(JSON.stringify(snapshot), { level: 1 }); +} + +function restoreRollbackSnapshot(compressed: Buffer): TodoSnapshot { + return normalizeTodoSnapshot(JSON.parse(gunzipSync(compressed).toString("utf8")) as unknown); +} function log(level: LogLevel, event: string, detail: Record = {}): void { const record = { at: new Date().toISOString(), service: "todo-note", level, event, ...detail }; @@ -93,26 +140,33 @@ async function initializeStorage(): Promise { }; return; } - const databaseSnapshot = await repository.exportSnapshot(); + const databaseReminderNotifications = await repository.listReminderNotifications(); const gitSnapshot = await gitStorage.readSnapshot(); if (gitSnapshot === null) { + const databaseSnapshot = await repository.exportSnapshot(); const result = await gitStorage.writeSnapshot(databaseSnapshot, "initialize storage"); + recordSuccessfulWrite(databaseSnapshot, result); log("info", "github_storage_initialized", { instances: databaseSnapshot.instances.length, result }); } else { const mergedSnapshot = { ...gitSnapshot, generatedAt: new Date().toISOString(), - reminderNotifications: mergeReminderNotifications(gitSnapshot.reminderNotifications, databaseSnapshot.reminderNotifications), + reminderNotifications: mergeReminderNotifications(gitSnapshot.reminderNotifications, databaseReminderNotifications), }; - await repository.replaceSnapshot(mergedSnapshot); - if (snapshotHash(mergedSnapshot) !== snapshotHash(gitSnapshot)) await gitStorage.writeSnapshot(mergedSnapshot, "merge reminder notifications"); + const hydratedSnapshot = await repository.replaceSnapshot(mergedSnapshot, {}, { trustedSnapshot: true }); + const mergedHash = snapshotHash(mergedSnapshot); + const persisted = mergedHash === snapshotHash(gitSnapshot) + ? { ok: true, changed: false, primary: "github-repo", hash: mergedHash, head: gitStorage.summary().head } + : await gitStorage.writeSnapshot(mergedSnapshot, "merge reminder notifications"); + recordSuccessfulWrite(hydratedSnapshot, persisted); log("info", "github_storage_hydrated", { instances: mergedSnapshot.instances.length, reminderNotifications: mergedSnapshot.reminderNotifications.length, - hash: snapshotHash(mergedSnapshot), + hash: mergedHash, + memory: memorySummary(), }); } - await refreshStorageState(); + scheduleGarbageCollection(); } function mergeReminderNotifications(gitValues: TodoReminderNotification[], databaseValues: TodoReminderNotification[]): TodoReminderNotification[] { @@ -123,12 +177,12 @@ function mergeReminderNotifications(gitValues: TodoReminderNotification[], datab async function enqueueMutation(operation: () => Promise): Promise { const next = mutationQueue.then(operation, operation); - mutationQueue = next.catch(() => undefined); + mutationQueue = next.then(() => undefined, () => undefined); return await next; } function recordSuccessfulWrite(snapshot: TodoSnapshot, persisted: Record): void { - const hash = snapshotHash(snapshot); + const hash = typeof persisted.hash === "string" ? persisted.hash : snapshotHash(snapshot); const config = gitStorage.summary(); storageState = { ok: true, @@ -149,60 +203,79 @@ function recordSuccessfulWrite(snapshot: TodoSnapshot, persisted: Record(reason: string, operation: () => Promise): Promise { - return await enqueueMutation(async () => { - if (!gitStorage.enabled()) return await operation(); - const before = await repository.exportSnapshot(); - const result = await operation(); - try { - const after = await repository.exportSnapshot(); - const persisted = await gitStorage.writeSnapshot(after, reason); - recordSuccessfulWrite(after, persisted); - return result; - } catch (error) { + try { + return await enqueueMutation(async () => { + if (!gitStorage.enabled()) return await operation(); + const rollbackSnapshot = await captureRollbackSnapshot(); + Bun.gc(true); + const result = await operation(); try { - await repository.replaceSnapshot(before, {}, { preserveReminderNotifications: true }); - } catch (rollbackError) { - log("error", "database_rollback_failed", { reason, error: errorDetail(rollbackError) }); + const after = await repository.exportSnapshot(); + const persisted = await gitStorage.writeSnapshot(after, reason); + recordSuccessfulWrite(after, persisted); + log("info", "github_storage_write_complete", { + reason, + instances: after.instances.length, + historyEntries: after.instances.reduce((total, entry) => total + entry.history.length, 0), + memory: memorySummary(), + }); + return result; + } catch (error) { + try { + await repository.replaceSnapshot(restoreRollbackSnapshot(rollbackSnapshot), {}, { + preserveReminderNotifications: true, + trustedSnapshot: true, + }); + } catch (rollbackError) { + log("error", "database_rollback_failed", { reason, error: errorDetail(rollbackError) }); + } + storageState = { + ok: false, + initialized: true, + checkedAt: new Date().toISOString(), + config: gitStorage.summary(), + error: errorDetail(error), + }; + throw error; } - storageState = { - ok: false, - initialized: true, - checkedAt: new Date().toISOString(), - config: gitStorage.summary(), - error: errorDetail(error), - }; - throw error; - } - }); + }); + } finally { + scheduleGarbageCollection(); + } } async function replaceFromImport(snapshotValue: unknown): Promise> { const snapshot = normalizeTodoSnapshot(snapshotValue); - return await enqueueMutation(async () => { - const before = await repository.exportSnapshot(); - let imported: TodoSnapshot; - let persisted: Record; - try { - imported = await repository.replaceSnapshot(snapshot); - persisted = await gitStorage.writeSnapshot(imported, "migrate from CC01"); - } catch (error) { + try { + return await enqueueMutation(async () => { + const rollbackSnapshot = await captureRollbackSnapshot(); + Bun.gc(true); + let imported: TodoSnapshot; + let persisted: Record; try { - await repository.replaceSnapshot(before); - } catch (rollbackError) { - log("error", "database_rollback_failed", { reason: "migrate from CC01", error: errorDetail(rollbackError) }); + imported = await repository.replaceSnapshot(snapshot, {}, { trustedSnapshot: true }); + persisted = await gitStorage.writeSnapshot(imported, "migrate from CC01"); + } catch (error) { + try { + await repository.replaceSnapshot(restoreRollbackSnapshot(rollbackSnapshot), {}, { trustedSnapshot: true }); + } catch (rollbackError) { + log("error", "database_rollback_failed", { reason: "migrate from CC01", error: errorDetail(rollbackError) }); + } + throw error; } - throw error; - } - recordSuccessfulWrite(imported, persisted); - return { - ok: true, - instances: imported.instances.length, - historyEntries: imported.instances.reduce((total, entry) => total + entry.history.length, 0), - reminderNotifications: imported.reminderNotifications.length, - hash: snapshotHash(imported), - persisted, - }; - }); + recordSuccessfulWrite(imported, persisted); + return { + ok: true, + instances: imported.instances.length, + historyEntries: imported.instances.reduce((total, entry) => total + entry.history.length, 0), + reminderNotifications: imported.reminderNotifications.length, + hash: typeof persisted.hash === "string" ? persisted.hash : snapshotHash(imported), + persisted, + }; + }); + } finally { + scheduleGarbageCollection(); + } } function importSnapshotValue(body: unknown): unknown { @@ -236,6 +309,7 @@ function healthPayload(): Record { startedAt, storage: storageState, deploy: deployInfo(), + memory: memorySummary(), }; } diff --git a/src/components/microservices/todo-note/src/repository.ts b/src/components/microservices/todo-note/src/repository.ts index dcee47ff..9204b5f0 100644 --- a/src/components/microservices/todo-note/src/repository.ts +++ b/src/components/microservices/todo-note/src/repository.ts @@ -94,7 +94,7 @@ function normalizeHistoryEntry(value: unknown): HistoryEntry { }; } -function normalizeReminderNotification(value: unknown): TodoReminderNotification { +export function normalizeTodoReminderNotification(value: unknown): TodoReminderNotification { if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error("Todo reminder notification is not an object"); const record = value as Partial; for (const key of ["instanceId", "todoId", "reminderAt", "target", "title", "message", "responsePreview", "sentAt"] as const) { @@ -106,28 +106,33 @@ function normalizeReminderNotification(value: unknown): TodoReminderNotification return record as TodoReminderNotification; } +export function normalizeTodoSnapshotEntry(value: unknown): TodoSnapshotEntry { + if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error("Todo snapshot entry is not an object"); + const entry = value as Partial; + return { + instance: normalizeInstance(entry.instance), + history: Array.isArray(entry.history) ? entry.history.map(normalizeHistoryEntry) : [], + source: asRecord(entry.source), + }; +} + export function normalizeTodoSnapshot(value: unknown): TodoSnapshot { if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error("Todo snapshot is not an object"); const record = value as Partial; if (record.schemaVersion !== 1 || !Array.isArray(record.instances)) throw new Error("Todo snapshot schemaVersion must be 1"); const seen = new Set(); const instances = record.instances.map((rawEntry) => { - if (typeof rawEntry !== "object" || rawEntry === null || Array.isArray(rawEntry)) throw new Error("Todo snapshot entry is not an object"); - const entry = rawEntry as Partial; - const instance = normalizeInstance(entry.instance); + const entry = normalizeTodoSnapshotEntry(rawEntry); + const instance = entry.instance; if (seen.has(instance.id)) throw new Error(`Duplicate Todo instance: ${instance.id}`); seen.add(instance.id); - return { - instance, - history: Array.isArray(entry.history) ? entry.history.map(normalizeHistoryEntry) : [], - source: asRecord(entry.source), - }; + return entry; }); return { schemaVersion: 1, generatedAt: typeof record.generatedAt === "string" ? record.generatedAt : nowIso(), instances, - reminderNotifications: Array.isArray(record.reminderNotifications) ? record.reminderNotifications.map(normalizeReminderNotification) : [], + reminderNotifications: Array.isArray(record.reminderNotifications) ? record.reminderNotifications.map(normalizeTodoReminderNotification) : [], }; } @@ -312,11 +317,6 @@ export class PostgresTodoRepository { FROM todo_note_history ORDER BY instance_id ASC, id ASC `; - const reminderRows = await this.sql>>` - SELECT instance_id, todo_id, reminder_at, target, title, message, response_preview, sent_at::text AS sent_at - FROM todo_note_reminder_notifications - ORDER BY id ASC - `; const historyByInstance = new Map(); for (const row of historyRows) { const instanceId = String(row.instance_id); @@ -339,21 +339,37 @@ export class PostgresTodoRepository { history: historyByInstance.get(String(row.id)) ?? [], source: asRecord(row.source), })), - reminderNotifications: reminderRows.map((row) => normalizeReminderNotification({ - instanceId: row.instance_id, - todoId: row.todo_id, - reminderAt: row.reminder_at, - target: row.target, - title: row.title, - message: row.message, - responsePreview: row.response_preview, - sentAt: row.sent_at, - })), + reminderNotifications: await this.listReminderNotifications(), }; } - async replaceSnapshot(snapshot: unknown, source: Record = {}, options: { preserveReminderNotifications?: boolean } = {}): Promise { - const normalized = normalizeTodoSnapshot(snapshot); + async listReminderNotifications(): Promise { + const rows = await this.sql>>` + SELECT instance_id, todo_id, reminder_at, target, title, message, response_preview, sent_at::text AS sent_at + FROM todo_note_reminder_notifications + ORDER BY id ASC + `; + return rows.map((row) => normalizeTodoReminderNotification({ + instanceId: row.instance_id, + todoId: row.todo_id, + reminderAt: row.reminder_at, + target: row.target, + title: row.title, + message: row.message, + responsePreview: row.response_preview, + sentAt: row.sent_at, + })); + } + + async replaceSnapshot( + snapshot: unknown, + source: Record = {}, + options: { preserveReminderNotifications?: boolean; trustedSnapshot?: boolean } = {}, + ): Promise { + const normalized = options.trustedSnapshot === true ? snapshot as TodoSnapshot : normalizeTodoSnapshot(snapshot); + if (normalized.schemaVersion !== 1 || !Array.isArray(normalized.instances) || !Array.isArray(normalized.reminderNotifications)) { + throw new Error("Trusted Todo snapshot has an unsupported schema"); + } await this.sql.begin(async (tx) => { await tx`DELETE FROM todo_note_history`; await tx`DELETE FROM todo_note_instances`; @@ -375,7 +391,8 @@ export class PostgresTodoRepository { for (const notification of normalized.reminderNotifications) await this.insertReminderNotification(tx, notification); } }); - return await this.exportSnapshot(); + if (options.preserveReminderNotifications === true) return await this.exportSnapshot(); + return normalized; } async readLogs(): Promise { @@ -436,7 +453,7 @@ export class PostgresTodoRepository { } private async insertReminderNotification(tx: postgres.TransactionSql, value: TodoReminderNotification): Promise { - const notification = normalizeReminderNotification(value); + const notification = normalizeTodoReminderNotification(value); await tx` INSERT INTO todo_note_reminder_notifications (instance_id, todo_id, reminder_at, target, title, message, response_preview, sent_at) VALUES (${notification.instanceId}, ${notification.todoId}, ${notification.reminderAt}, ${notification.target}, ${notification.title}, ${notification.message}, ${notification.responsePreview}, ${notification.sentAt})