Files
Genarrative/scripts/database-backup-to-oss.mjs
T
kdletters a7337c67a1
Project CI / Backend tests (push) Successful in 10m5s
Project CI / Native shell tests (push) Successful in 15m8s
Project CI / Repository checks (push) Successful in 3m24s
Project CI / Frontend tests (push) Successful in 4m40s
修复生产发布内存持续增长 (#203)
修复 release 内存持续增长问题。

本次范围:
- 收口备份扫描与历史维护的内存峰值。
- 限制外部生成 worker 脱管任务的实际并发。
- 为 API 内存态和历史数据增加有界留存。

验收:
- 定向测试、cargo 检查和生产运维门禁通过。
- 备份不再触发全局 OOM。
- worker/API/SpacetimeDB 内存曲线在空闲期停止单调增长。

Reviewed-on: http://192.168.35.82/git/GenarrativeAI/Genarrative/pulls/203
Co-authored-by: kdletters <kdletters@qq.com>
Co-committed-by: kdletters <kdletters@qq.com>
2026-08-27 21:46:56 +08:00

3596 lines
123 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env node
import {spawnSync} from 'node:child_process';
import {createHash, createHmac} from 'node:crypto';
import {
chmodSync,
closeSync,
createReadStream,
createWriteStream,
existsSync,
lstatSync,
mkdirSync,
openSync,
readdirSync,
readFileSync,
readlinkSync,
realpathSync,
renameSync,
rmSync,
statfsSync,
statSync,
symlinkSync,
writeFileSync,
} from 'node:fs';
import {basename, dirname, isAbsolute, join, relative, resolve, sep} from 'node:path';
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);
const REPO_ROOT = resolve(__dirname, '..');
const DEFAULT_LOCAL_DATA_DIR = resolve(REPO_ROOT, 'server-rs/.spacetimedb/local/data');
const DEFAULT_LOCAL_WORK_DIR = resolve(REPO_ROOT, 'server-rs/.data/database-backups');
const DEFAULT_PRODUCTION_DATA_DIR = '/stdb';
const DEFAULT_PRODUCTION_WORK_DIR = '/var/lib/genarrative/database-backups';
const DEFAULT_DATABASE_BACKUP_STOP_MARKER = join(DEFAULT_PRODUCTION_WORK_DIR, '.spacetimedb-stopped');
const DEFAULT_SPACE_SAFETY_RATIO = 1.1;
const DEFAULT_EXTRA_FREE_BYTES = 512 * 1024 * 1024;
const OSS_ALGORITHM = 'OSS4-HMAC-SHA256';
const OSS_SERVICE = 'oss';
const OSS_REQUEST = 'aliyun_v4_request';
const UNSIGNED_PAYLOAD = 'UNSIGNED-PAYLOAD';
const DEFAULT_OSS_MULTIPART_PART_SIZE_BYTES = 128 * 1024 * 1024;
const OSS_MIN_MULTIPART_PART_SIZE_BYTES = 100 * 1024;
const OSS_MAX_MULTIPART_PART_SIZE_BYTES = 5 * 1024 * 1024 * 1024;
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 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 = 2;
const LEGACY_DIRECT_FILES_STATE_SCHEMA_VERSION = 1;
const DIRECT_FILES_CATALOG_SCHEMA_VERSION = 1;
const DIRECT_FILES_LATEST_SCHEMA_VERSION = 1;
function usage() {
console.log(`用法:
npm run database:backup:oss -- [--mode full|history] [--storage-format archive|files] [--data-dir <path>] [--work-dir <path>] [--bucket <bucket>] [--object-prefix <prefix>] [--keep-local]
node -- scripts/database-backup-to-oss.mjs [--stop-service spacetimedb.service] [--restart-service-after genarrative-api.service] [--defer-upload]
node -- scripts/database-backup-to-oss.mjs --upload-archive <path>
node -- scripts/database-backup-to-oss.mjs --upload-deferred-dir <path>
node -- scripts/database-backup-to-oss.mjs --publish-manifest <path>
node -- scripts/database-backup-to-oss.mjs --restore-files-state <path> --restore-dir <path>
node -- scripts/database-backup-to-oss.mjs --restore-files-latest --restore-dir <path> [--dry-run]
说明:
将 SpacetimeDB 数据目录以 .tar.gz 或逐文件 catalog 形式上传到阿里云 OSS 指定 bucket。
默认 full 模式保持原有全量冷备行为;history 模式只归档已被最新 snapshot 覆盖的历史 commitlog 与旧 snapshot。
--storage-format files 不打包:按原相对路径建立 catalog,文件内容以 SHA-256 不可变对象上传;重复运行只上传新增或变化内容。
archive history 必须有已验真的 full baseline statefiles history 必须复用同一 work-dir 中已发布的 full catalog state。
--defer-upload 只生成本地冷备份和 manifest,不上传;后续用 --upload-archive 异步上传。
--upload-deferred-dir 串行收敛目录内 deferred/pending 及已上传未清理归档;只有 OSS 上传与验真完成后才按 keep-local 规则删除。
默认读取 .env / .env.local / .env.secrets.local;生产服务可传 --env-file /etc/genarrative/api-server.env。
shell 环境变量优先级最高,不会被 env 文件覆盖。
常用环境变量:
GENARRATIVE_DATABASE_BACKUP_DATA_DIR 数据目录;生产建议 /stdb
GENARRATIVE_DATABASE_BACKUP_WORK_DIR 本地临时备份目录;生产建议 /var/lib/genarrative/database-backups
GENARRATIVE_DATABASE_BACKUP_OSS_BUCKET 备份 bucket;未设置时回退 ALIYUN_OSS_BUCKET
GENARRATIVE_DATABASE_BACKUP_OSS_PREFIX 对象前缀,默认 database-backups
GENARRATIVE_DATABASE_BACKUP_OSS_ENDPOINT OSS endpoint;未设置时回退 ALIYUN_OSS_ENDPOINT
GENARRATIVE_DATABASE_BACKUP_STORAGE_FORMAT archive(默认)或 files
GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL true 时保留本地 tar.gz
GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES 备份前要求 work dir 所在文件系统至少有这些可用字节;未设置时按数据目录大小估算
GENARRATIVE_DATABASE_BACKUP_BASELINE_STATE history 使用的 full baseline 与追加批次状态文件
GENARRATIVE_DATABASE_BACKUP_BASELINE_MANIFEST 首次初始化 history state 的 uploaded full manifest
ALIYUN_OSS_ACCESS_KEY_ID / ALIYUN_OSS_ACCESS_KEY_SECRET
`);
}
function loadEnvFile(filePath, target, protectedKeys) {
if (!existsSync(filePath)) {
return;
}
const rawText = readFileSync(filePath, 'utf8');
for (const rawLine of rawText.split(/\r?\n/u)) {
const line = rawLine.trim();
if (!line || line.startsWith('#')) {
continue;
}
const match = line.match(/^([A-Za-z_][A-Za-z0-9_]*)=(.*)$/u);
if (!match) {
continue;
}
const [, key, rawValue] = match;
if (protectedKeys.has(key)) {
continue;
}
target[key] = rawValue.replace(/^['"]|['"]$/gu, '');
}
}
function loadRepoEnv() {
const env = {...process.env};
const protectedKeys = new Set(
Object.entries(process.env)
.filter(([, value]) => String(value ?? '').trim())
.map(([key]) => key),
);
for (const fileName of ['.env', '.env.local', '.env.secrets.local']) {
loadEnvFile(resolve(REPO_ROOT, fileName), env, protectedKeys);
}
return env;
}
function loadEffectiveEnv(envFiles) {
const env = loadRepoEnv();
const protectedKeys = new Set(
Object.entries(process.env)
.filter(([, value]) => String(value ?? '').trim())
.map(([key]) => key),
);
for (const filePath of envFiles) {
loadEnvFile(resolvePath(filePath), env, protectedKeys);
}
return env;
}
function parseArgs(argv) {
const options = {
dataDir: '',
workDir: '',
bucket: '',
endpoint: '',
objectPrefix: '',
accessKeyId: '',
accessKeySecret: '',
envFiles: [],
keepLocal: false,
stopService: '',
restartServicesAfter: [],
database: '',
dryRun: false,
deferUpload: false,
uploadArchive: '',
uploadDeferredDir: '',
manifestFile: '',
objectKey: '',
resultFile: '',
minFreeBytes: '',
mode: 'full',
baselineState: '',
baselineManifest: '',
publishManifest: '',
storageFormat: '',
restoreFilesState: '',
restoreFilesLatest: false,
restoreDir: '',
};
for (let index = 0; index < argv.length; index += 1) {
const arg = argv[index];
const readValue = () => {
const value = argv[index + 1];
if (!value || value.startsWith('--')) {
throw new Error(`${arg} 缺少参数值`);
}
index += 1;
return value;
};
switch (arg) {
case '--help':
case '-h':
usage();
process.exit(0);
break;
case '--data-dir':
options.dataDir = readValue();
break;
case '--work-dir':
options.workDir = readValue();
break;
case '--bucket':
options.bucket = readValue();
break;
case '--endpoint':
options.endpoint = readValue();
break;
case '--object-prefix':
options.objectPrefix = readValue();
break;
case '--object-key':
options.objectKey = readValue();
break;
case '--access-key-id':
options.accessKeyId = readValue();
break;
case '--access-key-secret':
options.accessKeySecret = readValue();
break;
case '--env-file':
options.envFiles.push(readValue());
break;
case '--database':
options.database = readValue();
break;
case '--stop-service':
options.stopService = readValue();
break;
case '--restart-service-after':
options.restartServicesAfter.push(readValue());
break;
case '--keep-local':
options.keepLocal = true;
break;
case '--dry-run':
options.dryRun = true;
break;
case '--defer-upload':
options.deferUpload = true;
options.keepLocal = true;
break;
case '--upload-archive':
options.uploadArchive = readValue();
break;
case '--upload-deferred-dir':
options.uploadDeferredDir = readValue();
break;
case '--manifest-file':
options.manifestFile = readValue();
break;
case '--result-file':
options.resultFile = readValue();
break;
case '--min-free-bytes':
options.minFreeBytes = readValue();
break;
case '--mode':
options.mode = readValue();
break;
case '--baseline-state':
options.baselineState = readValue();
break;
case '--baseline-manifest':
options.baselineManifest = readValue();
break;
case '--publish-manifest':
options.publishManifest = readValue();
break;
case '--storage-format':
options.storageFormat = readValue();
break;
case '--restore-files-state':
options.restoreFilesState = readValue();
break;
case '--restore-files-latest':
options.restoreFilesLatest = true;
break;
case '--restore-dir':
options.restoreDir = readValue();
break;
default:
throw new Error(`未知参数: ${arg}`);
}
}
return options;
}
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;
}
export function createUploadBandwidthLimiter(rawValue, {nowFn = Date.now, sleepImpl = sleep} = {}) {
const normalized = String(rawValue ?? '').trim();
if (!normalized || normalized === '0') {
return null;
}
const maxBytesPerSecond = Number(normalized);
if (!Number.isSafeInteger(maxBytesPerSecond) || maxBytesPerSecond < 1024) {
throw new Error(`GENARRATIVE_DATABASE_BACKUP_UPLOAD_MAX_BYTES_PER_SECOND 必须为空、0 或 >= 1024 的整数,实际: ${rawValue}`);
}
let nextAvailableAtMs = 0;
const waitForChunk = async (sizeBytes) => {
const now = nowFn();
const startAt = Math.max(now, nextAvailableAtMs);
const finishAt = startAt + (sizeBytes / maxBytesPerSecond) * 1000;
nextAvailableAtMs = finishAt;
const delayMs = Math.max(0, finishAt - now);
if (delayMs > 0) {
await sleepImpl(delayMs);
}
};
return {
maxBytesPerSecond,
wrap(readable) {
return Readable.from((async function* throttleUpload() {
for await (const chunk of readable) {
await waitForChunk(chunk.length);
yield chunk;
}
})(), {objectMode: false});
},
};
}
function createBufferReadStream(buffer, chunkSizeBytes = 64 * 1024) {
return Readable.from((function* readChunks() {
for (let offset = 0; offset < buffer.length; offset += chunkSizeBytes) {
yield buffer.subarray(offset, Math.min(offset + chunkSizeBytes, buffer.length));
}
})(), {objectMode: false});
}
function resolvePath(value) {
return isAbsolute(value) ? value : resolve(REPO_ROOT, value);
}
function normalizeEndpoint(raw) {
return String(raw ?? '')
.trim()
.replace(/^https?:\/\//u, '')
.replace(/\/+$/u, '');
}
function sanitizeObjectPart(value, fallback) {
const sanitized = String(value ?? '')
.trim()
.toLowerCase()
.replace(/[^a-z0-9._-]+/gu, '-')
.replace(/-+/gu, '-')
.replace(/^-|-$/gu, '');
return sanitized || fallback;
}
function timestampForFile(date = new Date()) {
const pad = (value) => String(value).padStart(2, '0');
return `${date.getUTCFullYear()}${pad(date.getUTCMonth() + 1)}${pad(date.getUTCDate())}T${pad(date.getUTCHours())}${pad(date.getUTCMinutes())}${pad(date.getUTCSeconds())}Z`;
}
function buildBackupNames({database, dataDir, objectPrefix}) {
const timestamp = timestampForFile();
const databasePart = sanitizeObjectPart(database || basename(dataDir), 'spacetimedb');
const fileName = `${databasePart}-${timestamp}.tar.gz`;
const prefix = String(objectPrefix || 'database-backups')
.trim()
.replace(/^\/+|\/+$/gu, '')
.split('/')
.filter(Boolean)
.map((part) => sanitizeObjectPart(part, 'backup'))
.join('/');
const objectKey = [prefix, databasePart, fileName].filter(Boolean).join('/');
return {fileName, objectKey};
}
function atomicWriteBuffer(filePath, body) {
mkdirSync(dirname(filePath), {recursive: true});
const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
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);
return true;
} catch (error) {
return error?.code === 'EPERM';
}
}
function acquireBackupLock({workDir, database}) {
mkdirSync(workDir, {recursive: true});
const lockPath = join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}.backup.lock`);
try {
const fd = openSync(lockPath, 'wx', 0o600);
writeFileSync(fd, `${process.pid}\n`, 'utf8');
closeSync(fd);
const release = () => {
try {
const ownerPid = Number(String(readFileSync(lockPath, 'utf8')).trim());
if (ownerPid === process.pid) {
rmSync(lockPath, {force: true});
}
} catch {
// The lock may already have been removed by the normal exit path.
}
};
process.once('exit', release);
for (const signal of ['SIGINT', 'SIGTERM']) {
process.once(signal, () => {
release();
process.exit(signal === 'SIGINT' ? 130 : 143);
});
}
return lockPath;
} catch (error) {
if (error?.code !== 'EEXIST') {
throw error;
}
}
const ownerPid = Number(String(readFileSync(lockPath, 'utf8')).trim());
if (Number.isSafeInteger(ownerPid) && ownerPid > 0 && processIsAlive(ownerPid)) {
throw new Error(`已有数据库备份进程持有锁: ${lockPath} pid=${ownerPid}`);
}
throw new Error(`发现失效数据库备份锁,拒绝自动抢锁;请核对 OSS multipart 与进程后手工删除: ${lockPath} pid=${ownerPid || '<invalid>'}`);
}
function historyStatePath({args, env, workDir, database}) {
return resolvePath(firstNonEmpty(
args.baselineState,
env.GENARRATIVE_DATABASE_BACKUP_BASELINE_STATE,
join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-history-state.json`),
));
}
function baselineIdFor(baseline) {
return sha256Hex([
baseline.bucket,
baseline.objectKey,
baseline.verifiedAt,
baseline.contentLength,
baseline.archiveSha256,
].join('\0')).slice(0, 24);
}
function normalizeUploadedBaselineManifest(manifest, {database, dataDir}) {
if (manifest.uploadStatus !== 'uploaded') {
throw new Error(`baseline manifest 必须是 uploaded,实际: ${manifest.uploadStatus ?? '<missing>'}`);
}
if (manifest.backupKind !== 'spacetimedb-data-dir') {
throw new Error(`baseline manifest backupKind 必须是 spacetimedb-data-dir,实际: ${manifest.backupKind ?? '<missing>'}`);
}
const baseline = {
backupKind: 'spacetimedb-data-dir',
database: firstNonEmpty(manifest.database, database),
dataDir: resolvePath(firstNonEmpty(manifest.dataDir, dataDir)),
bucket: String(manifest.bucket ?? '').trim(),
objectKey: String(manifest.objectKey ?? '').trim(),
verifiedAt: String(manifest.verifiedAt ?? '').trim(),
uploadedAt: String(manifest.uploadedAt ?? '').trim(),
contentLength: Number(manifest.contentLength),
archiveSha256: String(manifest.archiveSha256 ?? '').trim().toLowerCase(),
manifestObjectKey: String(manifest.manifestObjectKey ?? '').trim(),
manifestContentLength: Number(manifest.manifestContentLength),
manifestArchiveSha256: String(manifest.manifestArchiveSha256 ?? '').trim().toLowerCase(),
manifestVerifiedAt: String(manifest.manifestVerifiedAt ?? '').trim(),
};
if (
!baseline.bucket
|| !baseline.objectKey
|| !baseline.verifiedAt
|| !Number.isSafeInteger(baseline.contentLength)
|| baseline.contentLength <= 0
|| !/^[a-f0-9]{64}$/u.test(baseline.archiveSha256)
|| !baseline.manifestObjectKey
|| !Number.isSafeInteger(baseline.manifestContentLength)
|| baseline.manifestContentLength <= 0
|| !/^[a-f0-9]{64}$/u.test(baseline.manifestArchiveSha256)
|| !baseline.manifestVerifiedAt
) {
throw new Error('baseline manifest 缺少已验真 OSS 归档或 sidecar 信息。');
}
baseline.id = baselineIdFor(baseline);
return baseline;
}
function validateHistoryState(state, {database, dataDir}) {
if (state.schemaVersion !== HISTORY_STATE_SCHEMA_VERSION || !state.baseline) {
throw new Error('history state schemaVersion 或 baseline 无效。');
}
const baseline = normalizeUploadedBaselineManifest(
{...state.baseline, uploadStatus: 'uploaded'},
{database, dataDir},
);
if (baseline.database !== database) {
throw new Error(`history state database 不匹配: expected=${database}, actual=${baseline.database}`);
}
if (resolvePath(baseline.dataDir) !== resolvePath(dataDir)) {
throw new Error(`history state dataDir 不匹配: expected=${resolvePath(dataDir)}, actual=${resolvePath(baseline.dataDir)}`);
}
return {
...state,
baseline,
batches: Array.isArray(state.batches) ? state.batches : [],
};
}
function writeBaselineState({statePath, baseline, previousState = null}) {
const state = {
schemaVersion: HISTORY_STATE_SCHEMA_VERSION,
updatedAt: new Date().toISOString(),
baseline,
batches: previousState?.baseline?.id === baseline.id && Array.isArray(previousState.batches)
? previousState.batches
: [],
};
atomicWriteJson(statePath, state);
return state;
}
function loadOrImportHistoryState({args, env, statePath, database, dataDir}) {
if (existsSync(statePath)) {
return validateHistoryState(readManifest(statePath), {database, dataDir});
}
const importPath = firstNonEmpty(args.baselineManifest, env.GENARRATIVE_DATABASE_BACKUP_BASELINE_MANIFEST);
if (!importPath) {
throw new Error(`history 模式缺少已验真 baseline state: ${statePath};可用 --baseline-manifest 导入已有 uploaded baseline manifest。`);
}
const baseline = normalizeUploadedBaselineManifest(readManifest(resolvePath(importPath)), {database, dataDir});
return writeBaselineState({statePath, baseline});
}
function assertSafeRelativePath(dataDir, absolutePath) {
const relativePath = relative(resolvePath(dataDir), resolvePath(absolutePath));
if (!relativePath || relativePath === '..' || relativePath.startsWith(`..${sep}`) || isAbsolute(relativePath)) {
throw new Error(`history 候选路径越界或等于数据目录: ${absolutePath}`);
}
return relativePath.split(sep).join('/');
}
function statFingerprint(absolutePath, rootPath = absolutePath) {
// 候选 snapshot 可能包含数十万条目录项;增量更新摘要,避免把每条
// fingerprint 字符串同时保存在 entries[] 后再 join,造成一次性内存峰值。
const fingerprintHash = createHash('sha256');
let isFirstEntry = true;
let entryCount = 0;
let totalSize = 0n;
const visit = (currentPath) => {
const stat = lstatSync(currentPath, {bigint: true});
if (stat.isSymbolicLink()) {
throw new Error(`history 候选不得包含符号链接: ${currentPath}`);
}
const entryPath = currentPath === rootPath ? '.' : relative(rootPath, currentPath).split(sep).join('/');
const kind = stat.isDirectory() ? 'directory' : stat.isFile() ? 'file' : 'other';
if (kind === 'other') {
throw new Error(`history 候选只允许普通文件或目录: ${currentPath}`);
}
const entry = [
entryPath,
kind,
stat.dev.toString(),
stat.ino.toString(),
stat.mode.toString(),
stat.size.toString(),
stat.mtimeNs.toString(),
].join('\0');
if (!isFirstEntry) {
fingerprintHash.update('\n');
}
fingerprintHash.update(entry);
isFirstEntry = false;
entryCount += 1;
if (stat.isFile()) {
totalSize += stat.size;
} else {
for (const name of readdirSync(currentPath).sort()) {
visit(join(currentPath, name));
}
}
};
visit(rootPath);
return {
fingerprint: fingerprintHash.digest('hex'),
sizeBytes: totalSize.toString(),
entryCount,
};
}
function findReplicasDir(dataDir) {
const candidates = [resolve(dataDir, 'replicas'), resolve(dataDir, 'data', 'replicas')]
.filter((candidate) => existsSync(candidate) && lstatSync(candidate).isDirectory());
if (candidates.length !== 1) {
throw new Error(`无法唯一确定 replicas 目录: ${candidates.length === 0 ? '<none>' : candidates.join(', ')}`);
}
return candidates[0];
}
function historyCandidate({dataDir, absolutePath, kind, replicaId, transaction}) {
const stat = statFingerprint(absolutePath);
return {
path: assertSafeRelativePath(dataDir, absolutePath),
kind,
replicaId,
transaction: transaction.toString(),
...stat,
};
}
export function discoverHistoryPlan({dataDir}) {
const resolvedDataDir = resolvePath(dataDir);
const replicasDir = findReplicasDir(resolvedDataDir);
const replicaEntries = readdirSync(replicasDir, {withFileTypes: true});
const replicas = [];
const candidates = [];
for (const replicaEntry of replicaEntries.sort((left, right) => left.name.localeCompare(right.name))) {
if (!replicaEntry.isDirectory()) {
continue;
}
if (!/^\d+$/u.test(replicaEntry.name)) {
throw new Error(`replica 目录名不符合预期: ${replicaEntry.name}`);
}
const replicaId = replicaEntry.name;
const replicaDir = join(replicasDir, replicaId);
const snapshotsDir = join(replicaDir, 'snapshots');
const clogDir = join(replicaDir, 'clog');
if (!existsSync(snapshotsDir) || !lstatSync(snapshotsDir).isDirectory()) {
replicas.push({replicaId, status: 'skipped', reason: 'no-snapshots-directory'});
continue;
}
const snapshotEntries = readdirSync(snapshotsDir, {withFileTypes: true});
const snapshots = snapshotEntries.flatMap((entry) => {
const match = /^(\d{20})\.snapshot_dir$/u.exec(entry.name);
if (!match) {
return [];
}
const transaction = BigInt(match[1]);
if (transaction > 0xffff_ffff_ffff_ffffn) {
throw new Error(`snapshot transaction 超出 u64: ${entry.name}`);
}
if (!entry.isDirectory()) {
throw new Error(`snapshot 候选必须是目录: ${join(snapshotsDir, entry.name)}`);
}
const snapshotDir = join(snapshotsDir, entry.name);
const lockPath = join(snapshotsDir, `${match[1]}.lock`);
const snapshotFile = join(snapshotDir, `${match[1]}.snapshot_bsatn`);
if (existsSync(lockPath) || !existsSync(snapshotFile) || !lstatSync(snapshotFile).isFile()) {
return [];
}
return [{name: entry.name, transaction}];
}).sort((left, right) => left.transaction < right.transaction ? -1 : left.transaction > right.transaction ? 1 : 0);
if (snapshots.length === 0) {
replicas.push({replicaId, status: 'skipped', reason: 'no-snapshot'});
continue;
}
if (!existsSync(clogDir) || !lstatSync(clogDir).isDirectory()) {
throw new Error(`replica ${replicaId} 缺少 clog 目录。`);
}
const segmentFiles = new Map();
for (const entry of readdirSync(clogDir, {withFileTypes: true})) {
const match = /^(\d{20})\.stdb\.(log|ofs)$/u.exec(entry.name);
if (!match) {
throw new Error(`commitlog 文件名不符合预期: ${entry.name}`);
}
if (!entry.isFile()) {
throw new Error(`commitlog 候选必须是普通文件: ${join(clogDir, entry.name)}`);
}
const transaction = BigInt(match[1]);
if (transaction > 0xffff_ffff_ffff_ffffn) {
throw new Error(`commitlog transaction 超出 u64: ${entry.name}`);
}
const key = transaction.toString();
const group = segmentFiles.get(key) ?? {transaction};
group[match[2]] = entry.name;
segmentFiles.set(key, group);
}
for (const group of segmentFiles.values()) {
if (group.ofs && !group.log) {
throw new Error(`commitlog offset 缺少对应 log: replica=${replicaId}, transaction=${group.transaction}`);
}
}
const segments = [...segmentFiles.values()]
.filter((group) => group.log)
.sort((left, right) => left.transaction < right.transaction ? -1 : left.transaction > right.transaction ? 1 : 0);
const latestSnapshot = snapshots.at(-1).transaction;
const boundarySegment = segments.filter((segment) => segment.transaction <= latestSnapshot).at(-1);
if (!boundarySegment) {
throw new Error(`replica ${replicaId} 无法找到覆盖 latest snapshot ${latestSnapshot} 的 commitlog 边界。`);
}
for (const snapshot of snapshots.slice(0, -1)) {
candidates.push(historyCandidate({
dataDir: resolvedDataDir,
absolutePath: join(snapshotsDir, snapshot.name),
kind: 'snapshot',
replicaId,
transaction: snapshot.transaction,
}));
}
for (const segment of segments.filter((item) => item.transaction < boundarySegment.transaction)) {
candidates.push(historyCandidate({
dataDir: resolvedDataDir,
absolutePath: join(clogDir, segment.log),
kind: 'commitlog',
replicaId,
transaction: segment.transaction,
}));
if (segment.ofs) {
candidates.push(historyCandidate({
dataDir: resolvedDataDir,
absolutePath: join(clogDir, segment.ofs),
kind: 'commitlog-offset',
replicaId,
transaction: segment.transaction,
}));
}
}
replicas.push({
replicaId,
status: 'ready',
latestSnapshot: latestSnapshot.toString(),
boundarySegment: boundarySegment.transaction.toString(),
});
}
candidates.sort((left, right) => left.path.localeCompare(right.path));
return {
dataDir: resolvedDataDir,
replicasDir: assertSafeRelativePath(resolvedDataDir, replicasDir),
replicas,
candidates,
totalSizeBytes: candidates.reduce((sum, item) => sum + BigInt(item.sizeBytes), 0n).toString(),
};
}
function runCommand(command, args, options = {}) {
const result = spawnSync(command, args, {
cwd: options.cwd ?? REPO_ROOT,
env: options.env ?? process.env,
encoding: 'utf8',
stdio: options.stdio ?? 'pipe',
shell: process.platform === 'win32',
});
if (result.error) {
throw new Error(`${command} 启动失败: ${result.error.message}`);
}
if (result.status !== 0) {
const output = `${result.stdout ?? ''}\n${result.stderr ?? ''}`.trim();
throw new Error(`${command} 退出码 ${result.status}: ${output}`);
}
return result;
}
function parseByteSize(rawValue, label) {
const value = String(rawValue ?? '').trim();
if (!value) {
return null;
}
const match = /^(\d+)(?:\s*([KMGTPE]?)(?:I?B?)?)?$/iu.exec(value);
if (!match) {
throw new Error(`${label} 必须是字节数或 K/M/G/T/P/E 后缀大小,实际: ${rawValue}`);
}
const [, amountText, unitText = ''] = match;
const multipliers = {
'': 1n,
K: 1024n,
M: 1024n ** 2n,
G: 1024n ** 3n,
T: 1024n ** 4n,
P: 1024n ** 5n,
E: 1024n ** 6n,
};
return BigInt(amountText) * multipliers[unitText.toUpperCase()];
}
function formatBytes(bytes) {
const value = BigInt(bytes);
const gib = Number(value) / (1024 ** 3);
if (gib >= 1) {
return `${gib.toFixed(1)}GiB`;
}
const mib = Number(value) / (1024 ** 2);
if (mib >= 1) {
return `${mib.toFixed(1)}MiB`;
}
return `${value}B`;
}
function getDirectorySizeBytes(dataDir) {
const result = runCommand('du', ['-sk', dataDir]);
const [sizeKbText] = String(result.stdout ?? '').trim().split(/\s+/u);
if (!sizeKbText || !/^\d+$/u.test(sizeKbText)) {
throw new Error(`无法解析数据目录大小: ${result.stdout}`);
}
return BigInt(sizeKbText) * 1024n;
}
function getAvailableBytes(fileSystemPath) {
const stat = statfsSync(fileSystemPath, {bigint: true});
return stat.bavail * stat.bsize;
}
function parseSafetyRatio(rawValue) {
const value = String(rawValue ?? '').trim();
if (!value) {
return DEFAULT_SPACE_SAFETY_RATIO;
}
const ratio = Number(value);
if (!Number.isFinite(ratio) || ratio < 1) {
throw new Error(`GENARRATIVE_DATABASE_BACKUP_SPACE_SAFETY_RATIO 必须是 >= 1 的数字,实际: ${rawValue}`);
}
return ratio;
}
function calculateRequiredFreeBytes({dataSizeBytes, args, env}) {
const explicitMinFreeBytes = parseByteSize(
firstNonEmpty(args.minFreeBytes, env.GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES),
'GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES',
);
if (explicitMinFreeBytes !== null) {
return explicitMinFreeBytes;
}
const ratio = parseSafetyRatio(env.GENARRATIVE_DATABASE_BACKUP_SPACE_SAFETY_RATIO);
const ratioBasisPoints = BigInt(Math.ceil(ratio * 10000));
const ratioRequirement = (dataSizeBytes * ratioBasisPoints + 9999n) / 10000n;
const extraFreeBytes = parseByteSize(
firstNonEmpty(env.GENARRATIVE_DATABASE_BACKUP_EXTRA_FREE_BYTES, String(DEFAULT_EXTRA_FREE_BYTES)),
'GENARRATIVE_DATABASE_BACKUP_EXTRA_FREE_BYTES',
);
const extraRequirement = dataSizeBytes + extraFreeBytes;
return ratioRequirement > extraRequirement ? ratioRequirement : extraRequirement;
}
function assertSufficientWorkDirSpace({dataDir, workDir, args, env}) {
mkdirSync(workDir, {recursive: true});
const dataSizeBytes = getDirectorySizeBytes(dataDir);
const availableBytes = getAvailableBytes(workDir);
const requiredFreeBytes = calculateRequiredFreeBytes({dataSizeBytes, args, env});
console.log(
`[database-backup] 备份空间预检: data=${formatBytes(dataSizeBytes)}, available=${formatBytes(availableBytes)}, required=${formatBytes(requiredFreeBytes)}`,
);
if (availableBytes < requiredFreeBytes) {
throw new Error(
[
`备份工作目录所在文件系统剩余空间不足: ${workDir}`,
`available=${formatBytes(availableBytes)}`,
`required=${formatBytes(requiredFreeBytes)}`,
`dataDir=${dataDir}`,
`dataSize=${formatBytes(dataSizeBytes)}`,
'为避免停库后写满磁盘,本次备份已在停止服务前中止。',
].join(''),
);
}
}
function assertSufficientHistoryWorkDirSpace({historySizeBytes, workDir, args, env}) {
mkdirSync(workDir, {recursive: true});
const availableBytes = getAvailableBytes(workDir);
const requiredFreeBytes = calculateRequiredFreeBytes({dataSizeBytes: BigInt(historySizeBytes), args, env});
console.log(
`[database-backup] history 空间预检: candidates=${formatBytes(historySizeBytes)}, available=${formatBytes(availableBytes)}, required=${formatBytes(requiredFreeBytes)}`,
);
if (availableBytes < requiredFreeBytes) {
throw new Error(`history 工作目录剩余空间不足: available=${formatBytes(availableBytes)}required=${formatBytes(requiredFreeBytes)}`);
}
}
function collectRestartServicesAfterBackup({args, env}) {
const serviceNames = [
...String(env.GENARRATIVE_DATABASE_BACKUP_RESTART_SERVICE_AFTER ?? '')
.split(',')
.map((value) => value.trim())
.filter(Boolean),
...args.restartServicesAfter,
];
return [...new Set(serviceNames.filter(Boolean))];
}
function databaseBackupStopMarkerPath(workDir) {
return resolvePath(firstNonEmpty(
process.env.GENARRATIVE_DATABASE_BACKUP_STOP_MARKER,
workDir === DEFAULT_PRODUCTION_WORK_DIR
? DEFAULT_DATABASE_BACKUP_STOP_MARKER
: join(workDir, '.spacetimedb-stopped'),
));
}
function writeDatabaseBackupStopMarker(markerPath, serviceName) {
atomicWriteJson(markerPath, {
serviceName,
pid: process.pid,
stoppedAt: new Date().toISOString(),
});
}
function clearDatabaseBackupStopMarker(markerPath) {
if (markerPath) {
rmSync(markerPath, {force: true});
}
}
function stopServiceIfNeeded(serviceName, stopMarkerPath) {
if (!serviceName) {
return false;
}
console.log(`[database-backup] 停止服务以获取冷备份: ${serviceName}`);
writeDatabaseBackupStopMarker(stopMarkerPath, serviceName);
// stop 命令失败时仍保留 markersystemd 的 ExecStopPost 需要它判断是否要
// 兜底恢复,不能因为当前进程还能捕获异常就抹掉上一次停库证据。
runCommand('systemctl', ['stop', serviceName], {stdio: 'inherit'});
return true;
}
function startServiceIfNeeded(serviceName, wasStopped) {
if (!serviceName || !wasStopped) {
return;
}
console.log(`[database-backup] 恢复服务: ${serviceName}`);
runCommand('systemctl', ['start', serviceName], {stdio: 'inherit'});
}
function restartServicesAfterBackup(serviceNames) {
const errors = [];
for (const serviceName of serviceNames) {
if (!serviceName) {
continue;
}
console.log(`[database-backup] 冷备份后重启依赖服务: ${serviceName}`);
try {
runCommand('systemctl', ['restart', serviceName], {stdio: 'inherit'});
} catch (error) {
errors.push(error);
}
}
if (errors.length > 0) {
throw new AggregateError(errors, `冷备份后重启依赖服务失败: ${errors.map((error) => error.message).join('; ')}`);
}
}
function restoreServicesAfterBackup({stopService, serviceStopped, restartServicesAfter, stopMarkerPath}) {
const errors = [];
try {
startServiceIfNeeded(stopService, serviceStopped);
} catch (error) {
errors.push(error);
}
try {
restartServicesAfterBackup(restartServicesAfter);
} catch (error) {
errors.push(error);
}
if (errors.length > 0) {
throw new AggregateError(errors, `恢复冷备份相关服务失败: ${errors.map((error) => error.message).join('; ')}`);
}
clearDatabaseBackupStopMarker(stopMarkerPath);
}
function createArchive({dataDir, workDir, fileName}) {
if (!existsSync(dataDir)) {
throw new Error(`数据库数据目录不存在: ${dataDir}`);
}
const stat = statSync(dataDir);
if (!stat.isDirectory()) {
throw new Error(`数据库数据路径不是目录: ${dataDir}`);
}
mkdirSync(workDir, {recursive: true});
const archivePath = resolve(workDir, fileName);
const parentDir = dirname(dataDir);
const entryName = basename(dataDir);
console.log(`[database-backup] 打包: ${dataDir} -> ${archivePath}`);
runCommand('tar', ['-czf', archivePath, '-C', parentDir, entryName], {stdio: 'inherit'});
verifyArchive(archivePath);
return archivePath;
}
function verifyArchive(archivePath) {
console.log(`[database-backup] 校验归档: ${archivePath}`);
runCommand('tar', ['-tzf', archivePath], {stdio: 'ignore'});
}
function historyBatchId({baselineId, plan}) {
const identity = plan.candidates.map((candidate) => [
candidate.path,
candidate.kind,
candidate.transaction,
candidate.fingerprint,
].join('\0')).join('\n');
return sha256Hex(`${baselineId}\0${identity}`).slice(0, 32);
}
function buildHistoryNames({database, objectPrefix, baselineId, batchId}) {
const databasePart = sanitizeObjectPart(database, 'spacetimedb');
const prefix = String(objectPrefix || 'database-backups')
.trim()
.replace(/^\/+|\/+$/gu, '')
.split('/')
.filter(Boolean)
.map((part) => sanitizeObjectPart(part, 'backup'))
.join('/');
const fileName = `${databasePart}-history-${batchId}.tar.gz`;
return {
fileName,
objectKey: [prefix, databasePart, 'history', baselineId, fileName].filter(Boolean).join('/'),
};
}
function createHistoryArchive({dataDir, workDir, fileName, manifestPath, candidates}) {
mkdirSync(workDir, {recursive: true});
const archivePath = resolve(workDir, fileName);
const candidatePaths = candidates.map((candidate) => candidate.path);
console.log(`[database-backup] 打包 history: ${candidatePaths.length} 个候选 -> ${archivePath}`);
runCommand('tar', [
'-czf',
archivePath,
'-C',
dataDir,
...candidatePaths,
'-C',
dirname(manifestPath),
basename(manifestPath),
], {stdio: 'inherit'});
verifyArchive(archivePath);
return archivePath;
}
function recordHistoryBatch({statePath, state, manifest, uploadResult, manifestUpload, status, cleanedAt = ''}) {
const batch = {
batchId: manifest.batchId,
objectKey: uploadResult.objectKey,
contentLength: uploadResult.contentLength,
archiveSha256: uploadResult.archiveSha256,
verifiedAt: uploadResult.verifiedAt,
manifestObjectKey: manifestUpload.objectKey,
manifestContentLength: manifestUpload.contentLength,
manifestArchiveSha256: manifestUpload.archiveSha256,
manifestVerifiedAt: manifestUpload.verifiedAt,
uploadedAt: manifest.uploadedAt,
status,
cleanedAt,
candidates: manifest.candidates,
};
const batches = state.batches.filter((item) => item.batchId !== batch.batchId);
batches.push(batch);
const nextState = {...state, updatedAt: new Date().toISOString(), batches};
atomicWriteJson(statePath, nextState);
return nextState;
}
function candidateKey(candidate) {
return `${candidate.kind}\0${candidate.path}`;
}
export function cleanupHistoryCandidates({dataDir, candidates}) {
const currentPlan = discoverHistoryPlan({dataDir});
const eligible = new Map(currentPlan.candidates.map((candidate) => [candidateKey(candidate), candidate]));
const existing = [];
for (const candidate of candidates) {
const absolutePath = resolve(dataDir, candidate.path);
assertSafeRelativePath(dataDir, absolutePath);
if (!existsSync(absolutePath)) {
continue;
}
const current = eligible.get(candidateKey(candidate));
if (!current) {
throw new Error(`history 候选已不在当前安全边界内,拒绝删除: ${candidate.path}`);
}
const currentStat = statFingerprint(absolutePath);
if (currentStat.fingerprint !== candidate.fingerprint || currentStat.sizeBytes !== candidate.sizeBytes) {
throw new Error(`history 候选 stat 漂移,拒绝删除: ${candidate.path}`);
}
existing.push({candidate, absolutePath});
}
existing.sort((left, right) => {
const priority = {'commitlog-offset': 0, commitlog: 1, snapshot: 2};
return (priority[left.candidate.kind] ?? 3) - (priority[right.candidate.kind] ?? 3)
|| left.candidate.path.localeCompare(right.candidate.path);
});
for (const {candidate, absolutePath} of existing) {
rmSync(absolutePath, {recursive: candidate.kind === 'snapshot', force: false});
console.log(`[database-backup] 已清理 history 源文件: ${candidate.path}`);
}
return {deletedCount: existing.length, alreadyMissingCount: candidates.length - existing.length};
}
function writeManifest({manifestPath, payload}) {
writeFileSync(manifestPath, `${JSON.stringify(payload, null, 2)}\n`, 'utf8');
}
function readManifest(manifestPath) {
if (!existsSync(manifestPath)) {
throw new Error(`备份清单不存在: ${manifestPath}`);
}
return JSON.parse(readFileSync(manifestPath, 'utf8'));
}
function hmac(key, content, encoding) {
return createHmac('sha256', key).update(content).digest(encoding);
}
function sha256Hex(content) {
return createHash('sha256').update(content).digest('hex');
}
async function sha256FileHex(filePath) {
const hash = createHash('sha256');
for await (const chunk of createReadStream(filePath)) {
hash.update(chunk);
}
return hash.digest('hex');
}
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()
.replace(/^\/+|\/+$/gu, '')
.split('/')
.filter(Boolean)
.map((part) => sanitizeObjectPart(part, 'backup'))
.join('/');
return [prefix, sanitizeObjectPart(database, 'spacetimedb')].filter(Boolean).join('/');
}
function directFileIdentity(filePath) {
const stat = lstatSync(filePath, {bigint: true});
if (!stat.isFile() || stat.isSymbolicLink()) {
throw new Error(`files 模式只允许普通文件: ${filePath}`);
}
return {
dev: stat.dev.toString(),
ino: stat.ino.toString(),
size: stat.size.toString(),
mtimeNs: stat.mtimeNs.toString(),
mode: Number(stat.mode & 0o7777n),
};
}
function sameDirectFileIdentity(left, right) {
return left.dev === right.dev
&& left.ino === right.ino
&& left.size === right.size
&& left.mtimeNs === right.mtimeNs
&& left.mode === right.mode;
}
export async function collectDirectFileEntries({dataDir, candidates = null, objectPrefix, database}) {
const resolvedDataDir = resolvePath(dataDir);
if (!existsSync(resolvedDataDir) || !lstatSync(resolvedDataDir).isDirectory()) {
throw new Error(`files 数据目录不存在或不是目录: ${resolvedDataDir}`);
}
const files = new Map();
const symlinks = new Map();
const directories = new Set(['.']);
const roots = candidates === null
? [{absolutePath: resolvedDataDir, relativePath: '.'}]
: candidates.map((candidate) => ({
absolutePath: resolve(resolvedDataDir, candidate.path),
relativePath: assertSafeRelativePath(resolvedDataDir, resolve(resolvedDataDir, candidate.path)),
}));
const visit = async (absolutePath, relativePath) => {
const stat = lstatSync(absolutePath);
if (stat.isSymbolicLink()) {
const target = readlinkSync(absolutePath, 'utf8');
if (!target || isAbsolute(target)) {
throw new Error(`files 模式只允许 data-dir 内部的相对符号链接: ${absolutePath} -> ${target}`);
}
assertSafeRelativePath(resolvedDataDir, resolve(dirname(absolutePath), target));
symlinks.set(relativePath, {path: relativePath, target});
return;
}
if (stat.isDirectory()) {
directories.add(relativePath);
for (const name of readdirSync(absolutePath).sort()) {
const childRelative = relativePath === '.' ? name : `${relativePath}/${name}`;
await visit(join(absolutePath, name), childRelative);
}
return;
}
if (!stat.isFile()) {
throw new Error(`files 模式只允许普通文件或目录: ${absolutePath}`);
}
const before = directFileIdentity(absolutePath);
const sha256 = await sha256FileHex(absolutePath);
const after = directFileIdentity(absolutePath);
if (!sameDirectFileIdentity(before, after)) {
throw new Error(`files 扫描期间源文件发生变化: ${relativePath}`);
}
const basePrefix = normalizeObjectPrefix(objectPrefix, database);
const file = {
path: relativePath,
sizeBytes: Number(after.size),
sha256,
mode: after.mode,
objectKey: `${basePrefix}/files/sha256/${sha256.slice(0, 2)}/${sha256}`,
};
// 上传前后的 inode/stat 仍用于防止在线扫描漂移,但设为不可枚举,避免
// 把仅供本地校验的副本再次写入 catalog 或 result JSON。
Object.defineProperty(file, 'sourceStat', {value: after, enumerable: false});
files.set(relativePath, file);
};
for (const root of roots.sort((left, right) => left.relativePath.localeCompare(right.relativePath))) {
if (!existsSync(root.absolutePath)) {
throw new Error(`files 候选在扫描前消失: ${root.relativePath}`);
}
await visit(root.absolutePath, root.relativePath);
}
return {
directories: [...directories].sort(),
files: [...files.values()].sort((left, right) => left.path.localeCompare(right.path)),
symlinks: [...symlinks.values()].sort((left, right) => left.path.localeCompare(right.path)),
};
}
function directCatalogIdentity({mode, baselineCatalogId, rootName, directories, files, symlinks}) {
// 不把数十万条文件元数据先拼成一个巨型 JSON 字符串;分段写入 hash
// 保持与 JSON.stringify 同样的字段顺序和转义结果,同时把峰值降到单条记录。
const hash = createHash('sha256');
hash.update('{"mode":');
hash.update(JSON.stringify(mode));
hash.update(',"baselineCatalogId":');
hash.update(JSON.stringify(baselineCatalogId || ''));
hash.update(',"rootName":');
hash.update(JSON.stringify(rootName));
hash.update(',"directories":');
updateJsonArrayHash(hash, directories, (directory) => JSON.stringify(directory));
hash.update(',"files":');
updateJsonArrayHash(hash, files, (file) => JSON.stringify({
path: file.path,
sizeBytes: file.sizeBytes,
sha256: file.sha256,
mode: file.mode,
objectKey: file.objectKey,
}));
hash.update(',"symlinks":');
updateJsonArrayHash(hash, symlinks, (symlink) => JSON.stringify(symlink));
hash.update('}');
return hash.digest('hex');
}
function updateJsonArrayHash(hash, values, serialize) {
hash.update('[');
values.forEach((value, index) => {
if (index > 0) {
hash.update(',');
}
hash.update(serialize(value));
});
hash.update(']');
}
function readDirectFilesState(statePath, {database, bucket}) {
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 = readJsonOrGzip(existingPath);
if (
![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
) {
throw new Error(`files state 与本次数据源或 bucket 不匹配: ${statePath}`);
}
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,
uploadOptions,
previousFile,
verifyCatalogReuse = false,
uploadFn,
verifyFn,
}) {
const absolutePath = resolve(dataDir, file.path);
assertSafeRelativePath(dataDir, absolutePath);
if (previousFile?.sha256 === file.sha256
&& previousFile?.sizeBytes === file.sizeBytes
&& previousFile?.objectKey === file.objectKey) {
if (verifyCatalogReuse) {
await verifyFn({
...uploadOptions,
objectKey: file.objectKey,
contentLength: file.sizeBytes,
archiveSha256: file.sha256,
});
}
return {
status: verifyCatalogReuse ? 'catalog-reused-verified' : 'catalog-reused',
objectKey: file.objectKey,
};
}
try {
await verifyFn({
...uploadOptions,
objectKey: file.objectKey,
contentLength: file.sizeBytes,
archiveSha256: file.sha256,
});
return {status: 'oss-reused', objectKey: file.objectKey};
} catch (error) {
if (error?.status !== 404) {
throw error;
}
}
const beforeUpload = directFileIdentity(absolutePath);
if (!sameDirectFileIdentity(beforeUpload, file.sourceStat)) {
throw new Error(`files 上传前源文件 stat 漂移: ${file.path}`);
}
await uploadFn({
archivePath: absolutePath,
...uploadOptions,
objectKey: file.objectKey,
archiveSha256: file.sha256,
backupKind: 'spacetimedb-direct-file',
contentType: 'application/octet-stream',
allowEmpty: true,
});
const afterUpload = directFileIdentity(absolutePath);
if (!sameDirectFileIdentity(afterUpload, file.sourceStat)) {
throw new Error(`files 上传期间源文件 stat 漂移: ${file.path}`);
}
return {status: 'uploaded', objectKey: file.objectKey};
}
async function ensureDirectManifest({manifestPath, objectKey, uploadOptions, uploadManifestFn, verifyFn}) {
const body = readFileSync(manifestPath);
const archiveSha256 = sha256Hex(body);
try {
const verification = await verifyFn({
...uploadOptions,
objectKey,
contentLength: body.length,
archiveSha256,
});
return {objectKey, contentLength: body.length, archiveSha256, verifiedAt: verification.verifiedAt, reused: true};
} catch (error) {
if (error?.status !== 404) {
throw error;
}
}
return uploadManifestFn({manifestPath, ...uploadOptions, objectKey});
}
function directCatalogRef(catalog) {
return {
mode: catalog.mode,
catalogId: catalog.catalogId,
objectKey: catalog.objectKey,
contentLength: catalog.contentLength,
sha256: catalog.sha256,
verifiedAt: catalog.verifiedAt,
};
}
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: Array.isArray(catalog.files) ? catalog.files.length : 0,
symlinkCount: Array.isArray(catalog.symlinks) ? catalog.symlinks.length : 0,
},
};
}
function assertDirectCatalogRef(catalog, expectedMode, label) {
if (
!catalog
|| catalog.mode !== expectedMode
|| !/^[a-f0-9]{64}$/u.test(catalog.catalogId)
|| typeof catalog.objectKey !== 'string'
|| !catalog.objectKey
|| !Number.isSafeInteger(catalog.contentLength)
|| catalog.contentLength <= 0
|| !/^[a-f0-9]{64}$/u.test(catalog.sha256)
|| typeof catalog.verifiedAt !== 'string'
|| !catalog.verifiedAt
) {
throw new Error(`files ${label} catalog ref 无效。`);
}
return directCatalogRef(catalog);
}
function buildDirectFilesLatest({database, bucket, state}) {
const latestFullCatalog = assertDirectCatalogRef(state?.latestCatalog, 'full', 'latest full');
const historyCatalogs = (state?.historyCatalogs ?? []).map((catalog) => (
assertDirectCatalogRef(catalog, 'history', 'history')
));
return {
schemaVersion: DIRECT_FILES_LATEST_SCHEMA_VERSION,
backupKind: 'spacetimedb-direct-files-latest',
database,
bucket,
updatedAt: new Date().toISOString(),
latestFullCatalog,
historyCatalogs,
};
}
function validateDirectFilesLatest(latest, {database, bucket}) {
if (
latest?.schemaVersion !== DIRECT_FILES_LATEST_SCHEMA_VERSION
|| latest.backupKind !== 'spacetimedb-direct-files-latest'
|| latest.database !== database
|| latest.bucket !== bucket
|| !Array.isArray(latest.historyCatalogs)
) {
throw new Error('files latest pointer 契约无效。');
}
return {
...latest,
latestFullCatalog: assertDirectCatalogRef(latest.latestFullCatalog, 'full', 'latest full'),
historyCatalogs: latest.historyCatalogs.map((catalog) => assertDirectCatalogRef(catalog, 'history', 'history')),
};
}
async function publishDirectFilesLatest({
workDir,
database,
bucket,
objectPrefix,
state,
uploadOptions,
uploadManifestFn,
verifyFn,
}) {
const latest = buildDirectFilesLatest({database, bucket, state});
for (const catalogRef of [latest.latestFullCatalog, ...latest.historyCatalogs]) {
await verifyFn({
...uploadOptions,
objectKey: catalogRef.objectKey,
contentLength: catalogRef.contentLength,
archiveSha256: catalogRef.sha256,
});
}
const latestPath = join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-latest.json`);
const latestObjectKey = `${normalizeObjectPrefix(objectPrefix, database)}/latest.json`;
writeManifest({manifestPath: latestPath, payload: latest});
const uploaded = await uploadManifestFn({manifestPath: latestPath, ...uploadOptions, objectKey: latestObjectKey});
const verification = await verifyFn({
...uploadOptions,
objectKey: latestObjectKey,
contentLength: uploaded.contentLength,
archiveSha256: uploaded.archiveSha256,
});
return {
latest,
latestPath,
latestObjectKey,
contentLength: uploaded.contentLength,
sha256: uploaded.archiveSha256,
verifiedAt: verification.verifiedAt,
};
}
export async function runDirectFilesBackup({
mode,
dataDir,
workDir,
database,
bucket,
objectPrefix,
dryRun = false,
resultFile = '',
uploadOptions,
uploadFn = uploadDirectFile,
uploadManifestFn = uploadManifestFile,
verifyFn = verifyOssObject,
concurrency = 1,
}) {
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}`);
}
const plan = mode === 'history' ? discoverHistoryPlan({dataDir}) : null;
const collected = await collectDirectFileEntries({
dataDir,
candidates: plan?.candidates ?? null,
objectPrefix,
database,
});
const baselineCatalogId = mode === 'history' ? (state?.baselineCatalog?.catalogId ?? '') : '';
const rootName = basename(dataDir);
const catalogId = directCatalogIdentity({mode, baselineCatalogId, rootName, ...collected});
const basePrefix = normalizeObjectPrefix(objectPrefix, database);
const catalogObjectKey = `${basePrefix}/catalogs/${mode}/${catalogId}.json`;
const catalogPath = join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-${mode}-${catalogId}.catalog.json`);
const catalog = {
schemaVersion: DIRECT_FILES_CATALOG_SCHEMA_VERSION,
backupKind: mode === 'full' ? 'spacetimedb-data-dir-files' : 'spacetimedb-history-files',
database,
bucket,
mode,
catalogId,
catalogObjectKey,
baselineCatalogId,
rootName,
directories: collected.directories,
files: collected.files,
symlinks: collected.symlinks,
};
writeManifest({manifestPath: catalogPath, payload: catalog});
const summary = {
statePath,
catalogPath,
catalogObjectKey,
catalogId,
fileCount: collected.files.length,
symlinkCount: collected.symlinks.length,
totalSizeBytes: collected.files.reduce((sum, file) => sum + BigInt(file.sizeBytes), 0n).toString(),
candidateCount: plan?.candidates.length ?? 0,
};
if (resultFile) {
atomicWriteJson(resolvePath(resultFile), {...summary, dryRun});
}
console.log(`[database-backup] files ${mode}: files=${summary.fileCount}, symlinks=${summary.symlinkCount}, size=${formatBytes(summary.totalSizeBytes)}, catalog=${catalogId}`);
if (dryRun) {
console.log('[database-backup] files dry-run,仅扫描并生成本地 catalog,不上传或删除。');
return {...summary, catalog, uploadedCount: 0, reusedCount: 0};
}
if (mode === 'history' && plan.candidates.length === 0) {
await verifyFn({
...uploadOptions,
objectKey: state.latestCatalog.objectKey,
contentLength: state.latestCatalog.contentLength,
archiveSha256: state.latestCatalog.sha256,
});
const latestPointer = await publishDirectFilesLatest({
workDir,
database,
bucket,
objectPrefix,
state,
uploadOptions,
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,
metadataCleanup,
};
if (resultFile) {
atomicWriteJson(resolvePath(resultFile), compactDirectFilesResult(emptyResult));
}
return emptyResult;
}
if (state?.latestCatalog?.catalogId === catalogId && state.latestCatalog.mode === mode) {
await verifyFn({...uploadOptions, objectKey: state.latestCatalog.objectKey, contentLength: state.latestCatalog.contentLength, archiveSha256: state.latestCatalog.sha256});
if (mode === 'full') {
const latestPointer = await publishDirectFilesLatest({
workDir,
database,
bucket,
objectPrefix,
state,
uploadOptions,
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,
metadataCleanup,
};
}
}
if (state?.latestCatalog) {
await verifyFn({
...uploadOptions,
objectKey: state.latestCatalog.objectKey,
contentLength: state.latestCatalog.contentLength,
archiveSha256: state.latestCatalog.sha256,
});
}
const previousFiles = new Map(directPreviousFiles({state, workDir, database}).map((file) => [file.path, file]));
let uploadedCount = 0;
let reusedCount = 0;
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}`);
}
}
});
await Promise.all(workers);
const catalogUpload = await ensureDirectManifest({
manifestPath: catalogPath,
objectKey: catalogObjectKey,
uploadOptions,
uploadManifestFn,
verifyFn,
});
await verifyFn({...uploadOptions, objectKey: catalogObjectKey, contentLength: catalogUpload.contentLength, archiveSha256: catalogUpload.archiveSha256});
if (mode === 'history') {
await verifyFn({
...uploadOptions,
objectKey: state.baselineCatalog.objectKey,
contentLength: state.baselineCatalog.contentLength,
archiveSha256: state.baselineCatalog.sha256,
});
}
const catalogRef = {
mode,
catalogId,
objectKey: catalogObjectKey,
contentLength: catalogUpload.contentLength,
sha256: catalogUpload.archiveSha256,
verifiedAt: catalogUpload.verifiedAt,
};
const nextState = {
schemaVersion: DIRECT_FILES_STATE_SCHEMA_VERSION,
backupKind: 'spacetimedb-direct-files-state',
database,
dataDir,
bucket,
updatedAt: new Date().toISOString(),
baselineCatalog: directCatalogRef(state?.baselineCatalog ?? catalogRef),
latestCatalog: directCatalogRef(mode === 'full' ? catalogRef : state.latestCatalog),
historyCatalogs: mode === 'history'
? [
...(state.historyCatalogs ?? [])
.filter((item) => item.catalogId !== catalogId)
.map((item) => directCatalogRef(item)),
directCatalogRef(catalogRef),
]
: (state?.historyCatalogs ?? []).map((item) => directCatalogRef(item)),
};
const latestPointer = await publishDirectFilesLatest({
workDir,
database,
bucket,
objectPrefix,
state: nextState,
uploadOptions,
uploadManifestFn,
verifyFn,
});
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,
metadataCleanup,
};
if (resultFile) {
atomicWriteJson(resolvePath(resultFile), compactDirectFilesResult(finalResult));
}
return finalResult;
}
async function downloadOssBuffer({objectKey, uploadOptions}) {
const response = await signedOssRequest({
...ossRequestDefaults(uploadOptions),
method: 'GET',
objectKey,
operation: '下载对象',
});
return Buffer.from(await response.arrayBuffer());
}
async function downloadOssFile({objectKey, destinationPath, uploadOptions}) {
const response = await signedOssRequest({
...ossRequestDefaults(uploadOptions),
method: 'GET',
objectKey,
operation: '下载对象',
});
const tempPath = `${destinationPath}.partial-${process.pid}`;
rmSync(tempPath, {force: true});
try {
if (response.body) {
await pipeline(Readable.fromWeb(response.body), createWriteStream(tempPath, {mode: 0o600}));
} else {
writeFileSync(tempPath, Buffer.alloc(0), {mode: 0o600});
}
renameSync(tempPath, destinationPath);
} catch (error) {
rmSync(tempPath, {force: true});
throw error;
}
}
async function loadDirectFilesCatalog({catalogRef, database, bucket, uploadOptions, downloadBufferFn}) {
const catalogBody = await downloadBufferFn({objectKey: catalogRef.objectKey, uploadOptions});
if (catalogBody.length !== catalogRef.contentLength || sha256Hex(catalogBody) !== catalogRef.sha256) {
throw new Error(`files restore catalog 长度或 SHA-256 不一致: ${catalogRef.objectKey}`);
}
const catalog = JSON.parse(catalogBody.toString('utf8'));
if (
catalog.schemaVersion !== DIRECT_FILES_CATALOG_SCHEMA_VERSION
|| catalog.backupKind !== 'spacetimedb-data-dir-files'
|| catalog.database !== database
|| catalog.bucket !== bucket
|| catalog.catalogId !== catalogRef.catalogId
|| !Array.isArray(catalog.directories)
|| !Array.isArray(catalog.files)
) {
throw new Error(`files restore catalog 契约无效: ${catalogRef.objectKey}`);
}
return {...catalog, symlinks: catalog.symlinks ?? []};
}
function assertDirectCatalogFile(file, index) {
if (
!file
|| typeof file.path !== 'string'
|| !Number.isSafeInteger(file.sizeBytes)
|| file.sizeBytes < 0
|| !/^[a-f0-9]{64}$/u.test(file.sha256)
|| typeof file.objectKey !== 'string'
|| !file.objectKey
|| !Number.isSafeInteger(file.mode)
) {
throw new Error(`files restore catalog 文件项无效: index=${index}`);
}
}
function assertDirectCatalogSymlink(symlink, index, restoreDir) {
if (
!symlink
|| typeof symlink.path !== 'string'
|| !symlink.path
|| typeof symlink.target !== 'string'
|| !symlink.target
|| isAbsolute(symlink.target)
) {
throw new Error(`files restore catalog 符号链接项无效: index=${index}`);
}
const destinationPath = resolve(restoreDir, symlink.path);
assertSafeRelativePath(restoreDir, destinationPath);
assertSafeRelativePath(restoreDir, resolve(dirname(destinationPath), symlink.target));
}
async function restoreDirectFilesCatalog({
catalog,
restoreDir,
uploadOptions,
resultFile = '',
dryRun = false,
downloadFileFn = downloadOssFile,
}) {
const resolvedRestoreDir = resolvePath(restoreDir);
catalog.files.forEach(assertDirectCatalogFile);
catalog.symlinks.forEach((symlink, index) => assertDirectCatalogSymlink(symlink, index, resolvedRestoreDir));
const totalSizeBytes = catalog.files.reduce((sum, file) => sum + BigInt(file.sizeBytes), 0n).toString();
if (dryRun) {
const result = {
restoreDir: resolvedRestoreDir,
catalogId: catalog.catalogId,
fileCount: catalog.files.length,
symlinkCount: catalog.symlinks.length,
totalSizeBytes,
downloadedCount: 0,
reusedCount: 0,
dryRun: true,
};
if (resultFile) {
atomicWriteJson(resolvePath(resultFile), result);
}
return result;
}
mkdirSync(resolvedRestoreDir, {recursive: true, mode: 0o700});
for (const directoryPath of catalog.directories) {
if (directoryPath === '.') {
continue;
}
const absolutePath = resolve(resolvedRestoreDir, directoryPath);
assertSafeRelativePath(resolvedRestoreDir, absolutePath);
mkdirSync(absolutePath, {recursive: true});
}
let downloadedCount = 0;
let reusedCount = 0;
for (const [index, file] of catalog.files.entries()) {
const destinationPath = resolve(resolvedRestoreDir, file.path);
assertSafeRelativePath(resolvedRestoreDir, destinationPath);
mkdirSync(dirname(destinationPath), {recursive: true});
let reusable = false;
if (existsSync(destinationPath) && lstatSync(destinationPath).isFile()) {
const stat = statSync(destinationPath);
reusable = stat.size === file.sizeBytes && await sha256FileHex(destinationPath) === file.sha256;
}
if (reusable) {
reusedCount += 1;
} else {
rmSync(destinationPath, {force: true});
await downloadFileFn({objectKey: file.objectKey, destinationPath, uploadOptions});
const stat = statSync(destinationPath);
const sha256 = await sha256FileHex(destinationPath);
if (stat.size !== file.sizeBytes || sha256 !== file.sha256) {
rmSync(destinationPath, {force: true});
throw new Error(`files restore 对象长度或 SHA-256 不一致: ${file.path}`);
}
downloadedCount += 1;
}
chmodSync(destinationPath, file.mode & 0o7777);
console.log(`[database-backup] files restore: ${index + 1}/${catalog.files.length} (${reusable ? 'reused' : 'downloaded'}) ${file.path}`);
}
for (const symlink of catalog.symlinks) {
const destinationPath = resolve(resolvedRestoreDir, symlink.path);
assertSafeRelativePath(resolvedRestoreDir, destinationPath);
mkdirSync(dirname(destinationPath), {recursive: true});
rmSync(destinationPath, {recursive: true, force: true});
symlinkSync(symlink.target, destinationPath);
console.log(`[database-backup] files restore: symlink ${symlink.path} -> ${symlink.target}`);
}
const result = {
restoreDir: resolvedRestoreDir,
catalogId: catalog.catalogId,
fileCount: catalog.files.length,
symlinkCount: catalog.symlinks.length,
totalSizeBytes,
downloadedCount,
reusedCount,
};
if (resultFile) {
atomicWriteJson(resolvePath(resultFile), result);
}
return result;
}
export async function restoreDirectFilesBackup({
statePath,
restoreDir,
database,
bucket,
uploadOptions,
resultFile = '',
dryRun = false,
downloadBufferFn = downloadOssBuffer,
downloadFileFn = downloadOssFile,
}) {
const state = readDirectFilesState(resolvePath(statePath), {database, bucket});
if (!state?.latestCatalog || state.latestCatalog.mode !== 'full') {
throw new Error(`files restore 缺少 full baseline catalog: ${statePath}`);
}
const catalogRef = assertDirectCatalogRef(state.latestCatalog, 'full', 'latest full');
const catalog = await loadDirectFilesCatalog({catalogRef, database, bucket, uploadOptions, downloadBufferFn});
return restoreDirectFilesCatalog({
catalog,
restoreDir,
uploadOptions,
resultFile,
dryRun,
downloadFileFn,
});
}
export async function restoreDirectFilesLatest({
restoreDir,
database,
bucket,
objectPrefix,
uploadOptions,
resultFile = '',
dryRun = false,
downloadBufferFn = downloadOssBuffer,
downloadFileFn = downloadOssFile,
verifyFn = verifyOssObject,
}) {
const latestObjectKey = `${normalizeObjectPrefix(objectPrefix, database)}/latest.json`;
const latestBody = await downloadBufferFn({objectKey: latestObjectKey, uploadOptions});
const latestSha256 = sha256Hex(latestBody);
await verifyFn({
...uploadOptions,
objectKey: latestObjectKey,
contentLength: latestBody.length,
archiveSha256: latestSha256,
});
const latest = validateDirectFilesLatest(JSON.parse(latestBody.toString('utf8')), {database, bucket});
const catalogRef = latest.latestFullCatalog;
await verifyFn({
...uploadOptions,
objectKey: catalogRef.objectKey,
contentLength: catalogRef.contentLength,
archiveSha256: catalogRef.sha256,
});
const catalog = await loadDirectFilesCatalog({catalogRef, database, bucket, uploadOptions, downloadBufferFn});
return restoreDirectFilesCatalog({
catalog,
restoreDir,
uploadOptions,
resultFile,
dryRun,
downloadFileFn,
});
}
function regionFromEndpoint(endpoint) {
const match = /^oss-([a-z0-9-]+)\./u.exec(endpoint);
if (!match) {
throw new Error(`无法从 OSS endpoint 推断 region: ${endpoint}`);
}
return match[1];
}
function formatScopeDate(date) {
return timestampForFile(date).slice(0, 8);
}
function formatOssDate(date) {
return timestampForFile(date).replace(/[-:]/gu, '');
}
function encodePath(path) {
return path
.split('/')
.map((segment) => encodeURIComponent(segment).replace(/[!'()*]/gu, (char) => `%${char.charCodeAt(0).toString(16).toUpperCase()}`))
.join('/');
}
function encodeQueryComponent(value) {
return encodeURIComponent(String(value)).replace(
/[!'()*]/gu,
(char) => `%${char.charCodeAt(0).toString(16).toUpperCase()}`,
);
}
export function buildCanonicalQuery(queries = {}) {
return Object.entries(queries)
.map(([key, value]) => [encodeQueryComponent(key), value === null ? null : encodeQueryComponent(value)])
.sort(([leftKey, leftValue], [rightKey, rightValue]) => {
if (leftKey !== rightKey) {
return leftKey < rightKey ? -1 : 1;
}
const left = leftValue ?? '';
const right = rightValue ?? '';
return left === right ? 0 : left < right ? -1 : 1;
})
.map(([key, value]) => value === null ? key : `${key}=${value}`)
.join('&');
}
function canonicalHeaderValue(value) {
return String(value).trim().replace(/\s+/gu, ' ');
}
export function buildAuthorization({method, bucket, endpoint, objectKey, accessKeyId, accessKeySecret, headers, date, queries = {}}) {
const region = regionFromEndpoint(endpoint);
const scopeDate = formatScopeDate(date);
const scope = `${scopeDate}/${region}/${OSS_SERVICE}/${OSS_REQUEST}`;
const canonicalUri = `/${encodeURIComponent(bucket)}/${encodePath(objectKey)}`;
const signedHeaders = Object.fromEntries(
Object.entries(headers).map(([key, value]) => [key.toLowerCase(), canonicalHeaderValue(value)]),
);
const canonicalHeaders = Object.entries(signedHeaders)
.sort(([left], [right]) => left.localeCompare(right))
.map(([key, value]) => `${key}:${value}\n`)
.join('');
const additionalHeaders = 'host';
const canonicalRequest = [
method,
canonicalUri,
buildCanonicalQuery(queries),
canonicalHeaders,
additionalHeaders,
UNSIGNED_PAYLOAD,
].join('\n');
const stringToSign = [OSS_ALGORITHM, headers['x-oss-date'], scope, sha256Hex(canonicalRequest)].join('\n');
const signature = hmac(Buffer.from(`aliyun_v4${accessKeySecret}`, 'utf8'), scopeDate);
const regionKey = hmac(signature, region);
const serviceKey = hmac(regionKey, OSS_SERVICE);
const signingKey = hmac(serviceKey, OSS_REQUEST);
const finalSignature = hmac(signingKey, stringToSign, 'hex');
return `${OSS_ALGORITHM} Credential=${accessKeyId}/${scope},AdditionalHeaders=${additionalHeaders},Signature=${finalSignature}`;
}
function buildOssUrl({bucket, endpoint, objectKey, queries = {}}) {
const canonicalQuery = buildCanonicalQuery(queries);
return `https://${bucket}.${endpoint}/${encodePath(objectKey)}${canonicalQuery ? `?${canonicalQuery}` : ''}`;
}
function isRetryableOssStatus(status) {
return RETRYABLE_OSS_HTTP_STATUSES.has(status);
}
function retryDelayMs({attempt, baseDelayMs, maxDelayMs, randomFn}) {
const ceiling = Math.min(maxDelayMs, baseDelayMs * (2 ** Math.max(0, attempt - 1)));
return Math.floor(randomFn() * ceiling);
}
function ossRequestDefaults({
bucket,
endpoint,
accessKeyId,
accessKeySecret,
fetchImpl = globalThis.fetch,
nowFn = () => new Date(),
sleepImpl = sleep,
randomFn = Math.random,
maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
retryBaseDelayMs = DEFAULT_OSS_RETRY_BASE_DELAY_MS,
retryMaxDelayMs = DEFAULT_OSS_RETRY_MAX_DELAY_MS,
}) {
return {
bucket,
endpoint,
accessKeyId,
accessKeySecret,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs,
retryMaxDelayMs,
};
}
async function signedOssRequest({
method,
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
queries = {},
headers = {},
bodyFactory,
contentLength,
operation,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs,
retryMaxDelayMs,
}) {
const targetUrl = buildOssUrl({bucket, endpoint, objectKey, queries});
let lastError = null;
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
const now = nowFn();
const signedHeaders = {
host: `${bucket}.${endpoint}`,
...headers,
'x-oss-content-sha256': UNSIGNED_PAYLOAD,
'x-oss-date': formatOssDate(now),
};
const authorization = buildAuthorization({
method,
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
headers: signedHeaders,
date: now,
queries,
});
const requestHeaders = {...signedHeaders, authorization};
if (contentLength !== undefined) {
requestHeaders['content-length'] = String(contentLength);
}
const body = bodyFactory ? bodyFactory() : undefined;
const requestOptions = {method, headers: requestHeaders};
if (body !== undefined) {
requestOptions.body = body;
requestOptions.duplex = 'half';
}
let response;
try {
response = await fetchImpl(targetUrl, requestOptions);
} catch (error) {
lastError = new Error(`OSS ${operation}请求失败: oss://${bucket}/${objectKey}`, {cause: error});
}
if (response?.ok) {
return response;
}
if (response) {
const responseText = await response.text();
const requestId = response.headers.get('x-oss-request-id');
lastError = new Error(
`OSS ${operation}失败 HTTP ${response.status}${requestId ? ` requestId=${requestId}` : ''}: ${responseText.slice(0, 500)}`,
);
lastError.status = response.status;
}
const retryable = !response || isRetryableOssStatus(response.status);
if (!retryable || attempt >= maxAttempts) {
throw lastError;
}
const delayMs = retryDelayMs({attempt, baseDelayMs: retryBaseDelayMs, maxDelayMs: retryMaxDelayMs, randomFn});
console.warn(`[database-backup] OSS ${operation}失败,${delayMs}ms 后重试 (${attempt}/${maxAttempts})`);
await sleepImpl(delayMs);
}
throw lastError ?? new Error(`OSS ${operation}失败: oss://${bucket}/${objectKey}`);
}
function readXmlTag(xml, tagName) {
const match = new RegExp(`<${tagName}>([\\s\\S]*?)</${tagName}>`, 'u').exec(xml);
if (!match) {
return '';
}
return match[1]
.replace(/&lt;/gu, '<')
.replace(/&gt;/gu, '>')
.replace(/&quot;/gu, '"')
.replace(/&apos;/gu, "'")
.replace(/&amp;/gu, '&')
.trim();
}
function escapeXml(value) {
return String(value)
.replace(/&/gu, '&amp;')
.replace(/</gu, '&lt;')
.replace(/>/gu, '&gt;')
.replace(/"/gu, '&quot;')
.replace(/'/gu, '&apos;');
}
function buildCompleteMultipartBody(parts) {
const partXml = parts
.map(({partNumber, etag}) => [
'<Part>',
`<PartNumber>${partNumber}</PartNumber>`,
`<ETag>${escapeXml(etag)}</ETag>`,
'</Part>',
].join(''))
.join('');
return `<?xml version="1.0" encoding="UTF-8"?><CompleteMultipartUpload>${partXml}</CompleteMultipartUpload>`;
}
function resolveMultipartPartSize(fileSize, configuredPartSize) {
if (!Number.isSafeInteger(configuredPartSize) || configuredPartSize < OSS_MIN_MULTIPART_PART_SIZE_BYTES) {
throw new Error(`OSS multipart part size 必须是 >= ${OSS_MIN_MULTIPART_PART_SIZE_BYTES} 的安全整数`);
}
const minimumForPartLimit = Math.ceil(fileSize / OSS_MAX_MULTIPART_PARTS);
const partSize = Math.max(configuredPartSize, minimumForPartLimit);
if (partSize > OSS_MAX_MULTIPART_PART_SIZE_BYTES) {
throw new Error(`OSS multipart part size 超过 5GiB: ${partSize}`);
}
return partSize;
}
async function verifyUploadedObject({requestOptions, expectedContentLength, expectedArchiveSha256}) {
const response = await signedOssRequest({
...requestOptions,
method: 'HEAD',
operation: 'HEAD 验证',
});
const contentLengthHeader = response.headers.get('content-length');
const metadataLengthHeader = response.headers.get('x-oss-meta-file-size');
const effectiveLengthHeader = /^\d+$/u.test(contentLengthHeader ?? '')
? contentLengthHeader
: metadataLengthHeader;
if (!effectiveLengthHeader || !/^\d+$/u.test(effectiveLengthHeader)) {
throw new Error(
`OSS HEAD 验证缺少有效 content-length/file-size: content-length=${contentLengthHeader ?? '<missing>'}, file-size=${metadataLengthHeader ?? '<missing>'}`,
);
}
const remoteContentLength = Number(effectiveLengthHeader);
if (remoteContentLength !== expectedContentLength) {
throw new Error(`OSS HEAD 验证长度不一致: local=${expectedContentLength}, remote=${remoteContentLength}`);
}
const remoteArchiveSha256 = String(
response.headers.get('x-oss-meta-file-sha256')
?? response.headers.get('x-oss-meta-archive-sha256')
?? '',
).trim().toLowerCase();
if (remoteArchiveSha256 !== expectedArchiveSha256) {
throw new Error(`OSS HEAD 验证 SHA-256 不一致: local=${expectedArchiveSha256}, remote=${remoteArchiveSha256 || '<missing>'}`);
}
return {verifiedAt: new Date().toISOString(), remoteContentLength, remoteArchiveSha256};
}
export async function verifyOssObject({
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
contentLength,
archiveSha256,
fetchImpl = globalThis.fetch,
nowFn = () => new Date(),
sleepImpl = sleep,
randomFn = Math.random,
maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
}) {
const requestOptions = {
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs: DEFAULT_OSS_RETRY_BASE_DELAY_MS,
retryMaxDelayMs: DEFAULT_OSS_RETRY_MAX_DELAY_MS,
};
return verifyUploadedObject({
requestOptions,
expectedContentLength: Number(contentLength),
expectedArchiveSha256: String(archiveSha256 ?? '').trim().toLowerCase(),
});
}
async function abortMultipartUpload({requestOptions, uploadId}) {
try {
await signedOssRequest({
...requestOptions,
method: 'DELETE',
queries: {uploadId},
operation: 'AbortMultipartUpload',
maxAttempts: Math.min(2, requestOptions.maxAttempts),
});
console.warn(`[database-backup] 已清理失败的 multipart upload: ${uploadId}`);
} catch (error) {
console.warn(`[database-backup] 清理 multipart upload 失败: ${error.message}`);
}
}
export async function uploadArchive({
archivePath,
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
partSizeBytes = DEFAULT_OSS_MULTIPART_PART_SIZE_BYTES,
maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
retryBaseDelayMs = DEFAULT_OSS_RETRY_BASE_DELAY_MS,
retryMaxDelayMs = DEFAULT_OSS_RETRY_MAX_DELAY_MS,
fetchImpl = globalThis.fetch,
nowFn = () => new Date(),
sleepImpl = sleep,
randomFn = Math.random,
backupKind = 'spacetimedb-data-dir',
archiveSha256 = '',
contentType = 'application/gzip',
allowEmpty = false,
bandwidthLimiter = null,
}) {
const fileStat = statSync(archivePath);
if (!fileStat.isFile() || (!allowEmpty && fileStat.size <= 0)) {
throw new Error(`待上传备份必须是${allowEmpty ? '' : '非空'}普通文件: ${archivePath}`);
}
const verifiedArchiveSha256 = archiveSha256 || await sha256FileHex(archivePath);
if (!/^[a-f0-9]{64}$/u.test(verifiedArchiveSha256)) {
throw new Error(`归档 SHA-256 无效: ${verifiedArchiveSha256}`);
}
const requestOptions = {
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs,
retryMaxDelayMs,
};
if (fileStat.size === 0) {
await signedOssRequest({
...requestOptions,
method: 'PUT',
headers: {
'content-type': contentType,
'x-oss-meta-archive-sha256': verifiedArchiveSha256,
'x-oss-meta-file-sha256': verifiedArchiveSha256,
'x-oss-meta-file-size': '0',
'x-oss-meta-backup-kind': backupKind,
},
contentLength: 0,
bodyFactory: () => Buffer.alloc(0),
operation: '上传空文件',
});
const verification = await verifyUploadedObject({requestOptions, expectedContentLength: 0, expectedArchiveSha256: verifiedArchiveSha256});
return {bucket, objectKey, contentLength: 0, archiveSha256: verifiedArchiveSha256, etag: '', uploadMode: 'single', partCount: 1, partSizeBytes: 0, verifiedAt: verification.verifiedAt};
}
const partSize = resolveMultipartPartSize(fileStat.size, partSizeBytes);
const partCount = Math.ceil(fileStat.size / partSize);
let uploadId = '';
let uploadCompleted = false;
console.log(`[database-backup] multipart 上传 OSS: oss://${bucket}/${objectKey} (${partCount} parts)`);
try {
const initiateResponse = await signedOssRequest({
...requestOptions,
method: 'POST',
queries: {uploads: null},
headers: {
'content-type': contentType,
'x-oss-meta-archive-sha256': verifiedArchiveSha256,
'x-oss-meta-file-sha256': verifiedArchiveSha256,
'x-oss-meta-file-size': String(fileStat.size),
'x-oss-meta-backup-kind': backupKind,
},
operation: 'InitiateMultipartUpload',
});
uploadId = readXmlTag(await initiateResponse.text(), 'UploadId');
if (!uploadId) {
throw new Error('OSS InitiateMultipartUpload 响应缺少 UploadId');
}
const parts = [];
for (let partNumber = 1; partNumber <= partCount; partNumber += 1) {
const start = (partNumber - 1) * partSize;
const end = Math.min(fileStat.size, start + partSize) - 1;
const contentLength = end - start + 1;
const response = await signedOssRequest({
...requestOptions,
method: 'PUT',
queries: {partNumber, uploadId},
headers: {'content-type': 'application/octet-stream'},
contentLength,
bodyFactory: () => {
const stream = createReadStream(archivePath, {start, end});
return bandwidthLimiter ? bandwidthLimiter.wrap(stream) : stream;
},
operation: `UploadPart ${partNumber}/${partCount}`,
});
const etag = response.headers.get('etag');
if (!etag) {
throw new Error(`OSS UploadPart ${partNumber}/${partCount} 响应缺少 ETag`);
}
parts.push({partNumber, etag});
console.log(`[database-backup] multipart 进度: ${partNumber}/${partCount}`);
}
const completeBody = buildCompleteMultipartBody(parts);
let completeResponse;
try {
completeResponse = await signedOssRequest({
...requestOptions,
method: 'POST',
queries: {uploadId},
headers: {'content-type': 'application/xml'},
contentLength: Buffer.byteLength(completeBody),
bodyFactory: () => completeBody,
operation: 'CompleteMultipartUpload',
});
const completeResponseText = await completeResponse.text();
if (/<Error(?:\s|>)/u.test(completeResponseText)) {
throw new Error(`OSS CompleteMultipartUpload 返回错误: ${completeResponseText.slice(0, 500)}`);
}
} catch (completeError) {
try {
await verifyUploadedObject({requestOptions, expectedContentLength: fileStat.size, expectedArchiveSha256: verifiedArchiveSha256});
completeResponse = null;
} catch {
throw completeError;
}
}
const verification = await verifyUploadedObject({
requestOptions,
expectedContentLength: fileStat.size,
expectedArchiveSha256: verifiedArchiveSha256,
});
uploadCompleted = true;
return {
bucket,
objectKey,
contentLength: fileStat.size,
archiveSha256: verifiedArchiveSha256,
etag: completeResponse?.headers.get('etag')?.replace(/^"|"$/gu, '') ?? '',
uploadMode: 'multipart',
partCount,
partSizeBytes: partSize,
verifiedAt: verification.verifiedAt,
};
} catch (error) {
if (uploadId && !uploadCompleted) {
await abortMultipartUpload({requestOptions, uploadId});
}
throw error;
}
}
export async function uploadDirectFile({
archivePath,
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl = globalThis.fetch,
nowFn = () => new Date(),
sleepImpl = sleep,
randomFn = Math.random,
maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
retryBaseDelayMs = DEFAULT_OSS_RETRY_BASE_DELAY_MS,
retryMaxDelayMs = DEFAULT_OSS_RETRY_MAX_DELAY_MS,
backupKind = 'spacetimedb-direct-file',
archiveSha256 = '',
contentType = 'application/octet-stream',
allowEmpty = true,
bandwidthLimiter = null,
}) {
const fileStat = statSync(archivePath);
if (!fileStat.isFile() || (!allowEmpty && fileStat.size <= 0)) {
throw new Error(`待上传备份必须是${allowEmpty ? '' : '非空'}普通文件: ${archivePath}`);
}
if (fileStat.size > DIRECT_FILES_SINGLE_PUT_MAX_BYTES) {
return uploadArchive({
archivePath,
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs,
retryMaxDelayMs,
backupKind,
archiveSha256,
contentType,
allowEmpty,
bandwidthLimiter,
});
}
const verifiedArchiveSha256 = archiveSha256 || await sha256FileHex(archivePath);
if (!/^[a-f0-9]{64}$/u.test(verifiedArchiveSha256)) {
throw new Error(`归档 SHA-256 无效: ${verifiedArchiveSha256}`);
}
const requestOptions = {
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs,
retryMaxDelayMs,
};
await signedOssRequest({
...requestOptions,
method: 'PUT',
headers: {
'content-type': contentType,
'x-oss-meta-archive-sha256': verifiedArchiveSha256,
'x-oss-meta-file-sha256': verifiedArchiveSha256,
'x-oss-meta-file-size': String(fileStat.size),
'x-oss-meta-backup-kind': backupKind,
},
contentLength: fileStat.size,
bodyFactory: () => {
if (fileStat.size === 0) {
return Buffer.alloc(0);
}
const stream = createReadStream(archivePath);
return bandwidthLimiter ? bandwidthLimiter.wrap(stream) : stream;
},
operation: '上传逐文件对象',
});
const verification = await verifyUploadedObject({
requestOptions,
expectedContentLength: fileStat.size,
expectedArchiveSha256: verifiedArchiveSha256,
});
return {
bucket,
objectKey,
contentLength: fileStat.size,
archiveSha256: verifiedArchiveSha256,
etag: '',
uploadMode: 'single',
partCount: 1,
partSizeBytes: fileStat.size,
verifiedAt: verification.verifiedAt,
};
}
export async function uploadManifestFile({
manifestPath,
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl = globalThis.fetch,
nowFn = () => new Date(),
sleepImpl = sleep,
randomFn = Math.random,
maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
bandwidthLimiter = null,
}) {
const body = readFileSync(manifestPath);
if (body.length === 0) {
throw new Error(`待上传 manifest 不能为空: ${manifestPath}`);
}
const archiveSha256 = sha256Hex(body);
const requestOptions = {
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs: DEFAULT_OSS_RETRY_BASE_DELAY_MS,
retryMaxDelayMs: DEFAULT_OSS_RETRY_MAX_DELAY_MS,
};
await signedOssRequest({
...requestOptions,
method: 'PUT',
headers: {
'content-type': 'application/json',
'x-oss-meta-archive-sha256': archiveSha256,
'x-oss-meta-file-size': String(body.length),
'x-oss-meta-backup-kind': 'spacetimedb-backup-manifest',
},
contentLength: body.length,
bodyFactory: () => {
const stream = createBufferReadStream(body);
return bandwidthLimiter ? bandwidthLimiter.wrap(stream) : stream;
},
operation: '上传 manifest',
});
const verification = await verifyUploadedObject({
requestOptions,
expectedContentLength: body.length,
expectedArchiveSha256: archiveSha256,
});
return {objectKey, contentLength: body.length, archiveSha256, verifiedAt: verification.verifiedAt};
}
function uploadedManifestPayload({manifest, database, result}) {
return {
...manifest,
database,
bucket: result.bucket,
objectKey: result.objectKey,
manifestObjectKey: `${result.objectKey}.manifest.json`,
contentLength: result.contentLength,
archiveSha256: result.archiveSha256,
etag: result.etag,
uploadMode: result.uploadMode,
partCount: result.partCount,
partSizeBytes: result.partSizeBytes,
verifiedAt: result.verifiedAt,
uploadedAt: new Date().toISOString(),
uploadStatus: 'uploaded',
};
}
export async function uploadHistoryArchiveWithCleanup({
archivePath,
manifestPath,
manifest,
statePath,
uploadOptions,
uploadFn = uploadArchive,
manifestUploadFn = uploadManifestFile,
verifyFn = verifyOssObject,
}) {
const result = await uploadFn({
archivePath,
...uploadOptions,
backupKind: 'spacetimedb-history',
});
const uploadedManifest = uploadedManifestPayload({manifest, database: manifest.database, result});
writeManifest({manifestPath, payload: uploadedManifest});
const manifestUpload = await manifestUploadFn({
manifestPath,
...uploadOptions,
objectKey: uploadedManifest.manifestObjectKey,
});
uploadedManifest.manifestVerifiedAt = manifestUpload.verifiedAt;
uploadedManifest.manifestContentLength = manifestUpload.contentLength;
uploadedManifest.manifestArchiveSha256 = manifestUpload.archiveSha256;
writeManifest({manifestPath, payload: uploadedManifest});
let state = validateHistoryState(readManifest(statePath), {
database: uploadedManifest.database,
dataDir: uploadedManifest.dataDir,
});
if (state.baseline.id !== uploadedManifest.baselineId) {
throw new Error(`history manifest baselineId 与 state 不匹配: manifest=${uploadedManifest.baselineId}, state=${state.baseline.id}`);
}
await verifyFn({
...uploadOptions,
bucket: state.baseline.bucket,
objectKey: state.baseline.objectKey,
contentLength: state.baseline.contentLength,
archiveSha256: state.baseline.archiveSha256,
});
await verifyFn({
...uploadOptions,
bucket: state.baseline.bucket,
objectKey: state.baseline.manifestObjectKey,
contentLength: state.baseline.manifestContentLength,
archiveSha256: state.baseline.manifestArchiveSha256,
});
state = recordHistoryBatch({
statePath,
state,
manifest: uploadedManifest,
uploadResult: result,
manifestUpload,
status: 'uploaded',
});
const cleanup = cleanupHistoryCandidates({
dataDir: uploadedManifest.dataDir,
candidates: uploadedManifest.candidates,
});
state = recordHistoryBatch({
statePath,
state,
manifest: uploadedManifest,
uploadResult: result,
manifestUpload,
status: 'cleaned',
cleanedAt: new Date().toISOString(),
});
return {result, uploadedManifest, cleanup, state};
}
export function discoverDeferredArchiveUploads({workDir, database, includeUploaded = false}) {
const resolvedWorkDir = resolvePath(workDir);
if (!existsSync(resolvedWorkDir)) {
return {archives: [], missingArchives: []};
}
const archives = [];
const missingArchives = [];
const manifestSuffix = '.tar.gz.manifest.json';
const expectedDatabase = String(database || '').trim();
const entries = readdirSync(resolvedWorkDir, {withFileTypes: true})
.filter((candidate) => candidate.isFile() && candidate.name.endsWith(manifestSuffix))
.sort((left, right) => left.name.localeCompare(right.name, 'en'));
for (const entry of entries) {
const manifestPath = join(resolvedWorkDir, entry.name);
const manifest = readManifest(manifestPath);
const uploadStatus = String(manifest.uploadStatus || '').trim();
if (!['deferred', 'pending'].includes(uploadStatus) && !(includeUploaded && uploadStatus === 'uploaded')) {
continue;
}
if (expectedDatabase && String(manifest.database || '').trim() !== expectedDatabase) {
continue;
}
if (!manifest.archivePath) {
throw new Error(`deferred 备份清单缺少 archivePath: ${manifestPath}`);
}
const archivePath = resolvePath(manifest.archivePath);
if (dirname(archivePath) !== resolvedWorkDir || manifestPath !== `${archivePath}.manifest.json`) {
throw new Error(`deferred 备份路径与清单不匹配: ${manifestPath}`);
}
const candidate = {archivePath, manifestPath, manifest};
if (existsSync(archivePath)) {
const archiveStat = lstatSync(archivePath);
if (!archiveStat.isFile() || archiveStat.isSymbolicLink()) {
throw new Error(`deferred 备份归档必须是非符号链接的普通文件: ${archivePath}`);
}
archives.push(candidate);
} else {
missingArchives.push(candidate);
}
}
return {archives, missingArchives};
}
async function uploadExistingArchive({args, env, bucket, endpoint, accessKeyId, accessKeySecret, objectPrefix, bandwidthLimiter}) {
const archivePath = resolvePath(args.uploadArchive);
if (!existsSync(archivePath)) {
throw new Error(`待上传备份文件不存在: ${archivePath}`);
}
const manifestPath = resolvePath(args.manifestFile || `${archivePath}.manifest.json`);
const manifest = existsSync(manifestPath) ? readManifest(manifestPath) : {};
const dataDir = firstNonEmpty(manifest.dataDir, env.GENARRATIVE_DATABASE_BACKUP_DATA_DIR, DEFAULT_PRODUCTION_DATA_DIR);
const database = firstNonEmpty(args.database, manifest.database, env.GENARRATIVE_SPACETIME_DATABASE, basename(dataDir));
const objectKey = firstNonEmpty(args.objectKey, manifest.objectKey, buildBackupNames({database, dataDir, objectPrefix}).objectKey);
if (manifest.backupKind !== 'spacetimedb-history') {
manifest.backupKind = 'spacetimedb-data-dir';
manifest.baselineStatePath = firstNonEmpty(
manifest.baselineStatePath,
historyStatePath({args, env, workDir: dirname(archivePath), database}),
);
}
console.log(`[database-backup] 上传已有备份: ${archivePath}`);
console.log(`[database-backup] 目标对象: oss://${bucket}/${objectKey}`);
if (args.dryRun) {
console.log('[database-backup] dry-run,仅校验上传配置。');
return;
}
const statePath = resolvePath(firstNonEmpty(
manifest.baselineStatePath,
historyStatePath({args, env, workDir: dirname(archivePath), database}),
));
let result;
let uploadedAt;
if (manifest.backupKind === 'spacetimedb-history') {
const historyResult = await uploadHistoryArchiveWithCleanup({
archivePath,
manifestPath,
manifest,
statePath,
uploadOptions: {bucket, endpoint, objectKey, accessKeyId, accessKeySecret, bandwidthLimiter},
});
result = historyResult.result;
uploadedAt = historyResult.uploadedManifest.uploadedAt;
console.log(`[database-backup] history 上传并清理完成: ${JSON.stringify(historyResult.cleanup)}`);
} else {
result = await uploadArchive({archivePath, bucket, endpoint, objectKey, accessKeyId, accessKeySecret, bandwidthLimiter});
const uploadedManifest = uploadedManifestPayload({manifest, database, result});
uploadedAt = uploadedManifest.uploadedAt;
writeManifest({manifestPath, payload: uploadedManifest});
const manifestUpload = await uploadManifestFile({
manifestPath,
bucket,
endpoint,
objectKey: uploadedManifest.manifestObjectKey,
accessKeyId,
accessKeySecret,
bandwidthLimiter,
});
uploadedManifest.manifestVerifiedAt = manifestUpload.verifiedAt;
uploadedManifest.manifestContentLength = manifestUpload.contentLength;
uploadedManifest.manifestArchiveSha256 = manifestUpload.archiveSha256;
writeManifest({manifestPath, payload: uploadedManifest});
const previousState = existsSync(statePath)
? validateHistoryState(readManifest(statePath), {database, dataDir})
: null;
const baseline = normalizeUploadedBaselineManifest(uploadedManifest, {database, dataDir});
writeBaselineState({statePath, baseline, previousState});
console.log(`[database-backup] 已写入 baseline state: ${statePath}`);
}
console.log(`[database-backup] 上传完成: ${JSON.stringify(result)}`);
if (args.resultFile) {
writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, statePath, ...result, uploadedAt}, null, 2)}\n`, 'utf8');
}
const keepLocal = args.keepLocal || String(env.GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL ?? '').trim().toLowerCase() === 'true';
if (!keepLocal) {
rmSync(archivePath, {force: true});
rmSync(manifestPath, {force: true});
console.log('[database-backup] 已删除本地临时备份文件;如需保留请设置 --keep-local。');
} else {
console.log(`[database-backup] 已保留本地备份: ${archivePath}`);
console.log(`[database-backup] 已保留备份清单: ${manifestPath}`);
}
}
async function uploadDeferredArchives({args, env, bucket, endpoint, accessKeyId, accessKeySecret, objectPrefix, database, bandwidthLimiter}) {
const workDir = resolvePath(args.uploadDeferredDir);
const keepLocal = args.keepLocal || String(env.GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL ?? '').trim().toLowerCase() === 'true';
const {archives, missingArchives} = discoverDeferredArchiveUploads({
workDir,
database,
includeUploaded: !keepLocal,
});
for (const {manifestPath} of missingArchives) {
console.warn(`[database-backup] deferred 清单对应的本地归档不存在,跳过: ${manifestPath}`);
}
if (archives.length === 0) {
console.log(`[database-backup] 没有可补偿的本地归档: ${workDir}`);
return;
}
console.log(`[database-backup] 开始串行上传待补偿本地归档: count=${archives.length}`);
for (const {archivePath, manifestPath} of archives) {
await uploadExistingArchive({
args: {...args, uploadArchive: archivePath, manifestFile: manifestPath},
env,
bucket,
endpoint,
accessKeyId,
accessKeySecret,
objectPrefix,
bandwidthLimiter,
});
}
console.log(`[database-backup] 待补偿本地归档上传完成: count=${archives.length}`);
}
async function publishExistingManifest({args, bucket, endpoint, accessKeyId, accessKeySecret, bandwidthLimiter}) {
const manifestPath = resolvePath(args.publishManifest);
const manifest = readManifest(manifestPath);
if (manifest.uploadStatus !== 'uploaded' || !manifest.objectKey) {
throw new Error('只允许发布 uploadStatus=uploaded 且包含 objectKey 的备份 manifest。');
}
manifest.manifestObjectKey = manifest.manifestObjectKey || `${manifest.objectKey}.manifest.json`;
writeManifest({manifestPath, payload: manifest});
const result = await uploadManifestFile({
manifestPath,
bucket,
endpoint,
objectKey: manifest.manifestObjectKey,
accessKeyId,
accessKeySecret,
bandwidthLimiter,
});
manifest.manifestVerifiedAt = result.verifiedAt;
manifest.manifestContentLength = result.contentLength;
manifest.manifestArchiveSha256 = result.archiveSha256;
writeManifest({manifestPath, payload: manifest});
console.log(`[database-backup] manifest 上传并验真完成: ${JSON.stringify(result)}`);
}
export async function resumeUploadedHistoryBatch({statePath, state, dataDir, verificationOptions, verifyFn = verifyOssObject}) {
const pendingBatch = state.batches.find((batch) => batch.status === 'uploaded');
if (!pendingBatch) {
return state;
}
console.log(`[database-backup] 重试已上传 history 批次的本地清理: ${pendingBatch.batchId}`);
await verifyFn({
...verificationOptions,
objectKey: pendingBatch.objectKey,
contentLength: pendingBatch.contentLength,
archiveSha256: pendingBatch.archiveSha256,
});
await verifyFn({
...verificationOptions,
objectKey: pendingBatch.manifestObjectKey,
contentLength: pendingBatch.manifestContentLength,
archiveSha256: pendingBatch.manifestArchiveSha256,
});
const cleanup = cleanupHistoryCandidates({dataDir, candidates: pendingBatch.candidates});
const manifest = {
batchId: pendingBatch.batchId,
uploadedAt: pendingBatch.uploadedAt,
candidates: pendingBatch.candidates,
};
const uploadResult = {
objectKey: pendingBatch.objectKey,
contentLength: pendingBatch.contentLength,
archiveSha256: pendingBatch.archiveSha256,
verifiedAt: pendingBatch.verifiedAt,
};
const manifestUpload = {
objectKey: pendingBatch.manifestObjectKey,
contentLength: pendingBatch.manifestContentLength,
archiveSha256: pendingBatch.manifestArchiveSha256,
verifiedAt: pendingBatch.manifestVerifiedAt,
};
const nextState = recordHistoryBatch({
statePath,
state,
manifest,
uploadResult,
manifestUpload,
status: 'cleaned',
cleanedAt: new Date().toISOString(),
});
console.log(`[database-backup] 已完成 history 清理重试: ${JSON.stringify(cleanup)}`);
return nextState;
}
async function runHistoryBackup({
args,
env,
dataDir,
workDir,
database,
bucket,
endpoint,
accessKeyId,
accessKeySecret,
objectPrefix,
keepLocal,
bandwidthLimiter,
}) {
const statePath = historyStatePath({args, env, workDir, database});
let state = loadOrImportHistoryState({args, env, statePath, database, dataDir});
if (!args.dryRun && !args.deferUpload) {
console.log(`[database-backup] 重新验真 full baseline: oss://${state.baseline.bucket}/${state.baseline.objectKey}`);
await verifyOssObject({
bucket: state.baseline.bucket,
endpoint,
objectKey: state.baseline.objectKey,
accessKeyId,
accessKeySecret,
contentLength: state.baseline.contentLength,
archiveSha256: state.baseline.archiveSha256,
});
await verifyOssObject({
bucket: state.baseline.bucket,
endpoint,
objectKey: state.baseline.manifestObjectKey,
accessKeyId,
accessKeySecret,
contentLength: state.baseline.manifestContentLength,
archiveSha256: state.baseline.manifestArchiveSha256,
});
state = await resumeUploadedHistoryBatch({
statePath,
state,
dataDir,
verificationOptions: {bucket, endpoint, accessKeyId, accessKeySecret},
});
}
const plan = discoverHistoryPlan({dataDir});
console.log(`[database-backup] history replicas: ${JSON.stringify(plan.replicas)}`);
console.log(`[database-backup] history 候选: count=${plan.candidates.length}, size=${formatBytes(plan.totalSizeBytes)}`);
if (args.resultFile) {
writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({statePath, baseline: state.baseline, ...plan}, null, 2)}\n`, 'utf8');
}
if (args.dryRun) {
console.log('[database-backup] history dry-run,仅输出安全候选,不打包、上传或删除。');
return;
}
if (plan.candidates.length === 0) {
console.log('[database-backup] 没有可归档的 history 候选。');
return;
}
assertSufficientHistoryWorkDirSpace({historySizeBytes: plan.totalSizeBytes, workDir, args, env});
const batchId = historyBatchId({baselineId: state.baseline.id, plan});
const {fileName, objectKey} = buildHistoryNames({
database,
objectPrefix,
baselineId: state.baseline.id,
batchId,
});
const archivePath = resolve(workDir, fileName);
const manifestPath = `${archivePath}.manifest.json`;
const manifest = {
schemaVersion: HISTORY_MANIFEST_SCHEMA_VERSION,
backupKind: 'spacetimedb-history',
createdAt: new Date().toISOString(),
database,
dataDir,
bucket,
objectKey,
archivePath,
baselineId: state.baseline.id,
baselineStatePath: statePath,
batchId,
replicas: plan.replicas,
candidates: plan.candidates,
totalSizeBytes: plan.totalSizeBytes,
uploadStatus: args.deferUpload ? 'deferred' : 'pending',
};
writeManifest({manifestPath, payload: manifest});
createHistoryArchive({
dataDir,
workDir,
fileName,
manifestPath,
candidates: plan.candidates,
});
if (args.deferUpload) {
console.log(`[database-backup] 已生成 history 归档,延后上传且未清理源文件: ${archivePath}`);
if (args.resultFile) {
writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, statePath, bucket, objectKey, batchId}, null, 2)}\n`, 'utf8');
}
return;
}
const historyResult = await uploadHistoryArchiveWithCleanup({
archivePath,
manifestPath,
manifest,
statePath,
uploadOptions: {bucket, endpoint, objectKey, accessKeyId, accessKeySecret, bandwidthLimiter},
});
console.log(`[database-backup] history 上传并清理完成: ${JSON.stringify(historyResult.cleanup)}`);
if (args.resultFile) {
writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({
archivePath,
manifestPath,
statePath,
batchId,
...historyResult.result,
uploadedAt: historyResult.uploadedManifest.uploadedAt,
}, null, 2)}\n`, 'utf8');
}
if (!keepLocal) {
rmSync(archivePath, {force: true});
rmSync(manifestPath, {force: true});
console.log('[database-backup] 已删除本地 history 临时归档和清单。');
}
}
async function main() {
const args = parseArgs(process.argv.slice(2));
const env = loadEffectiveEnv(args.envFiles);
const isProductionLike = existsSync(DEFAULT_PRODUCTION_DATA_DIR) && process.platform !== 'win32';
const dataDir = resolvePath(firstNonEmpty(
args.dataDir,
env.GENARRATIVE_DATABASE_BACKUP_DATA_DIR,
isProductionLike ? DEFAULT_PRODUCTION_DATA_DIR : DEFAULT_LOCAL_DATA_DIR,
));
const workDir = resolvePath(firstNonEmpty(
args.workDir,
args.uploadDeferredDir,
env.GENARRATIVE_DATABASE_BACKUP_WORK_DIR,
isProductionLike ? DEFAULT_PRODUCTION_WORK_DIR : DEFAULT_LOCAL_WORK_DIR,
));
const bucket = firstNonEmpty(args.bucket, env.GENARRATIVE_DATABASE_BACKUP_OSS_BUCKET, env.ALIYUN_OSS_BUCKET);
const endpoint = normalizeEndpoint(firstNonEmpty(args.endpoint, env.GENARRATIVE_DATABASE_BACKUP_OSS_ENDPOINT, env.ALIYUN_OSS_ENDPOINT));
const accessKeyId = firstNonEmpty(args.accessKeyId, env.GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_ID, env.ALIYUN_OSS_ACCESS_KEY_ID);
const accessKeySecret = firstNonEmpty(args.accessKeySecret, env.GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_SECRET, env.ALIYUN_OSS_ACCESS_KEY_SECRET);
const objectPrefix = firstNonEmpty(args.objectPrefix, env.GENARRATIVE_DATABASE_BACKUP_OSS_PREFIX, 'database-backups');
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);
const uploadBandwidthLimiter = createUploadBandwidthLimiter(env.GENARRATIVE_DATABASE_BACKUP_UPLOAD_MAX_BYTES_PER_SECOND);
if (!['full', 'history'].includes(args.mode)) {
throw new Error(`--mode 只能是 full 或 history,实际: ${args.mode}`);
}
if (!['archive', 'files'].includes(storageFormat)) {
throw new Error(`--storage-format 只能是 archive 或 files,实际: ${storageFormat}`);
}
for (const [label, value] of Object.entries({bucket, endpoint, accessKeyId, accessKeySecret})) {
if (!value) {
throw new Error(`缺少 ${label} 配置`);
}
}
if (args.restoreFilesState && args.restoreFilesLatest) {
throw new Error('--restore-files-state 与 --restore-files-latest 不能同时使用。');
}
if (args.restoreFilesState) {
if (!args.restoreDir) {
throw new Error('--restore-files-state 必须同时传 --restore-dir。');
}
await restoreDirectFilesBackup({
statePath: args.restoreFilesState,
restoreDir: args.restoreDir,
database,
bucket,
uploadOptions: {bucket, endpoint, accessKeyId, accessKeySecret},
resultFile: args.resultFile,
dryRun: args.dryRun,
});
return;
}
if (args.restoreFilesLatest) {
if (!args.restoreDir) {
throw new Error('--restore-files-latest 必须同时传 --restore-dir。');
}
await restoreDirectFilesLatest({
restoreDir: args.restoreDir,
database,
bucket,
objectPrefix,
uploadOptions: {bucket, endpoint, accessKeyId, accessKeySecret},
resultFile: args.resultFile,
dryRun: args.dryRun,
});
return;
}
if (args.restoreDir) {
throw new Error('--restore-dir 只能与 --restore-files-state 或 --restore-files-latest 一起使用。');
}
if (args.uploadArchive && args.uploadDeferredDir) {
throw new Error('--upload-archive 与 --upload-deferred-dir 不能同时使用。');
}
if (!args.dryRun) {
const lockPath = acquireBackupLock({workDir, database});
console.log(`[database-backup] 已获取进程锁: ${lockPath}`);
}
if (args.publishManifest) {
await publishExistingManifest({args, bucket, endpoint, accessKeyId, accessKeySecret, bandwidthLimiter: uploadBandwidthLimiter});
return;
}
if (args.uploadArchive) {
await uploadExistingArchive({
args,
env,
bucket,
endpoint,
accessKeyId,
accessKeySecret,
objectPrefix,
bandwidthLimiter: uploadBandwidthLimiter,
});
return;
}
if (args.uploadDeferredDir) {
await uploadDeferredArchives({
args,
env,
bucket,
endpoint,
accessKeyId,
accessKeySecret,
objectPrefix,
database,
bandwidthLimiter: uploadBandwidthLimiter,
});
return;
}
if (storageFormat === 'files') {
if (args.deferUpload) {
throw new Error('files 模式无需本地归档且不支持 --defer-upload;失败后使用同一 work-dir 重跑即可续传。');
}
const stopService = args.stopService || firstNonEmpty(env.GENARRATIVE_DATABASE_BACKUP_STOP_SERVICE);
const restartServicesAfter = collectRestartServicesAfterBackup({args, env});
const stopMarkerPath = databaseBackupStopMarkerPath(workDir);
let serviceStopped = false;
let backupError = null;
let restoreError = null;
try {
if (args.mode === 'full' && !args.dryRun) {
serviceStopped = stopServiceIfNeeded(stopService, stopMarkerPath);
}
await runDirectFilesBackup({
mode: args.mode,
dataDir,
workDir,
database,
bucket,
objectPrefix,
dryRun: args.dryRun,
resultFile: args.resultFile,
uploadOptions: {bucket, endpoint, accessKeyId, accessKeySecret, bandwidthLimiter: uploadBandwidthLimiter},
concurrency: directFilesConcurrency,
});
} catch (error) {
backupError = error;
} finally {
try {
if (serviceStopped) {
restoreServicesAfterBackup({stopService, serviceStopped, restartServicesAfter, stopMarkerPath});
} else if (!backupError && args.mode === 'full' && !args.dryRun) {
restartServicesAfterBackup(restartServicesAfter);
}
} catch (error) {
restoreError = error;
}
}
if (backupError && restoreError) {
throw new AggregateError([backupError, restoreError], `files 备份失败,且恢复依赖服务时也失败: ${backupError.message}; ${restoreError.message}`);
}
if (backupError) {
throw backupError;
}
if (restoreError) {
throw restoreError;
}
return;
}
if (args.mode === 'history') {
await runHistoryBackup({
args,
env,
dataDir,
workDir,
database,
bucket,
endpoint,
accessKeyId,
accessKeySecret,
objectPrefix,
keepLocal,
bandwidthLimiter: uploadBandwidthLimiter,
});
return;
}
const {fileName, objectKey} = buildBackupNames({database, dataDir, objectPrefix});
console.log(`[database-backup] 数据目录: ${dataDir}`);
console.log(`[database-backup] 本地临时目录: ${workDir}`);
console.log(`[database-backup] 目标对象: oss://${bucket}/${objectKey}`);
if (args.dryRun) {
console.log('[database-backup] dry-run,仅校验配置,不打包上传。');
return;
}
let archivePath = '';
let serviceStopped = false;
let backupError = null;
let restoreError = null;
const stopService = args.stopService || firstNonEmpty(env.GENARRATIVE_DATABASE_BACKUP_STOP_SERVICE);
const restartServicesAfter = collectRestartServicesAfterBackup({args, env});
const stopMarkerPath = databaseBackupStopMarkerPath(workDir);
try {
assertSufficientWorkDirSpace({dataDir, workDir, args, env});
serviceStopped = stopServiceIfNeeded(stopService, stopMarkerPath);
archivePath = createArchive({dataDir, workDir, fileName});
} catch (error) {
backupError = error;
} finally {
try {
if (serviceStopped) {
restoreServicesAfterBackup({stopService, serviceStopped, restartServicesAfter, stopMarkerPath});
} else if (!backupError) {
restartServicesAfterBackup(restartServicesAfter);
}
} catch (error) {
restoreError = error;
}
}
if (backupError) {
if (restoreError) {
throw new AggregateError([backupError, restoreError], `数据库备份失败,且恢复依赖服务时也失败: ${backupError.message}; ${restoreError.message}`);
}
throw backupError;
}
if (restoreError) {
throw restoreError;
}
const manifestPath = `${archivePath}.manifest.json`;
const baselineStatePath = historyStatePath({args, env, workDir, database});
const fullManifest = {
backupKind: 'spacetimedb-data-dir',
createdAt: new Date().toISOString(),
database,
dataDir,
bucket,
objectKey,
archivePath,
baselineStatePath,
uploadStatus: args.deferUpload ? 'deferred' : 'pending',
};
writeManifest({
manifestPath,
payload: fullManifest,
});
if (args.deferUpload) {
console.log(`[database-backup] 已生成本地冷备份,延后上传: ${archivePath}`);
console.log(`[database-backup] 已写入备份清单: ${manifestPath}`);
if (args.resultFile) {
writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, baselineStatePath, bucket, objectKey}, null, 2)}\n`, 'utf8');
}
return;
}
const result = await uploadArchive({
archivePath,
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
bandwidthLimiter: uploadBandwidthLimiter,
});
console.log(`[database-backup] 上传完成: ${JSON.stringify(result)}`);
const uploadedManifest = uploadedManifestPayload({manifest: fullManifest, database, result});
writeManifest({manifestPath, payload: uploadedManifest});
const manifestUpload = await uploadManifestFile({
manifestPath,
bucket,
endpoint,
objectKey: uploadedManifest.manifestObjectKey,
accessKeyId,
accessKeySecret,
bandwidthLimiter: uploadBandwidthLimiter,
});
uploadedManifest.manifestVerifiedAt = manifestUpload.verifiedAt;
uploadedManifest.manifestContentLength = manifestUpload.contentLength;
uploadedManifest.manifestArchiveSha256 = manifestUpload.archiveSha256;
writeManifest({manifestPath, payload: uploadedManifest});
const previousState = existsSync(baselineStatePath)
? validateHistoryState(readManifest(baselineStatePath), {database, dataDir})
: null;
const baseline = normalizeUploadedBaselineManifest(uploadedManifest, {database, dataDir});
writeBaselineState({statePath: baselineStatePath, baseline, previousState});
console.log(`[database-backup] 已写入 baseline state: ${baselineStatePath}`);
if (!keepLocal) {
rmSync(archivePath, {force: true});
rmSync(manifestPath, {force: true});
console.log('[database-backup] 已删除本地临时备份文件;如需保留请设置 --keep-local。');
} else {
console.log(`[database-backup] 已保留本地备份: ${archivePath}`);
console.log(`[database-backup] 已保留备份清单: ${manifestPath}`);
}
}
function formatErrorDetails(error) {
if (!error || typeof error !== 'object') {
return '';
}
return ['code', 'errno', 'syscall', 'hostname', 'host', 'port', 'address']
.map((field) => {
const value = error[field];
return value === undefined || value === null || value === '' ? '' : `${field}=${String(value)}`;
})
.filter(Boolean)
.join(' ');
}
function describeError(error) {
const lines = [];
let current = error;
for (let depth = 0; current && depth < 5; depth += 1) {
const label = depth === 0 ? 'error' : `cause[${depth}]`;
if (!(current instanceof Error)) {
lines.push(`${label}: ${String(current)}`);
break;
}
lines.push(`${label}: ${current.name}: ${current.message}`);
const details = formatErrorDetails(current);
if (details) {
lines.push(`${label} details: ${details}`);
}
if (current instanceof AggregateError) {
current.errors.slice(0, 3).forEach((item, index) => {
const itemText = item instanceof Error ? `${item.name}: ${item.message}` : String(item);
const itemDetails = formatErrorDetails(item);
lines.push(`${label}.errors[${index}]: ${itemText}${itemDetails ? ` (${itemDetails})` : ''}`);
});
}
current = current.cause;
}
return lines;
}
if (process.argv[1] && realpathSync(resolve(process.argv[1])) === realpathSync(__filename)) {
main().catch((error) => {
for (const line of describeError(error)) {
console.error(`[database-backup] ${line}`);
}
process.exit(1);
});
}