storage: 服务端存储加固(原子落盘fsync/损坏隔离/墓碑14天GC/空同步不落盘/当日首份快照不可变/每周深快照/大批删除前置快照/health增强)

2026-10-26 20:51:12 +08:00
parent d48330e5c4
commit 1afc236d88
10 changed files with 216 additions and 40 deletions
@@ -11,6 +11,11 @@ const DATA_DIR = process.env.DATA_DIR || path.join(__dirname, 'data')
 const PORT = Number(process.env.PORT || 3780)
 const TOKEN_TTL_MS = 30 * 24 * 60 * 60 * 1000 // 30 天
 const ITER = 150000
+const BACKUP_DAYS = 14            // 每日快照保留天数
+const WEEKLY_KEEP = 8             // 每周深快照保留份数
+const TOMBSTONE_TTL_MS = 14 * 24 * 60 * 60 * 1000 // 墓碑保留 14 天：足够覆盖离线设备，防止无限膨胀与重放
+const LARGE_TOMBSTONE_BACKUP = 20 // 单次同步墓碑达到该数量时，先做数据快照再应用（大批删除的前置保险）
+let lastBackupAt = ''

 const TABLES = ['project', 'work_log', 'log_item', 'attendance', 'weekly_note', 'setting', 'memo', 'credential', 'goal', 'transaction', 'plan']

@@ -35,7 +40,13 @@ function readJson(file, fallback) {
 }
 function writeJson(file, data) {
  const tmp = path.join(DATA_DIR, file + '.tmp')
-  fs.writeFileSync(tmp, JSON.stringify(data))
+  const fd = fs.openSync(tmp, 'w')
+  try {
+    fs.writeSync(fd, JSON.stringify(data))
+    fs.fsyncSync(fd) // 落盘后再原子改名：进程崩溃/断电不会留下半截文件
+  } finally {
+    fs.closeSync(fd)
+  }
  fs.renameSync(tmp, path.join(DATA_DIR, file))
 }

