优化逐文件备份并发上传

增加files对象PUT和HEAD受控并发限制并发范围并降低大目录进度日志量补齐并发上限测试、环境示例与运维文档
2026-07-16 18:18:26 +08:00
parent 56d70bf720
commit 40684aafcd
4 changed files with 84 additions and 17 deletions
@@ -159,6 +159,8 @@ GENARRATIVE_DATABASE_BACKUP_OSS_PREFIX=database-backups
 GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL=false
 # 可选：files 为逐文件 CAS + catalog，不生成 tar.gz；archive 保留旧全量压缩包兼容行为。
 GENARRATIVE_DATABASE_BACKUP_STORAGE_FORMAT=archive
+# files 模式并行 PUT/HEAD 数量，必须为 1-64；默认 16。
+GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY=16
 # 可选：显式要求备份工作目录所在文件系统至少保留的可用空间；为空时按数据目录大小 + 安全余量估算。
 GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES=
 # archive history 模式持久化已验真 full baseline 与追加批次；files 模式改用 work-dir 下的 files state。
@@ -327,6 +327,7 @@ GENARRATIVE_DATABASE_BACKUP_OSS_ENDPOINT=oss-cn-shanghai.aliyuncs.com
 GENARRATIVE_DATABASE_BACKUP_OSS_PREFIX=database-backups
 GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL=false
 GENARRATIVE_DATABASE_BACKUP_STORAGE_FORMAT=archive
+GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY=16
 GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES=
 GENARRATIVE_DATABASE_BACKUP_BASELINE_STATE=/var/lib/genarrative/database-backups/genarrative-prod-history-state.json
 # 仅 archive history 首次从一份 uploadStatus=uploaded 的全量 manifest 初始化 state 时设置或传 --baseline-manifest。
@@ -339,7 +340,7 @@ GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_SECRET=

 `files-history` 使用仓库模板 `deploy/systemd/genarrative-database-backup-files-history.conf` 覆盖主 service 的 `ExecStart`，从 `/etc/genarrative/api-server.env` 读取 data-dir、database、bucket、prefix 与 OSS 凭据，不在 unit 写死环境目标，也不传 `--stop-service`。Server-Provision 在改动 drop-in 前，先用 current release 的同一脚本、同一 env 和 `DATABASE_BACKUP_FILES_HISTORY_WORK_DIR` 执行一次 history `--dry-run`；缺少已发布 full catalog 的 files state、current 脚本过旧或配置不匹配都会在安装 drop-in 和 `daemon-reload` 前失败。选择 `archive-full` 会主动删除仓库托管的 `10-files-history.conf` 与 dev 试点遗留的 `10-dev-files.conf`，防止 systemd 继续合并旧覆盖。dev 可继续指定已有 `/var/lib/genarrative/database-backups/dev-files`，release 建议先在 `/var/lib/genarrative/database-backups/release-files` 建立自己的 full baseline；两台机器不得复用或互传本地 state 目录冒充本机基线。启用时通过 Server-Provision Job 选择目标、`DATABASE_BACKUP_PROFILE=files-history` 和对应 work-dir，先保持 `DRY_RUN=true` 核对，再以同参数正式 provision。不要直接在 `/etc/systemd/system` 手写第二份 drop-in。

