diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index b2f4725e4..de8d780f7 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -3997,3 +3997,11 @@ - 参数刷新：Jenkinsfile 是参数事实源。推送后必须让 Full 与 API Deploy live Job 安全加载一次新 Jenkinsfile，再只读确认两个参数已进入 `config.xml`；只在 Jenkins UI 手工加参数不是持久修复。 - 影响范围：Full / API Deploy Jenkinsfile、API 发布脚本、生产 API deploy fixture、生产运维门禁与 live Job 参数 schema。 - 验证方式：`bash -n scripts/deploy/production-api-deploy.sh`、`npm run check:production-api-deploy`、`npm run check:production-ops`、`npm run check:encoding`、`git diff --check`。 + +## 2026-07-13 数据库冷备使用 OSS Multipart 上传并在清理前验真 + +- 背景：SpacetimeDB 冷备归档已经超过 OSS 单次 PutObject 的 5 GiB 上限，单请求上传会稳定失败并让本地归档持续积压；网络中断还可能让 CompleteMultipartUpload 的客户端结果不确定。 +- 决策：`scripts/database-backup-to-oss.mjs` 对备份归档统一使用 OSS Multipart Upload，默认按 128 MiB 顺序分片；每个分片请求重新创建文件流、时间和 V4 签名，仅对网络错误、HTTP 408 / 429 / 5xx 做有限重试。V4 canonical query 必须同时支持无等号的 `uploads` 子资源和带值的 `partNumber` / `uploadId` 参数。 +- 验真与清理边界：Complete 后必须发送签名 HEAD，并严格核对 OSS `Content-Length` 与本地归档大小；Complete 响应不确定时也先用 HEAD 判定对象是否已经完整落盘。只有验真成功后才能把 manifest 标记为 `uploaded`，并按 `keepLocal` 决定是否删除本地归档；失败时 best-effort AbortMultipartUpload，不得提前更新 manifest 或清理本地文件。 +- 影响范围：数据库备份 OSS 上传实现、备份回归门禁、release 本地归档保留与 timer 恢复流程。 +- 验证方式：`npm run check:database-backup`、`npm run check:production-ops`、`npm run check:encoding`、`git diff --check`；线上先对既有归档使用 `--upload-archive ... --keep-local`，确认 OSS 对象长度和可恢复性后再清理积压并恢复 timer。 diff --git a/scripts/check-database-backup-to-oss.mjs b/scripts/check-database-backup-to-oss.mjs index af154f6cf..bbc8b0fb2 100644 --- a/scripts/check-database-backup-to-oss.mjs +++ b/scripts/check-database-backup-to-oss.mjs @@ -5,12 +5,14 @@ import {existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync} import {tmpdir} from 'node:os'; import path from 'node:path'; +import {buildAuthorization, buildCanonicalQuery, uploadArchive} from './database-backup-to-oss.mjs'; + const BACKUP_SCRIPT = path.resolve('scripts/database-backup-to-oss.mjs'); const tmpRoot = mkdtempSync(path.join(tmpdir(), 'genarrative-database-backup-check-')); const failures = []; try { - main(); + await main(); } finally { rmSync(tmpRoot, {recursive: true, force: true}); } @@ -25,9 +27,58 @@ if (failures.length > 0) { console.log('[check:database-backup] OK'); -function main() { +async function main() { + assertCanonicalQueryAndAuthorizationIncludeMultipartParameters(); assertInsufficientSpaceStopsBeforeServiceChanges(); assertArchiveFailureStillRestoresDependentServices(); + await assertMultipartUploadRetriesAndVerifiesRemoteLength(); + await assertMissingPartEtagAbortsMultipartUpload(); + await assertCompleteResponseAmbiguityUsesHeadVerification(); + await assertHeadLengthMismatchAbortsMultipartUpload(); +} + +function assertCanonicalQueryAndAuthorizationIncludeMultipartParameters() { + assertEqual(buildCanonicalQuery({uploads: null}), 'uploads', 'InitiateMultipartUpload 必须使用无等号的 uploads 参数。'); + assertEqual( + buildCanonicalQuery({uploadId: 'abc+/= xyz', partNumber: 12}), + 'partNumber=12&uploadId=abc%2B%2F%3D%20xyz', + 'multipart query 必须按 key 排序并使用 RFC3986 编码。', + ); + + const date = new Date('2026-07-13T10:20:30.000Z'); + const headers = { + host: 'genarrative-test.oss-cn-shanghai.aliyuncs.com', + 'content-type': 'application/octet-stream', + 'x-oss-content-sha256': 'UNSIGNED-PAYLOAD', + 'x-oss-date': '20260713T102030Z', + }; + const withoutQuery = buildAuthorization({ + method: 'PUT', + bucket: 'genarrative-test', + endpoint: 'oss-cn-shanghai.aliyuncs.com', + objectKey: 'database-backups/test/archive.tar.gz', + accessKeyId: 'test-access-key', + accessKeySecret: 'test-access-secret', + headers, + date, + }); + const withQuery = buildAuthorization({ + method: 'PUT', + bucket: 'genarrative-test', + endpoint: 'oss-cn-shanghai.aliyuncs.com', + objectKey: 'database-backups/test/archive.tar.gz', + accessKeyId: 'test-access-key', + accessKeySecret: 'test-access-secret', + headers, + date, + queries: {partNumber: 12, uploadId: 'abc+/= xyz'}, + }); + assertNotEqual(withQuery, withoutQuery, 'multipart query 必须参与 V4 Authorization 计算。'); + assertEqual( + withQuery, + 'OSS4-HMAC-SHA256 Credential=test-access-key/20260713/cn-shanghai/oss/aliyun_v4_request,AdditionalHeaders=host,Signature=9323dd3b7272b52f416c4d32115fcc00460eaccdcdaf011575c2502a63a27b1f', + 'multipart V4 Authorization 必须保持固定签名向量。', + ); } function assertInsufficientSpaceStopsBeforeServiceChanges() { @@ -78,6 +129,279 @@ function assertArchiveFailureStillRestoresDependentServices() { } } +async function assertMultipartUploadRetriesAndVerifiesRemoteLength() { + const root = path.join(tmpRoot, 'multipart-success'); + const archivePath = path.join(root, 'backup.tar.gz'); + const partSizeBytes = 100 * 1024; + const payload = Buffer.concat([ + Buffer.alloc(partSizeBytes, 'a'), + Buffer.alloc(partSizeBytes, 'b'), + Buffer.alloc(17, 'c'), + ]); + mkdirSync(root, {recursive: true}); + writeFileSync(archivePath, payload); + + const requests = []; + const retryDelays = []; + let firstPartAttempts = 0; + const uploadId = 'upload+/= id'; + const fetchImpl = async (url, options) => { + const body = await readRequestBody(options.body); + requests.push({url, method: options.method, headers: options.headers, body}); + const parsedUrl = new URL(url); + + if (options.method === 'POST' && parsedUrl.search === '?uploads') { + return new Response(`${uploadId}`, {status: 200}); + } + if (options.method === 'PUT') { + const partNumber = Number(parsedUrl.searchParams.get('partNumber')); + if (partNumber === 1) { + firstPartAttempts += 1; + if (firstPartAttempts === 1) { + return new Response('ServiceUnavailable', {status: 503}); + } + } + return new Response('', {status: 200, headers: {etag: `"etag-${partNumber}"`}}); + } + if (options.method === 'POST' && parsedUrl.searchParams.has('uploadId')) { + return new Response('', {status: 200, headers: {etag: '"complete-etag"'}}); + } + if (options.method === 'HEAD') { + return new Response(null, {status: 200, headers: {'content-length': String(payload.length)}}); + } + throw new Error(`unexpected request: ${options.method} ${url}`); + }; + + const result = await uploadArchive({ + archivePath, + bucket: 'genarrative-test', + endpoint: 'oss-cn-shanghai.aliyuncs.com', + objectKey: 'database-backups/test/backup.tar.gz', + accessKeyId: 'test-access-key', + accessKeySecret: 'test-access-secret', + partSizeBytes, + maxAttempts: 3, + retryBaseDelayMs: 1, + retryMaxDelayMs: 1, + fetchImpl, + nowFn: () => new Date('2026-07-13T10:20:30.000Z'), + sleepImpl: async (delayMs) => retryDelays.push(delayMs), + randomFn: () => 0, + }); + + assertEqual(result.uploadMode, 'multipart', '上传结果必须记录 multipart 模式。'); + assertEqual(result.partCount, 3, 'multipart 应按配置大小切成三段。'); + assertEqual(result.contentLength, payload.length, '上传结果应保留完整归档长度。'); + assertEqual(result.etag, 'complete-etag', '上传结果应保留 CompleteMultipartUpload ETag。'); + assertEqual(firstPartAttempts, 2, '503 后应仅重试失败的第一段。'); + assertEqual(retryDelays.length, 1, '一次可重试失败应触发一次退避。'); + + const initiateRequest = requests[0]; + assertTrue(initiateRequest.url.endsWith('?uploads'), 'InitiateMultipartUpload URL 必须使用裸 uploads 参数。'); + assertTrue(!initiateRequest.url.endsWith('?uploads='), 'InitiateMultipartUpload URL 不能把裸参数写成 uploads=。'); + const firstPartRequests = requests.filter(({method, url}) => method === 'PUT' && new URL(url).searchParams.get('partNumber') === '1'); + assertEqual(firstPartRequests.length, 2, '第一段应产生原请求和一次重试。'); + assertBufferEqual(firstPartRequests[0].body, payload.subarray(0, partSizeBytes), '第一段原请求内容必须完整。'); + assertBufferEqual(firstPartRequests[1].body, payload.subarray(0, partSizeBytes), '第一段重试必须重新创建并完整读取 stream。'); + assertTrue( + firstPartRequests[0].url.includes('?partNumber=1&uploadId=upload%2B%2F%3D%20id'), + 'UploadPart URL 必须使用排序并编码后的 canonical query。', + ); + + const completeRequest = requests.find(({method, url}) => method === 'POST' && new URL(url).searchParams.has('uploadId')); + assertIncludes(completeRequest?.body.toString('utf8') ?? '', '1"etag-1"', 'Complete XML 应包含第一段 ETag。'); + assertIncludes(completeRequest?.body.toString('utf8') ?? '', '3"etag-3"', 'Complete XML 应包含最后一段 ETag。'); + assertTrue(requests.some(({method}) => method === 'HEAD'), 'Complete 后必须执行签名 HEAD 验证。'); + for (const request of requests) { + assertTrue(String(request.headers.authorization ?? '').startsWith('OSS4-HMAC-SHA256 '), `${request.method} 请求必须携带 V4 Authorization。`); + } +} + +async function assertHeadLengthMismatchAbortsMultipartUpload() { + const root = path.join(tmpRoot, 'multipart-head-mismatch'); + const archivePath = path.join(root, 'backup.tar.gz'); + const partSizeBytes = 100 * 1024; + const payload = Buffer.alloc(partSizeBytes + 1, 'x'); + mkdirSync(root, {recursive: true}); + writeFileSync(archivePath, payload); + + const requests = []; + const fetchImpl = async (url, options) => { + await readRequestBody(options.body); + requests.push({url, method: options.method}); + const parsedUrl = new URL(url); + if (options.method === 'POST' && parsedUrl.search === '?uploads') { + return new Response('mismatch-upload', {status: 200}); + } + if (options.method === 'PUT') { + return new Response('', {status: 200, headers: {etag: `"etag-${parsedUrl.searchParams.get('partNumber')}"`}}); + } + if (options.method === 'POST') { + return new Response('', {status: 200, headers: {etag: '"complete-etag"'}}); + } + if (options.method === 'HEAD') { + return new Response(null, {status: 200, headers: {'content-length': String(payload.length - 1)}}); + } + if (options.method === 'DELETE') { + return new Response(null, {status: 204}); + } + throw new Error(`unexpected request: ${options.method} ${url}`); + }; + + let uploadError = null; + try { + await uploadArchive({ + archivePath, + bucket: 'genarrative-test', + endpoint: 'oss-cn-shanghai.aliyuncs.com', + objectKey: 'database-backups/test/mismatch.tar.gz', + accessKeyId: 'test-access-key', + accessKeySecret: 'test-access-secret', + partSizeBytes, + maxAttempts: 2, + retryBaseDelayMs: 1, + retryMaxDelayMs: 1, + fetchImpl, + nowFn: () => new Date('2026-07-13T10:20:30.000Z'), + sleepImpl: async () => {}, + randomFn: () => 0, + }); + } catch (error) { + uploadError = error; + } + + assertTrue(uploadError instanceof Error, 'HEAD 长度不一致时上传必须失败。'); + assertIncludes(uploadError?.message ?? '', 'HEAD 验证长度不一致', 'HEAD 长度不一致错误应保留本地和远端长度。'); + const abortRequest = requests.find(({method}) => method === 'DELETE'); + assertTrue(Boolean(abortRequest), 'HEAD 长度不一致后必须 best-effort AbortMultipartUpload。'); + assertTrue(abortRequest?.url.endsWith('?uploadId=mismatch-upload'), 'AbortMultipartUpload 必须携带同一 uploadId。'); +} + +async function assertMissingPartEtagAbortsMultipartUpload() { + const root = path.join(tmpRoot, 'multipart-missing-etag'); + const archivePath = path.join(root, 'backup.tar.gz'); + mkdirSync(root, {recursive: true}); + writeFileSync(archivePath, Buffer.alloc(100 * 1024, 'e')); + + const requests = []; + const fetchImpl = async (url, options) => { + await readRequestBody(options.body); + requests.push({url, method: options.method}); + const parsedUrl = new URL(url); + if (options.method === 'POST' && parsedUrl.search === '?uploads') { + return new Response('missing-etag-upload', {status: 200}); + } + if (options.method === 'PUT') { + return new Response('', {status: 200}); + } + if (options.method === 'DELETE') { + return new Response(null, {status: 204}); + } + throw new Error(`unexpected request: ${options.method} ${url}`); + }; + + let uploadError = null; + try { + await uploadArchive({ + archivePath, + bucket: 'genarrative-test', + endpoint: 'oss-cn-shanghai.aliyuncs.com', + objectKey: 'database-backups/test/missing-etag.tar.gz', + accessKeyId: 'test-access-key', + accessKeySecret: 'test-access-secret', + partSizeBytes: 100 * 1024, + maxAttempts: 1, + fetchImpl, + nowFn: () => new Date('2026-07-13T10:20:30.000Z'), + sleepImpl: async () => {}, + randomFn: () => 0, + }); + } catch (error) { + uploadError = error; + } + + assertIncludes(uploadError?.message ?? '', '响应缺少 ETag', 'UploadPart 缺少 ETag 时必须失败。'); + assertTrue( + requests.some(({method, url}) => method === 'DELETE' && url.endsWith('?uploadId=missing-etag-upload')), + 'UploadPart 缺少 ETag 后必须 AbortMultipartUpload。', + ); +} + +async function assertCompleteResponseAmbiguityUsesHeadVerification() { + const root = path.join(tmpRoot, 'multipart-complete-ambiguity'); + const archivePath = path.join(root, 'backup.tar.gz'); + const payload = Buffer.alloc(100 * 1024, 'c'); + mkdirSync(root, {recursive: true}); + writeFileSync(archivePath, payload); + + const requests = []; + let completeAttempts = 0; + const fetchImpl = async (url, options) => { + await readRequestBody(options.body); + requests.push({url, method: options.method}); + const parsedUrl = new URL(url); + if (options.method === 'POST' && parsedUrl.search === '?uploads') { + return new Response('ambiguous-upload', {status: 200}); + } + if (options.method === 'PUT') { + return new Response('', {status: 200, headers: {etag: '"part-etag"'}}); + } + if (options.method === 'POST' && parsedUrl.searchParams.has('uploadId')) { + completeAttempts += 1; + if (completeAttempts === 1) { + throw new TypeError('socket closed after remote complete'); + } + return new Response('NoSuchUpload', {status: 404}); + } + if (options.method === 'HEAD') { + return new Response(null, {status: 200, headers: {'content-length': String(payload.length)}}); + } + if (options.method === 'DELETE') { + return new Response(null, {status: 204}); + } + throw new Error(`unexpected request: ${options.method} ${url}`); + }; + + const result = await uploadArchive({ + archivePath, + bucket: 'genarrative-test', + endpoint: 'oss-cn-shanghai.aliyuncs.com', + objectKey: 'database-backups/test/complete-ambiguity.tar.gz', + accessKeyId: 'test-access-key', + accessKeySecret: 'test-access-secret', + partSizeBytes: 100 * 1024, + maxAttempts: 2, + retryBaseDelayMs: 1, + retryMaxDelayMs: 1, + fetchImpl, + nowFn: () => new Date('2026-07-13T10:20:30.000Z'), + sleepImpl: async () => {}, + randomFn: () => 0, + }); + + assertEqual(completeAttempts, 2, 'Complete 网络错误后应按策略重试。'); + assertEqual(result.contentLength, payload.length, 'Complete 结果不确定时应以 HEAD 长度验真收口。'); + assertTrue(requests.some(({method}) => method === 'HEAD'), 'Complete 结果不确定时必须执行 HEAD 验真。'); + assertTrue(!requests.some(({method}) => method === 'DELETE'), 'HEAD 已证实对象完整时不得 Abort 已完成上传。'); +} + +async function readRequestBody(body) { + if (body === undefined || body === null) { + return Buffer.alloc(0); + } + if (typeof body === 'string') { + return Buffer.from(body); + } + if (Buffer.isBuffer(body)) { + return body; + } + const chunks = []; + for await (const chunk of body) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + return Buffer.concat(chunks); +} + function createFixture(name) { const root = path.join(tmpRoot, name); const binDir = path.join(root, 'bin'); @@ -160,6 +484,30 @@ function assertIncludes(content, expected, reason) { } } +function assertEqual(actual, expected, reason) { + if (actual !== expected) { + failures.push(`${reason} 预期: ${String(expected)}，实际: ${String(actual)}`); + } +} + +function assertNotEqual(actual, expected, reason) { + if (actual === expected) { + failures.push(`${reason} 两者均为: ${String(actual)}`); + } +} + +function assertTrue(condition, reason) { + if (!condition) { + failures.push(reason); + } +} + +function assertBufferEqual(actual, expected, reason) { + if (!Buffer.isBuffer(actual) || !actual.equals(expected)) { + failures.push(`${reason} 预期 ${expected.length} bytes，实际 ${actual?.length ?? ''} bytes。`); + } +} + function assertFileMissing(filePath, reason) { if (existsSync(filePath)) { failures.push(`${reason} 实际存在: ${filePath}\n${readFile(filePath)}`); diff --git a/scripts/database-backup-to-oss.mjs b/scripts/database-backup-to-oss.mjs index 4d4e969a2..d38a28598 100644 --- a/scripts/database-backup-to-oss.mjs +++ b/scripts/database-backup-to-oss.mjs @@ -1,8 +1,9 @@ #!/usr/bin/env node import {spawnSync} from 'node:child_process'; import {createHash, createHmac} from 'node:crypto'; -import {createReadStream, existsSync, mkdirSync, readFileSync, rmSync, statSync, statfsSync, writeFileSync} from 'node:fs'; +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); @@ -18,6 +19,14 @@ 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(`用法: @@ -484,11 +493,33 @@ function encodePath(path) { .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, ' '); } -function buildAuthorization({method, bucket, endpoint, objectKey, accessKeyId, accessKeySecret, headers, date}) { +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}`; @@ -504,7 +535,7 @@ function buildAuthorization({method, bucket, endpoint, objectKey, accessKeyId, a const canonicalRequest = [ method, canonicalUri, - '', + buildCanonicalQuery(queries), canonicalHeaders, additionalHeaders, UNSIGNED_PAYLOAD, @@ -518,56 +549,304 @@ function buildAuthorization({method, bucket, endpoint, objectKey, accessKeyId, a return `${OSS_ALGORITHM} Credential=${accessKeyId}/${scope},AdditionalHeaders=${additionalHeaders},Signature=${finalSignature}`; } -async function uploadArchive({archivePath, bucket, endpoint, objectKey, accessKeyId, accessKeySecret}) { +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]*?)`, 'u').exec(xml); + if (!match) { + return ''; + } + return match[1] + .replace(/</gu, '<') + .replace(/>/gu, '>') + .replace(/"/gu, '"') + .replace(/'/gu, "'") + .replace(/&/gu, '&') + .trim(); +} + +function escapeXml(value) { + return String(value) + .replace(/&/gu, '&') + .replace(//gu, '>') + .replace(/"/gu, '"') + .replace(/'/gu, '''); +} + +function buildCompleteMultipartBody(parts) { + const partXml = parts + .map(({partNumber, etag}) => [ + '', + `${partNumber}`, + `${escapeXml(etag)}`, + '', + ].join('')) + .join(''); + return `${partXml}`; +} + +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 ?? ''}`); + } + 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); - const now = new Date(); - const targetUrl = `https://${bucket}.${endpoint}/${encodePath(objectKey)}`; - const headers = { - host: `${bucket}.${endpoint}`, - 'content-type': 'application/gzip', - 'x-oss-content-sha256': UNSIGNED_PAYLOAD, - 'x-oss-date': formatOssDate(now), - 'x-oss-meta-backup-kind': 'spacetimedb-data-dir', - }; - const authorization = buildAuthorization({ - method: 'PUT', + 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, - headers, - date: now, - }); - - console.log(`[database-backup] 上传 OSS: oss://${bucket}/${objectKey}`); - let response; - try { - response = await fetch(targetUrl, { - method: 'PUT', - headers: { - ...headers, - authorization, - 'content-length': String(fileStat.size), - }, - body: createReadStream(archivePath), - duplex: 'half', - }); - } catch (error) { - throw new Error(`OSS 上传请求失败: oss://${bucket}/${objectKey}`, {cause: error}); - } - - const responseText = await response.text(); - if (!response.ok) { - throw new Error(`OSS 上传失败 HTTP ${response.status}: ${responseText.slice(0, 500)}`); - } - - return { - bucket, - objectKey, - contentLength: fileStat.size, - etag: response.headers.get('etag')?.replace(/^"|"$/gu, '') ?? '', + 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 (/)/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}) { @@ -603,6 +882,10 @@ async function uploadExistingArchive({args, env, bucket, endpoint, accessKeyId, objectKey: result.objectKey, contentLength: result.contentLength, etag: result.etag, + uploadMode: result.uploadMode, + partCount: result.partCount, + partSizeBytes: result.partSizeBytes, + verifiedAt: result.verifiedAt, uploadedAt, uploadStatus: 'uploaded', }, @@ -736,6 +1019,10 @@ async function main() { 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', }, @@ -791,9 +1078,11 @@ function describeError(error) { return lines; } -main().catch((error) => { - for (const line of describeError(error)) { - console.error(`[database-backup] ${line}`); - } - process.exit(1); -}); +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); + }); +}