@@ -126,11 +137,37 @@ function isAdminToken(token) {
 function saveRegistry() { writeJson(registryFile, registry); try { doBackup() } catch { /* 忽略 */ } }

 function userFile(userId) { return path.join(DATA_DIR, 'users', userId + '.json') }
+function emptyUserData(userId) {
+  return { userId, cursor: 0, tables: {}, tombstones: {}, created_at: '' }
+}
 function loadUserData(userId) {
-  const data = readJson('users/' + userId + '.json', { userId, cursor: 0, tables: {}, tombstones: {} })
-  // 数据纪元：用户数据文件首次创建时间。目录被误删重建后该值改变，客户端据此识别并全量重推自救
-  if (!data.created_at) data.created_at = new Date().toISOString()
-  return data
+  const full = userFile(userId)
+  if (fs.existsSync(full)) {
+    try {
+      return JSON.parse(fs.readFileSync(full, 'utf-8'))
+    } catch (e) {
+      // 数据文件损坏：隔离保留原件并大声报错，绝不静默当作空数据（否则客户端同步会引发连锁覆盖）
+      const quarantine = full + '.corrupt-' + new Date().toISOString().replace(/[:.]/g, '-')
+      try { fs.renameSync(full, quarantine) } catch { /* 忽略 */ }
+      console.error('[存储] 用户数据文件解析失败，已隔离：' + quarantine + '（' + (e && e.message) + '）')
+      return emptyUserData(userId)
+    }
+  }
+  return emptyUserData(userId)
+}
+// 墓碑回收：超过保留期的墓碑删除（全量快照重建的客户端不依赖过期墓碑也能收敛）
+function gcTombstones(data) {
+  const cutoff = Date.now() - TOMBSTONE_TTL_MS
+  let pruned = 0
+  for (const t of Object.keys(data.tombstones || {})) {
+    const bucket = data.tombstones[t]
+    for (const u of Object.keys(bucket)) {
+      const ts = Date.parse(bucket[u])
+      if (Number.isFinite(ts) && ts < cutoff) { delete bucket[u]; pruned++ }
+    }
+    if (!Object.keys(bucket).length) delete data.tombstones[t]
+  }
+  return pruned
 }
 function saveUserData(userId, data) {
  fs.mkdirSync(path.join(DATA_DIR, 'users'), { recursive: true })
@@ -269,7 +306,15 @@ async function handle(req, res) {
    return
  }

-  if (req.method === 'GET' && route === '/api/health') { json(res, 200, { ok: true, backup: lastOffsiteBackup.state === 'disabled' ? 'local-only' : lastOffsiteBackup }); return }
+  if (req.method === 'GET' && route === '/api/health') {
+    json(res, 200, {
+      ok: true,
+      backup: lastOffsiteBackup.state === 'disabled' ? 'local-only' : lastOffsiteBackup,
+      users: registry.users.length,
+      lastBackup: lastBackupAt
+    })
+    return
+  }

  // SSE 事件流：token 经查询参数传递（EventSource 无法自定义请求头）
  if (req.method === 'GET' && route === '/api/events') {
@@ -359,9 +404,11 @@ async function handle(req, res) {
    if (!auth || !tokenUsable(auth)) { json(res, 401, { error: '未认证' }); return }
    const body = await readBody(req)
    const userData = loadUserData(auth.userId)
-    if (!userData.cursor) userData.cursor = 0
-    if (!userData.tables) userData.tables = {}
-    if (!userData.tombstones) userData.tombstones = {}
+    let dirty = false
+    if (userData.cursor == null) { userData.cursor = 0; dirty = true }
+    if (!userData.tables) { userData.tables = {}; dirty = true }
+    if (!userData.tombstones) { userData.tombstones = {}; dirty = true }
+    if (!userData.created_at) { userData.created_at = new Date().toISOString(); dirty = true } // 数据纪元（客户端灾难自检依据）

    // 应用推送的变更（新者胜 + 自然键去重）
    const incomingTables = body.tables || {}
@@ -411,12 +458,18 @@ async function handle(req, res) {
          userData.tombstones[t][conflictUuid] = now
          changed++
        }
+        // 内容与已存行完全一致：不计变更、不触发保存（空推送不落盘）
+        if (!conflictUuid && existing && JSON.stringify(existing) === JSON.stringify(clean)) continue
        userData.tables[t][row.uuid] = clean
        changed++
      }
    }
    // 应用墓碑
    const incomingTombs = body.tombstones || []
+    if (incomingTombs.length >= LARGE_TOMBSTONE_BACKUP) {
+      // 大批删除的前置保险：先把当前数据快照留存，再应用删除（可追溯、可回滚）
+      try { doBackup(); console.log('[存储] 检测到大批删除（' + incomingTombs.length + ' 条墓碑），已先行快照') } catch { /* 忽略 */ }
+    }
    for (const tb of incomingTombs) {
      if (!TABLES.includes(tb.table) || !tb.uuid) continue
      userData.tombstones[tb.table] = userData.tombstones[tb.table] || {}
@@ -434,7 +487,8 @@ async function handle(req, res) {
        if (row && String(row.updated_at || '') <= String(deletedAt)) delete userData.tables[t][u]
      }
    }
-    saveUserData(auth.userId, userData)
+    if (gcTombstones(userData) > 0) dirty = true
+    if (dirty || changed > 0) saveUserData(auth.userId, userData)
    if (changed) broadcast(auth.userId, 'change')

    // 返回权威全量
@@ -511,27 +565,53 @@ async function handle(req, res) {
 }

 // ---------- 自动备份：服务器是唯一数据源，必须自带兜底 ----------
+function snapshotTo(dir) {
+  fs.mkdirSync(path.join(dir, 'users'), { recursive: true })
+  try { fs.copyFileSync(path.join(DATA_DIR, registryFile), path.join(dir, 'registry.json')) } catch { /* 忽略 */ }
+  try { fs.copyFileSync(path.join(DATA_DIR, 'secret.key'), path.join(dir, 'secret.key')) } catch { /* 忽略 */ }
+  try {
+    const usersDir = path.join(DATA_DIR, 'users')
+    if (fs.existsSync(usersDir)) {
+      for (const f of fs.readdirSync(usersDir)) {
+        if (f.endsWith('.json') && !f.includes('.corrupt-')) fs.copyFileSync(path.join(usersDir, f), path.join(dir, 'users', f))
+      }
+    }
+  } catch { /* 忽略 */ }
+}
+function pruneBackups() {
+  try {
+    const root = path.join(DATA_DIR, 'backups')
+    const entries = fs.readdirSync(root).filter(d => /^\d{4}-\d{2}-\d{2}/.test(d))
+    const days = Array.from(new Set(entries.map(d => d.slice(0, 10)))).sort()
+    while (days.length > BACKUP_DAYS) {
+      const dead = days.shift()
+      for (const d of entries) {
+        if (d.startsWith(dead)) { try { fs.rmSync(path.join(root, d), { recursive: true, force: true }) } catch { /* 忽略 */ } }
+      }
+    }
+  } catch { /* 忽略 */ }
+}
 function doBackup() {
  try {
-    const day = new Date().toISOString().slice(0, 10)
-    const dir = path.join(DATA_DIR, 'backups', day)
-    fs.mkdirSync(path.join(dir, 'users'), { recursive: true })
-    try { fs.copyFileSync(path.join(DATA_DIR, registryFile), path.join(dir, 'registry.json')) } catch { /* 忽略 */ }
-    try { fs.copyFileSync(path.join(DATA_DIR, 'secret.key'), path.join(dir, 'secret.key')) } catch { /* 忽略 */ }
-    try {
-      const usersDir = path.join(DATA_DIR, 'users')
-      if (fs.existsSync(usersDir)) {
-        for (const f of fs.readdirSync(usersDir)) {
-          if (f.endsWith('.json')) fs.copyFileSync(path.join(usersDir, f), path.join(dir, 'users', f))
-        }
-      }
-    } catch { /* 忽略 */ }
-    // 保留最近 14 天
-    try {
-      const backupsRoot = path.join(DATA_DIR, 'backups')
-      const days = fs.readdirSync(backupsRoot).filter(d => /^\d{4}-\d{2}-\d{2}$/.test(d)).sort()
-      while (days.length > 14) { const old = days.shift(); try { fs.rmSync(path.join(backupsRoot, old), { recursive: true, force: true }) } catch { /* 忽略 */ } }
-    } catch { /* 忽略 */ }
+    const now = new Date()
+    const day = now.toISOString().slice(0, 10)
+    const root = path.join(DATA_DIR, 'backups')
+    let dir = path.join(root, day)
+    // 当日首份快照不可变（当日完整证据链）；之后的同日备份落时间戳副本
+    if (fs.existsSync(dir)) dir = path.join(root, day + '_t' + now.toTimeString().slice(0, 8).replace(/:/g, ''))
+    snapshotTo(dir)
+    // 每周一额外做周快照（更深恢复点），保留 8 份
+    if (now.getDay() === 1) {
+      const wdir = path.join(root, 'weekly', day)
+      if (!fs.existsSync(wdir)) snapshotTo(wdir)
+      try {
+        const wroot = path.join(root, 'weekly')
+        const ws = fs.readdirSync(wroot).filter(d => /^\d{4}-\d{2}-\d{2}$/.test(d)).sort()
+        while (ws.length > WEEKLY_KEEP) { const old = ws.shift(); try { fs.rmSync(path.join(wroot, old), { recursive: true, force: true }) } catch { /* 忽略 */ } }
+      } catch { /* 忽略 */ }
+    }
+    pruneBackups()
+    lastBackupAt = new Date().toISOString()
  } catch (e) { console.error('backup failed:', e && e.message) }
  void backupOffsite()
 }