diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 8604b6bdc..94f25be03 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -227,6 +227,14 @@ - 验证方式：`npm run check:database-backup`、`npm run check:production-ops`、`npm run check:encoding`、`git diff --check`；dev 现场必须完成逐文件 full catalog、重复 full 零 PUT、history dry-run、上传后清理、STDB 重启和按 catalog 隔离恢复 roundtrip。 - 关联：。 +## 2026-07-27 逐文件备份本地元数据采用去重 state 与 gzip 保留 + +- 背景：files v1 本地 state 同时在 `baselineCatalog`、`latestCatalog` 和每个 `historyCatalogs[]` 中嵌入完整文件清单，且每次成功 history 的本地 catalog 与 `--result-file` 再复制同一清单；release 独立 work-dir 已由此累积约 840 MiB JSON，但 OSS-only 恢复实际只依赖 `latest.json`、catalog 引用和 CAS 对象。 +- 决策：OSS catalog/latest schema、对象 key、序列化字节与恢复链保持不变。本地 state 升级为 gzip v2，只保存去重后的 catalog 引用；旧 v1 JSON 可读，并且只在非 dry-run 成功发布、验真 latest、原子写入 v2 后删除。full 增量复用从本地 latest full catalog gzip 缓存读取，缓存缺失时退化为逐对象 OSS HEAD，不影响正确性；缓存长度或 SHA 与 state 不一致时失败关闭。 +- 清理边界：成功运行后只压缩保留 latest full catalog；已验真的本地 history catalog、旧 full catalog和严格文件名匹配的失败/dry-run 遗留 catalog 清理。`--result-file` 只写紧凑引用和计数。任一 state 压缩或本地 metadata 清理失败都发生在 `/stdb` history 源文件删除之前；OSS catalog、latest 和 CAS 对象永不由本地 metadata 清理删除。 +- 兼容与验证：`--restore-files-state` 同时接受 v1 JSON 与 v2 gzip，`--restore-files-latest` 不受本地格式影响。门禁覆盖 v1 迁移、gzip/state/catalog 损坏拒绝、full 增量复用、history catalog 本地清理、紧凑 result、pointer 失败不清源和 OSS-only 恢复。 +- 关联：`scripts/database-backup-to-oss.mjs`、`scripts/check-database-backup-to-oss.mjs`、`docs/【开发运维】本地开发验证与生产运维-2026-05-15.md`。 + ## 2026-07-14 后台账号采用 owner 引导账号与一级 Tab 实时授权 - 背景：后台此前只支持一组环境变量管理员，所有 `/admin/api/*` 共用统一 admin 门禁，无法给运营、审核等人员分配独立账号和页面范围。 diff --git a/docs/project-memory/shared-memory/pitfalls.md b/docs/project-memory/shared-memory/pitfalls.md index 94fe87ed5..cb4dbfd78 100644 --- a/docs/project-memory/shared-memory/pitfalls.md +++ b/docs/project-memory/shared-memory/pitfalls.md @@ -3239,8 +3239,8 @@ - 现象：看到最新 `N.snapshot_dir` 后，把所有起始 offset 小于 `N` 的 `.stdb.log` 删除，或者只把旧日志上传 OSS 就宣称已有完整增量灾备。 - 原因：segment 文件名只表示该段最早事务；起始 offset 小于等于最新 snapshot 的最后一个 segment 可能跨越 snapshot 边界，重启仍需要它。历史归档也不会及时覆盖 control-db、program bytes、最新 snapshot 和 active segment。 -- 处理：latest snapshot 必须是未锁定且存在同 offset `.snapshot_bsatn` 的完整目录，空目录或同名 `.lock` 存在时忽略。每个 replica 独立保留 `max(segment_start <= latest_snapshot)` 及全部后缀，只处理更早 segment 对；旧 snapshot 只保留最新一个。`--storage-format files` 必须先发布完整 full catalog；history 对每个候选文件 CAS 对象、history catalog 和 full catalog 执行 HEAD 长度/SHA 验真，再复算边界与 stat fingerprint，最后发布并验真固定 `latest.json`；pointer 失败时不得推进 state 或删除源文件。不要把在线逐文件 full 扫描当成跨文件一致备份，基线必须来自停库目录或已验证冻结副本。SSH 或工具超时后先检查 work-dir PID lock 与原进程，不要直接并发重跑；不要在 history 模式传 `--stop-service`。定时任务通过 Server-Provision 的显式 profile 和仓库 drop-in 管理，启用前 dry-run 验证 baseline，切回 archive 时同时移除托管与现场遗留 drop-in；不要在 `/etc/systemd/system` 长期保留手写覆盖，release 必须建立和验证自己的 full baseline 与 work-dir，不能直接复用 dev 的本地 state。 -- 验证：dry-run 输出 replica 的 `latestSnapshot`、`boundarySegment` 和候选清单；从另一台机器仅凭 OSS `latest.json` 自动定位 full catalog，创建目录、下载文件并逐项校验长度/SHA，启动隔离 data-dir 验证 `/v1/ping`、snapshot restore、commitlog replay、module launch、代表性 SQL 与 reducer。 +- 处理：latest snapshot 必须是未锁定且存在同 offset `.snapshot_bsatn` 的完整目录，空目录或同名 `.lock` 存在时忽略。每个 replica 独立保留 `max(segment_start <= latest_snapshot)` 及全部后缀，只处理更早 segment 对；旧 snapshot 只保留最新一个。`--storage-format files` 必须先发布完整 full catalog；history 对每个候选文件 CAS 对象、history catalog 和 full catalog 执行 HEAD 长度/SHA 验真，再复算边界与 stat fingerprint，最后发布并验真固定 `latest.json`；pointer 失败时不得推进 state 或删除源文件。不要把在线逐文件 full 扫描当成跨文件一致备份，基线必须来自停库目录或已验证冻结副本。SSH 或工具超时后先检查 work-dir PID lock 与原进程，不要直接并发重跑；不要在 history 模式传 `--stop-service`。定时任务通过 Server-Provision 的显式 profile 和仓库 drop-in 管理，启用前 dry-run 验证 baseline，切回 archive 时同时移除托管与现场遗留 drop-in；不要在 `/etc/systemd/system` 长期保留手写覆盖，release 必须建立和验证自己的 full baseline 与 work-dir，不能直接复用 dev 的本地 state。files 本地 state 只能保存去重后的 catalog 引用并使用 gzip 原子落盘；本地只保留 latest full catalog 压缩缓存，history/旧 full catalog 和紧凑 result 不得再次复制完整清单。metadata 压缩或清理失败必须早于 `/stdb` history 源文件删除。 +- 验证：dry-run 输出 replica 的 `latestSnapshot`、`boundarySegment` 和候选清单；从另一台机器仅凭 OSS `latest.json` 自动定位 full catalog，创建目录、下载文件并逐项校验长度/SHA，启动隔离 data-dir 验证 `/v1/ping`、snapshot restore、commitlog replay、module launch、代表性 SQL 与 reducer。备份门禁还必须覆盖 v1 JSON 到 v2 gzip 迁移、损坏 gzip 不回退、full 增量复用、本地 catalog SHA 校验、history catalog 清理与紧凑 result。 - 关联：`scripts/database-backup-to-oss.mjs`、`scripts/check-database-backup-to-oss.mjs`、`docs/【开发运维】本地开发验证与生产运维-2026-05-15.md`。 ## Procedure 事务鉴权要兼容 SpacetimeDB 2.6.0 的匿名 TxContext diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 96f99c19b..0c37b4e33 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -398,6 +398,8 @@ files full 会递归扫描 data-dir，保留空目录、每个普通文件的相 history 的安全边界按每个 replica 独立计算。设最新完整且未锁定的 snapshot offset 为 `S`；数字更大但缺少同 offset `.snapshot_bsatn`、仍存在同名 `.lock` 的目录不能参与边界计算。脚本必须保留起始 offset 小于等于 `S` 的最后一个 commitlog segment，以及它之后的全部 segment；只处理更早的 `.stdb.log` / `.stdb.ofs`，snapshot 只处理最新目录之前的旧目录。files history 会递归展开候选目录，逐对象复用或上传，随后依次验真候选对象、history catalog 与 full baseline catalog，再重新扫描边界和 stat fingerprint，最后覆盖发布并验真 `latest.json`；任何一步失败都不推进 state 或删除源文件。脚本在 work-dir 使用 PID lock 拒绝同库并发上传，SSH 超时后必须先检查原进程，不能直接重跑。 +files 本地续跑 state 使用 `-files-state.json.gz` v2：只保存 full/history catalog 的 object key、长度、SHA 与验真时间，不再重复嵌入每份 catalog 的完整 `files` / `symlinks` 清单。旧 v1 `.json` 仍可读取，并且只在一次非 dry-run 备份的 OSS catalog、`latest.json` 与本地新 state 全部成功后原子迁移为 gzip v2，再删除旧 state；gzip 已存在但损坏时必须失败，不能回退到可能过期的旧 JSON。成功运行后，本地只压缩保留 latest full catalog 作为 full 增量复用缓存，已上传并验真的 history catalog、旧 full catalog、失败或 dry-run 遗留 catalog 自动清理；OSS catalog、CAS 对象与 `latest.json` 不删除、不改 schema。`--result-file` 只写 catalog 引用和计数，不再复制完整文件清单。metadata 压缩或清理失败时不得继续删除 `/stdb` history 源文件。 + ```bash # 从停库目录或已验证冻结副本建立逐文件完整基线；相同 work-dir 重跑只传变化内容。 node -- scripts/database-backup-to-oss.mjs \ @@ -431,7 +433,7 @@ node -- scripts/database-backup-to-oss.mjs \ dev 出口过慢时，可以把冻结基线经内网 rsync 到 release 独立 staging，再由 release 上传 dev bucket。staging 必须位于 `/var/lib/genarrative/dev-database-backup-staging/` 一类隔离目录，命令显式传 staging `--data-dir`、独立 `--work-dir`、dev `--bucket`，且不得传 `--stop-service`；禁止指向或修改 release `/stdb`。中转 key 只为本次传输临时授权，结束后从 dev 私钥和 release `authorized_keys` 同时移除。上传完成后把整个 files work-dir/state 回传 dev，history 才能延续同一 baseline catalog。 -完整恢复默认从 OSS 固定 `latest.json` 读取最新 full catalog：先创建 `directories`，再把每个 `files[].objectKey` 下载到 `/` 并逐项核对 `sizeBytes` / `sha256`；history catalog 用于证明已清理历史仍有 OSS 对象，不需要把已被 full baseline 覆盖的旧文件叠回当前恢复目录。本地 state 仍可作为兼容入口，但不再是异机恢复的前置条件。随后用隔离 data-dir 启动同版本 standalone，验证 `/v1/ping`、日志中的 snapshot restore / commitlog replay / module launch、代表性 SQL 和 reducer。dev 已完成这轮 OSS-only 异机恢复与重启演练；当前 live release 仍保持 `archive-full`，需要切换时先为 release 建立并恢复验证独立 full baseline，再通过 Server-Provision 显式选择 `files-history`，无需修改代码或解除额外硬门禁。 +完整恢复默认从 OSS 固定 `latest.json` 读取最新 full catalog：先创建 `directories`，再把每个 `files[].objectKey` 下载到 `/` 并逐项核对 `sizeBytes` / `sha256`；history catalog 用于证明已清理历史仍有 OSS 对象，不需要把已被 full baseline 覆盖的旧文件叠回当前恢复目录。本地 state 仍可作为兼容入口，并同时支持旧 v1 JSON 与 v2 gzip，但不再是异机恢复的前置条件。随后用隔离 data-dir 启动同版本 standalone，验证 `/v1/ping`、日志中的 snapshot restore / commitlog replay / module launch、代表性 SQL 和 reducer。dev 已完成这轮 OSS-only 异机恢复与重启演练；release 已使用独立 `/var/lib/genarrative/database-backups/release-files` full baseline 和 `files-history` profile，现场最终 `ExecStart`、timer 状态与最近备份结果仍须在变更时重新核对。 ```bash node -- scripts/database-backup-to-oss.mjs \ diff --git a/scripts/check-database-backup-to-oss.mjs b/scripts/check-database-backup-to-oss.mjs index 809d54266..2e61af738 100644 --- a/scripts/check-database-backup-to-oss.mjs +++ b/scripts/check-database-backup-to-oss.mjs @@ -6,6 +6,7 @@ import {chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readFileSync, import {tmpdir} from 'node:os'; import path from 'node:path'; import {Readable} from 'node:stream'; +import {gunzipSync, gzipSync} from 'node:zlib'; import { buildAuthorization, @@ -65,12 +66,17 @@ async function main() { await assertHistorySuccessfulUploadCleansAndIsIdempotent(); await assertHistoryResumeReverifiesArchiveAndManifest(); await assertDirectFilesPreservePathsAndIncrementWithoutDuplicateUpload(); + await assertDirectFilesMigratesLegacyStateAndPrunesEmbeddedCatalogs(); await assertDirectFilesConcurrencyIsBounded(); await assertDirectHistoryPublishesCatalogBeforeCleanup(); await assertDirectHistoryWithoutCandidatesPublishesLatest(); await assertDirectFilesRestoreDownloadsCatalogAndObjects(); } +function readGzipJson(filePath) { + return JSON.parse(gunzipSync(readFileSync(filePath)).toString('utf8')); +} + function createDirectOssHarness() { const objects = new Map(); const uploadedKeys = []; @@ -217,6 +223,13 @@ async function assertDirectFilesPreservePathsAndIncrementWithoutDuplicateUpload( const first = await runDirectFilesBackup(options); assertEqual(first.uploadedCount, 2, '首次 files full 应上传全部普通文件。'); assertTrue(!Object.hasOwn(first.catalog, 'dataDir'), '远端 files catalog 不得绑定 staging 主机的绝对 data-dir。'); + assertTrue(first.statePath.endsWith('.json.gz'), 'files state 必须使用 gzip 压缩文件。'); + const compactState = readGzipJson(first.statePath); + assertEqual(compactState.schemaVersion, 2, 'files state 必须使用去重后的 v2 契约。'); + assertTrue(!Object.hasOwn(compactState.baselineCatalog, 'files'), 'baseline ref 不得重复嵌入 files。'); + assertTrue(!Object.hasOwn(compactState.latestCatalog, 'files'), 'latest ref 不得重复嵌入 files。'); + assertTrue(!existsSync(first.catalogPath), '本地 full catalog 原始 JSON 应在成功后压缩。'); + assertTrue(existsSync(`${first.catalogPath}.gz`), '本地应保留压缩后的 latest full catalog 供增量复用。'); const latestObjectKey = 'database-backups/test-db/latest.json'; const latest = JSON.parse(harness.objects.get(latestObjectKey).body.toString('utf8')); assertEqual(latest.latestFullCatalog.catalogId, first.catalogId, 'latest pointer 必须指向已验真的最新 full catalog。'); @@ -231,11 +244,77 @@ async function assertDirectFilesPreservePathsAndIncrementWithoutDuplicateUpload( '相同 catalog 重跑不得重复 PUT 文件或 catalog，但应覆盖验真 latest pointer。', ); + rmSync(`${first.catalogPath}.gz`, {force: false}); writeFileSync(path.join(dataDir, 'control-db'), 'control changed'); writeFileSync(path.join(dataDir, 'new-program.bin'), 'new program'); const incremental = await runDirectFilesBackup(options); assertEqual(incremental.uploadedCount, 2, '增量 files full 只应上传新增和变化文件。'); - assertEqual(incremental.reusedCount, 1, '增量 files full 应复用未变化 snapshot 文件。'); + assertEqual(incremental.reusedCount, 1, '本地 full catalog 缓存缺失时仍应通过 OSS HEAD 复用未变化文件。'); +} + +async function assertDirectFilesMigratesLegacyStateAndPrunesEmbeddedCatalogs() { + const root = path.join(tmpRoot, 'direct-files-state-migration'); + const dataDir = path.join(root, 'stdb'); + const workDir = path.join(root, 'work'); + mkdirSync(dataDir, {recursive: true}); + writeFileSync(path.join(dataDir, 'control-db'), 'control'); + const harness = createDirectOssHarness(); + const options = { + mode: 'full', dataDir, workDir, database: 'test-db', bucket: 'backup-bucket', objectPrefix: 'database-backups', + uploadOptions: {}, uploadFn: harness.uploadFn, uploadManifestFn: harness.uploadManifestFn, verifyFn: harness.verifyFn, + }; + const first = await runDirectFilesBackup(options); + const compactState = readGzipJson(first.statePath); + const catalog = JSON.parse(gunzipSync(readFileSync(`${first.catalogPath}.gz`)).toString('utf8')); + const legacyCatalogRef = { + ...compactState.latestCatalog, + files: catalog.files, + symlinks: catalog.symlinks, + }; + const legacyStatePath = first.statePath.slice(0, -3); + writeFileSync(legacyStatePath, `${JSON.stringify({ + ...compactState, + schemaVersion: 1, + baselineCatalog: legacyCatalogRef, + latestCatalog: legacyCatalogRef, + }, null, 2)}\n`); + rmSync(first.statePath, {force: false}); + + const migrated = await runDirectFilesBackup(options); + assertTrue(migrated.unchanged, '旧 state 迁移不得改变相同 full catalog 的零上传语义。'); + assertTrue(existsSync(migrated.statePath), '旧 state 成功运行后必须生成压缩 state。'); + assertTrue(!existsSync(legacyStatePath), '压缩 state 原子落盘后应删除旧未压缩 state。'); + const migratedState = readGzipJson(migrated.statePath); + assertEqual(migratedState.schemaVersion, 2, '旧 state 必须迁移到 v2。'); + assertTrue(!Object.hasOwn(migratedState.latestCatalog, 'files'), '迁移后 state 不得保留重复 files 清单。'); + + const latestCatalogPath = `${migrated.catalogPath}.gz`; + const validCatalogBody = readFileSync(latestCatalogPath); + writeFileSync(latestCatalogPath, gzipSync(Buffer.from('{}\n'))); + writeFileSync(path.join(dataDir, 'control-db'), 'control changed'); + let corruptCatalogFailure = null; + try { + await runDirectFilesBackup(options); + } catch (error) { + corruptCatalogFailure = error; + } + assertIncludes( + corruptCatalogFailure?.message ?? '', + '长度或 SHA 与 state 引用不匹配', + '本地 latest full catalog 损坏时不得作为增量复用缓存。', + ); + writeFileSync(latestCatalogPath, validCatalogBody); + + writeFileSync(legacyStatePath, `${JSON.stringify({...migratedState, schemaVersion: 1})}\n`); + writeFileSync(migrated.statePath, 'not-a-gzip-state'); + let corruptStateFailure = null; + try { + await runDirectFilesBackup(options); + } catch (error) { + corruptStateFailure = error; + } + assertTrue(corruptStateFailure instanceof Error, '压缩 state 损坏时必须失败。'); + assertTrue(existsSync(legacyStatePath), '压缩 state 损坏时不得静默回退并删除旧 state。'); } async function assertDirectFilesConcurrencyIsBounded() { @@ -290,7 +369,12 @@ async function assertDirectHistoryPublishesCatalogBeforeCleanup() { uploadFn: harness.uploadFn, verifyFn: harness.verifyFn, }; - await runDirectFilesBackup({...common, mode: 'full', uploadManifestFn: harness.uploadManifestFn}); + const baseline = await runDirectFilesBackup({...common, mode: 'full', uploadManifestFn: harness.uploadManifestFn}); + const legacyResultFile = path.join(fixture.workDir, 'legacy-full-result.json'); + writeFileSync(legacyResultFile, `${JSON.stringify({ + ...baseline, + catalog: baseline.catalog, + }, null, 2)}\n`); const plan = discoverHistoryPlan({dataDir: fixture.dataDir}); let failure = null; try { @@ -330,7 +414,13 @@ async function assertDirectHistoryPublishesCatalogBeforeCleanup() { assertTrue(existsSync(path.join(fixture.dataDir, candidate.path)), `latest pointer 发布失败不得删除: ${candidate.path}`); } - const success = await runDirectFilesBackup({...common, mode: 'history', uploadManifestFn: harness.uploadManifestFn}); + const resultFile = path.join(fixture.workDir, 'history-result.json'); + const success = await runDirectFilesBackup({ + ...common, + mode: 'history', + resultFile, + uploadManifestFn: harness.uploadManifestFn, + }); assertEqual(success.uploadedCount, 0, 'history 文件已在 full CAS baseline 时不应重复上传内容。'); for (const file of success.catalog.files) { assertTrue( @@ -340,7 +430,18 @@ async function assertDirectHistoryPublishesCatalogBeforeCleanup() { } assertEqual(success.cleanup?.deletedCount, plan.candidates.length, 'catalog 和 baseline 验真后才应清理全部安全候选。'); - const state = JSON.parse(readFileSync(success.statePath, 'utf8')); + const state = readGzipJson(success.statePath); + assertEqual(state.schemaVersion, 2, 'files state 必须迁移为去重后的 v2 契约。'); + assertTrue(!Object.hasOwn(state.latestCatalog, 'files'), 'files state latest ref 不得重复嵌入 files。'); + assertTrue(state.historyCatalogs.every((catalog) => !Object.hasOwn(catalog, 'files')), 'files state history ref 不得重复嵌入 files。'); + assertTrue(!existsSync(success.catalogPath), '已上传并验真的 history catalog 本地 JSON 应被清理。'); + assertTrue(!existsSync(`${success.catalogPath}.gz`), 'history catalog 本地压缩副本也不应保留。'); + const diskResult = JSON.parse(readFileSync(resultFile, 'utf8')); + assertTrue(!Object.hasOwn(diskResult.catalog, 'files'), 'files result 文件不得重复写入完整 files 清单。'); + assertEqual(diskResult.catalog.fileCount, success.fileCount, '紧凑 result 仍应保留文件计数。'); + const compactedLegacyResult = JSON.parse(readFileSync(legacyResultFile, 'utf8')); + assertTrue(!Object.hasOwn(compactedLegacyResult.catalog, 'files'), '旧 result 中重复的 files 清单应在成功运行后压缩。'); + assertTrue((success.metadataCleanup?.compactedResultCount ?? 0) >= 1, 'metadata 清理应报告已压缩旧 result。'); const historyCatalogObjectKey = state.historyCatalogs[0].objectKey; harness.objects.delete(historyCatalogObjectKey); let brokenHistoryFailure = null; @@ -425,6 +526,29 @@ async function assertDirectFilesRestoreDownloadsCatalogAndObjects() { assertTrue(lstatSync(path.join(restoreDir, 'bin', 'current')).isSymbolicLink(), 'files restore 必须重建符号链接。'); assertEqual(readlinkSync(path.join(restoreDir, 'bin', 'current'), 'utf8'), '2.6.0', 'files restore 必须保留符号链接目标。'); + rmSync(restoreDir, {recursive: true, force: true}); + const legacyRestoreStatePath = path.join(workDir, 'legacy-restore-state.json'); + writeFileSync(legacyRestoreStatePath, `${JSON.stringify({ + ...readGzipJson(latestFull.statePath), + schemaVersion: 1, + })}\n`); + const legacyRestored = await restoreDirectFilesBackup({ + statePath: legacyRestoreStatePath, + restoreDir, + database: 'test-db', + bucket: 'backup-bucket', + uploadOptions: {}, + downloadBufferFn: async ({objectKey}) => Buffer.from(harness.objects.get(objectKey)?.body ?? ''), + downloadFileFn: async ({objectKey, destinationPath}) => { + const object = harness.objects.get(objectKey); + if (!object) { + throw new Error(`missing ${objectKey}`); + } + writeFileSync(destinationPath, object.body); + }, + }); + assertEqual(legacyRestored.downloadedCount, 1, 'files restore 必须继续兼容 v1 JSON state。'); + rmSync(restoreDir, {recursive: true, force: true}); const downloadBufferFn = async ({objectKey}) => { const object = harness.objects.get(objectKey); diff --git a/scripts/database-backup-to-oss.mjs b/scripts/database-backup-to-oss.mjs index 80fb5392c..3fe0a3191 100644 --- a/scripts/database-backup-to-oss.mjs +++ b/scripts/database-backup-to-oss.mjs @@ -26,6 +26,7 @@ import {Readable} from 'node:stream'; import {pipeline} from 'node:stream/promises'; import {setTimeout as sleep} from 'node:timers/promises'; import {fileURLToPath} from 'node:url'; +import {gunzipSync, gzipSync} from 'node:zlib'; const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); @@ -53,7 +54,8 @@ const DIRECT_FILES_SINGLE_PUT_MAX_BYTES = 16 * 1024 * 1024; const RETRYABLE_OSS_HTTP_STATUSES = new Set([408, 429, 500, 502, 503, 504]); const HISTORY_STATE_SCHEMA_VERSION = 1; const HISTORY_MANIFEST_SCHEMA_VERSION = 1; -const DIRECT_FILES_STATE_SCHEMA_VERSION = 1; +const DIRECT_FILES_STATE_SCHEMA_VERSION = 2; +const LEGACY_DIRECT_FILES_STATE_SCHEMA_VERSION = 1; const DIRECT_FILES_CATALOG_SCHEMA_VERSION = 1; const DIRECT_FILES_LATEST_SCHEMA_VERSION = 1; @@ -370,14 +372,23 @@ function buildBackupNames({database, dataDir, objectPrefix}) { return {fileName, objectKey}; } -function atomicWriteJson(filePath, payload) { +function atomicWriteBuffer(filePath, body) { mkdirSync(dirname(filePath), {recursive: true}); const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`; - writeFileSync(tempPath, `${JSON.stringify(payload, null, 2)}\n`, {encoding: 'utf8', mode: 0o600}); + writeFileSync(tempPath, body, {mode: 0o600}); chmodSync(tempPath, 0o600); renameSync(tempPath, filePath); } +function atomicWriteJson(filePath, payload) { + atomicWriteBuffer(filePath, Buffer.from(`${JSON.stringify(payload, null, 2)}\n`, 'utf8')); +} + +function atomicWriteGzipJson(filePath, payload) { + const body = Buffer.from(`${JSON.stringify(payload)}\n`, 'utf8'); + atomicWriteBuffer(filePath, gzipSync(body, {level: 9})); +} + function processIsAlive(pid) { try { process.kill(pid, 0); @@ -1070,9 +1081,127 @@ async function sha256FileHex(filePath) { } function directFilesStatePath({workDir, database}) { + return join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-files-state.json.gz`); +} + +function legacyDirectFilesStatePath({workDir, database}) { return join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-files-state.json`); } +function directCatalogLocalPaths({workDir, database, catalog}) { + const baseName = `${sanitizeObjectPart(database, 'spacetimedb')}-${catalog.mode}-${catalog.catalogId}.catalog.json`; + return { + jsonPath: join(workDir, baseName), + gzipPath: join(workDir, `${baseName}.gz`), + }; +} + +function readLocalDirectCatalog({workDir, database, catalog}) { + const {jsonPath, gzipPath} = directCatalogLocalPaths({workDir, database, catalog}); + let body = null; + if (existsSync(jsonPath)) { + body = readFileSync(jsonPath); + } else if (existsSync(gzipPath)) { + body = gunzipSync(readFileSync(gzipPath)); + } + if (!body) { + return null; + } + if (body.length !== catalog.contentLength || sha256Hex(body) !== catalog.sha256) { + throw new Error(`本地 files catalog 长度或 SHA 与 state 引用不匹配: ${jsonPath}`); + } + const payload = JSON.parse(body.toString('utf8')); + if ( + payload.schemaVersion !== DIRECT_FILES_CATALOG_SCHEMA_VERSION + || payload.database !== database + || payload.mode !== catalog.mode + || payload.catalogId !== catalog.catalogId + || !Array.isArray(payload.files) + ) { + throw new Error(`本地 files catalog 与 state 引用不匹配: ${jsonPath}`); + } + return payload; +} + +function readJsonOrGzip(filePath) { + const body = readFileSync(filePath); + const decoded = filePath.endsWith('.gz') || (body[0] === 0x1f && body[1] === 0x8b) + ? gunzipSync(body) + : body; + return JSON.parse(decoded.toString('utf8')); +} + +function compactDirectCatalogFile({workDir, database, catalog}) { + const {jsonPath, gzipPath} = directCatalogLocalPaths({workDir, database, catalog}); + if (!existsSync(jsonPath)) { + return existsSync(gzipPath) ? {compressed: false, gzipPath} : null; + } + const body = readFileSync(jsonPath); + atomicWriteBuffer(gzipPath, gzipSync(body, {level: 9})); + rmSync(jsonPath, {force: false}); + return {compressed: true, gzipPath}; +} + +function compactDirectFilesLocalMetadata({workDir, database, nextState, transientCatalogPaths = []}) { + const keepCatalog = nextState.latestCatalog; + const keepCatalogIds = new Set([keepCatalog?.catalogId].filter(Boolean)); + const databasePart = sanitizeObjectPart(database, 'spacetimedb'); + const catalogPattern = new RegExp(`^${databasePart.replace(/[.*+?^${}()|[\]\\]/gu, '\\$&')}-(full|history)-([a-f0-9]{64})\\.catalog\\.json(?:\\.gz)?$`, 'u'); + let compressedCatalogCount = 0; + let deletedCatalogCount = 0; + let compactedResultCount = 0; + for (const entry of readdirSync(workDir, {withFileTypes: true})) { + if (!entry.isFile()) { + continue; + } + const match = catalogPattern.exec(entry.name); + if (!match) { + continue; + } + const catalog = {mode: match[1], catalogId: match[2]}; + if (keepCatalogIds.has(catalog.catalogId) && catalog.mode === 'full') { + readLocalDirectCatalog({workDir, database, catalog: keepCatalog}); + if (!entry.name.endsWith('.gz') && compactDirectCatalogFile({workDir, database, catalog})?.compressed) { + compressedCatalogCount += 1; + } + continue; + } + rmSync(join(workDir, entry.name), {force: false}); + deletedCatalogCount += 1; + } + for (const filePath of transientCatalogPaths) { + if (existsSync(filePath)) { + rmSync(filePath, {force: false}); + deletedCatalogCount += 1; + } + } + for (const entry of readdirSync(workDir, {withFileTypes: true})) { + if (!entry.isFile() || !entry.name.endsWith('.json') || entry.name.endsWith('.catalog.json')) { + continue; + } + const filePath = join(workDir, entry.name); + let payload = null; + try { + payload = JSON.parse(readFileSync(filePath, 'utf8')); + } catch { + continue; + } + if ( + payload?.catalog?.schemaVersion !== DIRECT_FILES_CATALOG_SCHEMA_VERSION + || payload.catalog.database !== database + || payload.catalog.bucket !== nextState.bucket + || !Array.isArray(payload.catalog.files) + ) { + continue; + } + atomicWriteJson(filePath, compactDirectFilesResult(payload)); + compactedResultCount += 1; + } + const result = {compressedCatalogCount, deletedCatalogCount, compactedResultCount}; + console.log(`[database-backup] files 本地元数据清理: ${JSON.stringify(result)}`); + return result; +} + function normalizeObjectPrefix(objectPrefix, database) { const prefix = String(objectPrefix || 'database-backups') .trim() @@ -1185,12 +1314,16 @@ function directCatalogIdentity({mode, baselineCatalogId, rootName, directories, } function readDirectFilesState(statePath, {database, bucket}) { - if (!existsSync(statePath)) { + const candidates = statePath.endsWith('.gz') + ? [statePath, statePath.slice(0, -3)] + : [statePath, `${statePath}.gz`]; + const existingPath = candidates.find((candidate) => existsSync(candidate)); + if (!existingPath) { return null; } - const state = readManifest(statePath); + const state = readJsonOrGzip(existingPath); if ( - state.schemaVersion !== DIRECT_FILES_STATE_SCHEMA_VERSION + ![LEGACY_DIRECT_FILES_STATE_SCHEMA_VERSION, DIRECT_FILES_STATE_SCHEMA_VERSION].includes(state.schemaVersion) || state.backupKind !== 'spacetimedb-direct-files-state' || state.database !== database || state.bucket !== bucket @@ -1200,6 +1333,16 @@ function readDirectFilesState(statePath, {database, bucket}) { return state; } +function directPreviousFiles({state, workDir, database}) { + if (!state?.latestCatalog) { + return []; + } + if (Array.isArray(state?.latestCatalog?.files)) { + return state.latestCatalog.files; + } + return readLocalDirectCatalog({workDir, database, catalog: state?.latestCatalog})?.files ?? []; +} + async function ensureDirectObject({ file, dataDir, @@ -1290,6 +1433,52 @@ function directCatalogRef(catalog) { }; } +function normalizeDirectFilesState({state, dataDir, database, bucket}) { + return { + schemaVersion: DIRECT_FILES_STATE_SCHEMA_VERSION, + backupKind: 'spacetimedb-direct-files-state', + database, + dataDir, + bucket, + updatedAt: new Date().toISOString(), + baselineCatalog: assertDirectCatalogRef(state?.baselineCatalog, 'full', 'baseline full'), + latestCatalog: assertDirectCatalogRef(state?.latestCatalog, 'full', 'latest full'), + historyCatalogs: (state?.historyCatalogs ?? []).map((catalog) => ( + assertDirectCatalogRef(catalog, 'history', 'history') + )), + }; +} + +function persistDirectFilesState({statePath, legacyStatePath, state}) { + atomicWriteGzipJson(statePath, state); + if (legacyStatePath !== statePath && existsSync(legacyStatePath)) { + rmSync(legacyStatePath, {force: false}); + } +} + +function compactDirectFilesResult(result) { + if (!result.catalog) { + return result; + } + const {catalog, ...rest} = result; + return { + ...rest, + catalog: { + schemaVersion: catalog.schemaVersion, + backupKind: catalog.backupKind, + database: catalog.database, + bucket: catalog.bucket, + mode: catalog.mode, + catalogId: catalog.catalogId, + catalogObjectKey: catalog.catalogObjectKey, + baselineCatalogId: catalog.baselineCatalogId, + rootName: catalog.rootName, + fileCount: catalog.files.length, + symlinkCount: catalog.symlinks.length, + }, + }; +} + function assertDirectCatalogRef(catalog, expectedMode, label) { if ( !catalog @@ -1397,6 +1586,7 @@ export async function runDirectFilesBackup({ }) { mkdirSync(workDir, {recursive: true}); const statePath = directFilesStatePath({workDir, database}); + const legacyStatePath = legacyDirectFilesStatePath({workDir, database}); const state = readDirectFilesState(statePath, {database, bucket}); if (mode === 'history' && (!state?.baselineCatalog || state?.latestCatalog?.mode !== 'full')) { throw new Error(`files history 模式缺少已发布 full baseline catalog: ${statePath}`); @@ -1464,10 +1654,26 @@ export async function runDirectFilesBackup({ uploadManifestFn, verifyFn, }); + const compactedState = normalizeDirectFilesState({state, dataDir, database, bucket}); + persistDirectFilesState({statePath, legacyStatePath, state: compactedState}); + const metadataCleanup = compactDirectFilesLocalMetadata({ + workDir, + database, + nextState: compactedState, + transientCatalogPaths: [catalogPath], + }); console.log('[database-backup] files history 没有可归档候选。'); - const emptyResult = {...summary, catalog, latestPointer, uploadedCount: 0, reusedCount: 0, cleanup: null}; + const emptyResult = { + ...summary, + catalog, + latestPointer, + uploadedCount: 0, + reusedCount: 0, + cleanup: null, + metadataCleanup, + }; if (resultFile) { - atomicWriteJson(resolvePath(resultFile), emptyResult); + atomicWriteJson(resolvePath(resultFile), compactDirectFilesResult(emptyResult)); } return emptyResult; } @@ -1485,8 +1691,23 @@ export async function runDirectFilesBackup({ uploadManifestFn, verifyFn, }); + const compactedState = normalizeDirectFilesState({state, dataDir, database, bucket}); + persistDirectFilesState({statePath, legacyStatePath, state: compactedState}); + const metadataCleanup = compactDirectFilesLocalMetadata({ + workDir, + database, + nextState: compactedState, + }); console.log('[database-backup] files catalog 未变化，无文件需要上传。'); - return {...summary, catalog, latestPointer, uploadedCount: 0, reusedCount: collected.files.length, unchanged: true}; + return { + ...summary, + catalog, + latestPointer, + uploadedCount: 0, + reusedCount: collected.files.length, + unchanged: true, + metadataCleanup, + }; } } @@ -1499,7 +1720,7 @@ export async function runDirectFilesBackup({ }); } - const previousFiles = new Map((state?.latestCatalog?.files ?? []).map((file) => [file.path, file])); + const previousFiles = new Map(directPreviousFiles({state, workDir, database}).map((file) => [file.path, file])); let uploadedCount = 0; let reusedCount = 0; let nextIndex = 0; @@ -1555,8 +1776,6 @@ export async function runDirectFilesBackup({ contentLength: catalogUpload.contentLength, sha256: catalogUpload.archiveSha256, verifiedAt: catalogUpload.verifiedAt, - files: catalog.files, - symlinks: catalog.symlinks, }; const nextState = { schemaVersion: DIRECT_FILES_STATE_SCHEMA_VERSION, @@ -1565,11 +1784,16 @@ export async function runDirectFilesBackup({ dataDir, bucket, updatedAt: new Date().toISOString(), - baselineCatalog: state?.baselineCatalog ?? catalogRef, - latestCatalog: mode === 'full' ? catalogRef : state.latestCatalog, + baselineCatalog: directCatalogRef(state?.baselineCatalog ?? catalogRef), + latestCatalog: directCatalogRef(mode === 'full' ? catalogRef : state.latestCatalog), historyCatalogs: mode === 'history' - ? [...(state.historyCatalogs ?? []).filter((item) => item.catalogId !== catalogId), catalogRef] - : (state?.historyCatalogs ?? []), + ? [ + ...(state.historyCatalogs ?? []) + .filter((item) => item.catalogId !== catalogId) + .map((item) => directCatalogRef(item)), + directCatalogRef(catalogRef), + ] + : (state?.historyCatalogs ?? []).map((item) => directCatalogRef(item)), }; const latestPointer = await publishDirectFilesLatest({ workDir, @@ -1581,14 +1805,27 @@ export async function runDirectFilesBackup({ uploadManifestFn, verifyFn, }); - atomicWriteJson(statePath, nextState); + persistDirectFilesState({statePath, legacyStatePath, state: nextState}); + const metadataCleanup = compactDirectFilesLocalMetadata({ + workDir, + database, + nextState, + }); let cleanup = null; if (mode === 'history') { cleanup = cleanupHistoryCandidates({dataDir, candidates: plan.candidates}); } - const finalResult = {...summary, catalog, latestPointer, uploadedCount, reusedCount, cleanup}; + const finalResult = { + ...summary, + catalog, + latestPointer, + uploadedCount, + reusedCount, + cleanup, + metadataCleanup, + }; if (resultFile) { - atomicWriteJson(resolvePath(resultFile), finalResult); + atomicWriteJson(resolvePath(resultFile), compactDirectFilesResult(finalResult)); } return finalResult; }