Merge pull request #1843 from pikasTech/fix/1841-todo-note-oom

修复 Todo Note GitHub 持久化写入 OOM

config/unidesk-host-k8s.yaml

@@ -229,7 +229,7 @@ services:
         cpu: 50m
         memory: 96Mi
       limits:
-        memory: 512Mi
+        memory: 768Mi
 
 runtimeConfig:
   heartBeatTimeoutMs: "30000"

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<string, unknown>).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<string, unknown>)
+      .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<string>();
       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<Record<string, unknown>> {
     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<Record<string, unknown>> {
     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<T>(operation: () => Promise<T>): Promise<T> {
     const next = this.queue.then(operation, operation);
-    this.queue = next.catch(() => undefined);
+    this.queue = next.then(() => undefined, () => undefined);
     return await next;
   }
 

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<string, unknown> = { ok: false, initialized: false };
 let mutationQueue: Promise<unknown> = 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<string, unknown> {
+  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<Buffer> {
+  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<string, unknown> = {}): void {
   const record = { at: new Date().toISOString(), service: "todo-note", level, event, ...detail };
@@ -93,26 +140,33 @@ async function initializeStorage(): Promise<void> {
     };
     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<T>(operation: () => Promise<T>): Promise<T> {
   const next = mutationQueue.then(operation, operation);
-  mutationQueue = next.catch(() => undefined);
+  mutationQueue = next.then(() => undefined, () => undefined);
   return await next;
 }
 
 function recordSuccessfulWrite(snapshot: TodoSnapshot, persisted: Record<string, unknown>): 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<string,
 }
 
 async function mutate<T>(reason: string, operation: () => Promise<T>): Promise<T> {
-  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<Record<string, unknown>> {
   const snapshot = normalizeTodoSnapshot(snapshotValue);
-  return await enqueueMutation(async () => {
-    const before = await repository.exportSnapshot();
-    let imported: TodoSnapshot;
-    let persisted: Record<string, unknown>;
-    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<string, unknown>;
       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<string, unknown> {
     startedAt,
     storage: storageState,
     deploy: deployInfo(),
+    memory: memorySummary(),
   };
 }
 

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<TodoReminderNotification>;
   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<TodoSnapshotEntry>;
+  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<TodoSnapshot>;
   if (record.schemaVersion !== 1 || !Array.isArray(record.instances)) throw new Error("Todo snapshot schemaVersion must be 1");
   const seen = new Set<string>();
   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<TodoSnapshotEntry>;
-    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<Array<Record<string, unknown>>>`
-      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<string, HistoryEntry[]>();
     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<string, unknown> = {}, options: { preserveReminderNotifications?: boolean } = {}): Promise<TodoSnapshot> {
-    const normalized = normalizeTodoSnapshot(snapshot);
+  async listReminderNotifications(): Promise<TodoReminderNotification[]> {
+    const rows = await this.sql<Array<Record<string, unknown>>>`
+      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<string, unknown> = {},
+    options: { preserveReminderNotifications?: boolean; trustedSnapshot?: boolean } = {},
+  ): Promise<TodoSnapshot> {
+    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<string> {
@@ -436,7 +453,7 @@ export class PostgresTodoRepository {
   }
 
   private async insertReminderNotification(tx: postgres.TransactionSql, value: TodoReminderNotification): Promise<void> {
-    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})