-files full 会递归扫描 data-dir，保留空目录、每个普通文件的相对路径，以及目标仍位于 data-dir 内部的相对符号链接；绝对链接或解析后越界的链接直接拒绝。文件按 SHA-256 上传到不可变对象 key，catalog 记录目录、路径、长度、SHA、对象 key 和相对链接目标，不写 staging 主机的绝对路径。相同 catalog 重跑不重复 PUT；新增或变化文件先 HEAD CAS 对象，存在且长度/SHA 元数据一致就复用，否则上传。full 基线必须来自停库后的 data-dir 或已通过恢复验证的冻结副本；源文件上传前后 stat 虽会复核，但在线扫描不能保证大量文件属于同一跨文件一致时点。catalog 验真后，脚本把最新 full/history 引用发布到固定 `<prefix>/<database>/latest.json`，全新机器不需要本地 state 即可自动发现恢复入口。
+files full 会递归扫描 data-dir，保留空目录、每个普通文件的相对路径，以及目标仍位于 data-dir 内部的相对符号链接；绝对链接或解析后越界的链接直接拒绝。文件按 SHA-256 上传到不可变对象 key，catalog 记录目录、路径、长度、SHA、对象 key 和相对链接目标，不写 staging 主机的绝对路径。相同 catalog 重跑不重复 PUT；新增或变化文件先 HEAD CAS 对象，存在且长度/SHA 元数据一致就复用，否则上传。对象 PUT/HEAD 默认以 16 路并行执行，可用 `GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY=1..64` 调整；并发只缩短传输与验真时间，不改变“全部对象、catalog 与 latest pointer 成功后才推进 state/清理”的顺序。full 基线必须来自停库后的 data-dir 或已通过恢复验证的冻结副本；源文件上传前后 stat 虽会复核，但在线扫描不能保证大量文件属于同一跨文件一致时点。catalog 验真后，脚本把最新 full/history 引用发布到固定 `<prefix>/<database>/latest.json`，全新机器不需要本地 state 即可自动发现恢复入口。

 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 超时后必须先检查原进程，不能直接重跑。

