Files
Genarrative/scripts/database-backup-to-oss.mjs
T
kdletters aeca412cdb 修复数据库冷备分片上传
冷备归档改用 OSS Multipart 顺序分片上传并加入可重试请求。

Complete 后通过签名 HEAD 严格核对远端对象长度,再更新清单或清理本地归档。

补齐签名查询、分片重试、缺失 ETag、Complete 歧义和 Abort 回归测试。

记录数据库冷备上传与清理边界的长期运维决策。
2026-07-13 11:13:57 +08:00

1089 lines
37 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 {createReadStream, existsSync, mkdirSync, readFileSync, realpathSync, rmSync, statSync, statfsSync, writeFileSync} from 'node:fs';
import {basename, dirname, isAbsolute, resolve} from 'node:path';
import {setTimeout as sleep} from 'node:timers/promises';
import {fileURLToPath} from 'node:url';
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_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 RETRYABLE_OSS_HTTP_STATUSES = new Set([408, 429, 500, 502, 503, 504]);
function usage() {
console.log(`用法:
npm run database:backup:oss -- [--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>
说明:
将 SpacetimeDB 数据目录打包成 .tar.gz,并上传到阿里云 OSS 指定 bucket。
--defer-upload 只生成本地冷备份和 manifest,不上传;后续用 --upload-archive 异步上传。
默认读取 .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_KEEP_LOCAL true 时保留本地 tar.gz
GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES 备份前要求 work dir 所在文件系统至少有这些可用字节;未设置时按数据目录大小估算
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: '',
manifestFile: '',
objectKey: '',
resultFile: '',
minFreeBytes: '',
};
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 '--manifest-file':
options.manifestFile = readValue();
break;
case '--result-file':
options.resultFile = readValue();
break;
case '--min-free-bytes':
options.minFreeBytes = readValue();
break;
default:
throw new Error(`未知参数: ${arg}`);
}
}
return options;
}
function firstNonEmpty(...values) {
return values.map((value) => String(value ?? '').trim()).find(Boolean) ?? '';
}
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 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 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 stopServiceIfNeeded(serviceName) {
if (!serviceName) {
return false;
}
console.log(`[database-backup] 停止服务以获取冷备份: ${serviceName}`);
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}) {
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('; ')}`);
}
}
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'});
return archivePath;
}
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');
}
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);
}
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}) {
const response = await signedOssRequest({
...requestOptions,
method: 'HEAD',
operation: 'HEAD 验证',
});
const contentLengthHeader = response.headers.get('content-length');
if (!contentLengthHeader || !/^\d+$/u.test(contentLengthHeader)) {
throw new Error(`OSS HEAD 验证缺少有效 content-length: ${contentLengthHeader ?? '<missing>'}`);
}
const remoteContentLength = Number(contentLengthHeader);
if (remoteContentLength !== expectedContentLength) {
throw new Error(`OSS HEAD 验证长度不一致: local=${expectedContentLength}, remote=${remoteContentLength}`);
}
return {verifiedAt: new Date().toISOString(), remoteContentLength};
}
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,
}) {
const fileStat = statSync(archivePath);
if (!fileStat.isFile() || fileStat.size <= 0) {
throw new Error(`待上传备份必须是非空文件: ${archivePath}`);
}
const partSize = resolveMultipartPartSize(fileStat.size, partSizeBytes);
const partCount = Math.ceil(fileStat.size / partSize);
const requestOptions = {
bucket,
endpoint,
objectKey,
accessKeyId,
accessKeySecret,
fetchImpl,
nowFn,
sleepImpl,
randomFn,
maxAttempts,
retryBaseDelayMs,
retryMaxDelayMs,
};
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': 'application/gzip',
'x-oss-meta-backup-kind': 'spacetimedb-data-dir',
},
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: () => createReadStream(archivePath, {start, end}),
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});
completeResponse = null;
} catch {
throw completeError;
}
}
const verification = await verifyUploadedObject({requestOptions, expectedContentLength: fileStat.size});
uploadCompleted = true;
return {
bucket,
objectKey,
contentLength: fileStat.size,
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;
}
}
async function uploadExistingArchive({args, env, bucket, endpoint, accessKeyId, accessKeySecret, objectPrefix}) {
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);
console.log(`[database-backup] 上传已有备份: ${archivePath}`);
console.log(`[database-backup] 目标对象: oss://${bucket}/${objectKey}`);
if (args.dryRun) {
console.log('[database-backup] dry-run,仅校验上传配置。');
return;
}
const result = await uploadArchive({archivePath, bucket, endpoint, objectKey, accessKeyId, accessKeySecret});
console.log(`[database-backup] 上传完成: ${JSON.stringify(result)}`);
const uploadedAt = new Date().toISOString();
writeManifest({
manifestPath,
payload: {
...manifest,
database,
bucket: result.bucket,
objectKey: result.objectKey,
contentLength: result.contentLength,
etag: result.etag,
uploadMode: result.uploadMode,
partCount: result.partCount,
partSizeBytes: result.partSizeBytes,
verifiedAt: result.verifiedAt,
uploadedAt,
uploadStatus: 'uploaded',
},
});
if (args.resultFile) {
writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, ...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 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,
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';
for (const [label, value] of Object.entries({bucket, endpoint, accessKeyId, accessKeySecret})) {
if (!value) {
throw new Error(`缺少 ${label} 配置`);
}
}
if (args.uploadArchive) {
await uploadExistingArchive({args, env, bucket, endpoint, accessKeyId, accessKeySecret, objectPrefix});
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});
try {
assertSufficientWorkDirSpace({dataDir, workDir, args, env});
serviceStopped = stopServiceIfNeeded(stopService);
archivePath = createArchive({dataDir, workDir, fileName});
} catch (error) {
backupError = error;
} finally {
try {
if (serviceStopped) {
restoreServicesAfterBackup({stopService, serviceStopped, restartServicesAfter});
} 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`;
writeManifest({
manifestPath,
payload: {
createdAt: new Date().toISOString(),
database,
dataDir,
bucket,
objectKey,
archivePath,
uploadStatus: args.deferUpload ? 'deferred' : 'pending',
},
});
if (args.deferUpload) {
console.log(`[database-backup] 已生成本地冷备份,延后上传: ${archivePath}`);
console.log(`[database-backup] 已写入备份清单: ${manifestPath}`);
if (args.resultFile) {
writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, bucket, objectKey}, null, 2)}\n`, 'utf8');
}
return;
}
const result = await uploadArchive({archivePath, bucket, endpoint, objectKey, accessKeyId, accessKeySecret});
console.log(`[database-backup] 上传完成: ${JSON.stringify(result)}`);
writeManifest({
manifestPath,
payload: {
createdAt: new Date().toISOString(),
database,
dataDir,
bucket: result.bucket,
objectKey: result.objectKey,
archivePath,
contentLength: result.contentLength,
etag: result.etag,
uploadMode: result.uploadMode,
partCount: result.partCount,
partSizeBytes: result.partSizeBytes,
verifiedAt: result.verifiedAt,
uploadedAt: new Date().toISOString(),
uploadStatus: 'uploaded',
},
});
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);
});
}