@@ -60,6 +60,7 @@ async function main() {
  await assertHistorySuccessfulUploadCleansAndIsIdempotent();
  await assertHistoryResumeReverifiesArchiveAndManifest();
  await assertDirectFilesPreservePathsAndIncrementWithoutDuplicateUpload();
+  await assertDirectFilesConcurrencyIsBounded();
  await assertDirectHistoryPublishesCatalogBeforeCleanup();
  await assertDirectHistoryWithoutCandidatesPublishesLatest();
  await assertDirectFilesRestoreDownloadsCatalogAndObjects();
@@ -153,6 +154,44 @@ async function assertDirectFilesPreservePathsAndIncrementWithoutDuplicateUpload(
  assertEqual(incremental.reusedCount, 1, '增量 files full 应复用未变化 snapshot 文件。');
 }

+async function assertDirectFilesConcurrencyIsBounded() {
+  const root = path.join(tmpRoot, 'direct-files-concurrency');
+  const dataDir = path.join(root, 'stdb');
+  const workDir = path.join(root, 'work');
+  mkdirSync(dataDir, {recursive: true});
+  for (let index = 0; index < 12; index += 1) {
+    writeFileSync(path.join(dataDir, `file-${index}.bin`), `content-${index}`);
+  }
+  const harness = createDirectOssHarness();
+  let activeUploads = 0;
+  let maxActiveUploads = 0;
+  const uploadFn = async (options) => {
+    activeUploads += 1;
+    maxActiveUploads = Math.max(maxActiveUploads, activeUploads);
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    try {
+      return await harness.uploadFn(options);
+    } finally {
+      activeUploads -= 1;
+    }
+  };
+  await runDirectFilesBackup({
+    mode: 'full',
+    dataDir,
+    workDir,
+    database: 'test-db',
+    bucket: 'backup-bucket',
+    objectPrefix: 'database-backups',
+    uploadOptions: {},
+    uploadFn,
+    uploadManifestFn: harness.uploadManifestFn,
+    verifyFn: harness.verifyFn,
+    concurrency: 3,
+  });
+  assertTrue(maxActiveUploads > 1, 'files 备份应按配置并发处理多个对象。');
+  assertTrue(maxActiveUploads <= 3, 'files 备份对象并发数不得超过配置上限。');
+}
+
 async function assertDirectHistoryPublishesCatalogBeforeCleanup() {
  const fixture = createHistoryFixture('direct-files-history-cleanup', {nestedData: false});
  createReplicaHistory(fixture.replicasDir, '1', {snapshots: [0, 10], segments: [0, 1, 11]});
@@ -47,6 +47,8 @@ const OSS_MAX_MULTIPART_PARTS = 10_000;
 const DEFAULT_OSS_REQUEST_MAX_ATTEMPTS = 5;
 const DEFAULT_OSS_RETRY_BASE_DELAY_MS = 1_000;
 const DEFAULT_OSS_RETRY_MAX_DELAY_MS = 30_000;
+const DEFAULT_DIRECT_FILES_CONCURRENCY = 16;
+const MAX_DIRECT_FILES_CONCURRENCY = 64;
 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;
@@ -277,6 +279,14 @@ function firstNonEmpty(...values) {
  return values.map((value) => String(value ?? '').trim()).find(Boolean) ?? '';
 }

+function parseDirectFilesConcurrency(rawValue) {
+  const value = Number(String(rawValue ?? DEFAULT_DIRECT_FILES_CONCURRENCY).trim());
+  if (!Number.isSafeInteger(value) || value < 1 || value > MAX_DIRECT_FILES_CONCURRENCY) {
+    throw new Error(`GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY 必须是 1-${MAX_DIRECT_FILES_CONCURRENCY} 的整数，实际: ${rawValue}`);
+  }
+  return value;
+}
+
 function resolvePath(value) {
  return isAbsolute(value) ? value : resolve(REPO_ROOT, value);
 }
@@ -1341,6 +1351,7 @@ export async function runDirectFilesBackup({
  uploadFn = uploadArchive,
  uploadManifestFn = uploadManifestFile,
  verifyFn = verifyOssObject,
+  concurrency = 1,
 }) {
  mkdirSync(workDir, {recursive: true});
  const statePath = directFilesStatePath({workDir, database});
@@ -1449,23 +1460,35 @@ export async function runDirectFilesBackup({
  const previousFiles = new Map((state?.latestCatalog?.files ?? []).map((file) => [file.path, file]));
  let uploadedCount = 0;
  let reusedCount = 0;
-  for (const [index, file] of collected.files.entries()) {
-    const result = await ensureDirectObject({
-      file,
-      dataDir,
-      uploadOptions,
-      previousFile: previousFiles.get(file.path),
-      verifyCatalogReuse: mode === 'history',
-      uploadFn,
-      verifyFn,
-    });
-    if (result.status === 'uploaded') {
-      uploadedCount += 1;
-    } else {
-      reusedCount += 1;
+  let nextIndex = 0;
+  let completedCount = 0;
+  const workerCount = Math.min(concurrency, collected.files.length);
+  const workers = Array.from({length: workerCount}, async () => {
+    while (nextIndex < collected.files.length) {
+      const index = nextIndex;
+      nextIndex += 1;
+      const file = collected.files[index];
+      const result = await ensureDirectObject({
+        file,
+        dataDir,
+        uploadOptions,
+        previousFile: previousFiles.get(file.path),
+        verifyCatalogReuse: mode === 'history',
+        uploadFn,
+        verifyFn,
+      });
+      if (result.status === 'uploaded') {
+        uploadedCount += 1;
+      } else {
+        reusedCount += 1;
+      }
+      completedCount += 1;
+      if (collected.files.length <= 100 || completedCount % 1000 === 0 || completedCount === collected.files.length) {
+        console.log(`[database-backup] files 进度: ${completedCount}/${collected.files.length} (${result.status}) ${file.path}`);
+      }
    }
-    console.log(`[database-backup] files 进度: ${index + 1}/${collected.files.length} (${result.status}) ${file.path}`);
-  }
+  });
+  await Promise.all(workers);
  const catalogUpload = await ensureDirectManifest({
    manifestPath: catalogPath,
    objectKey: catalogObjectKey,
@@ -2704,6 +2727,7 @@ async function main() {
  const database = firstNonEmpty(args.database, env.GENARRATIVE_SPACETIME_DATABASE, basename(dataDir));
  const keepLocal = args.keepLocal || String(env.GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL ?? '').trim().toLowerCase() === 'true';
  const storageFormat = firstNonEmpty(args.storageFormat, env.GENARRATIVE_DATABASE_BACKUP_STORAGE_FORMAT, 'archive');
+  const directFilesConcurrency = parseDirectFilesConcurrency(env.GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY);

  if (!['full', 'history'].includes(args.mode)) {
    throw new Error(`--mode 只能是 full 或 history，实际: ${args.mode}`);
@@ -2793,6 +2817,7 @@ async function main() {
        dryRun: args.dryRun,
        resultFile: args.resultFile,
        uploadOptions: {bucket, endpoint, accessKeyId, accessKeySecret},
+        concurrency: directFilesConcurrency,
      });
    } catch (error) {
      backupError = error;