diff --git a/.github/workflows/gridfs-fix-validation.yml b/.github/workflows/gridfs-fix-validation.yml index 0063466..48b72b8 100644 --- a/.github/workflows/gridfs-fix-validation.yml +++ b/.github/workflows/gridfs-fix-validation.yml @@ -2,7 +2,11 @@ name: GridFS Ownership Safety Validation on: pull_request: - branches: [main] + branches: [main, 'fix/gridfs-id-ownership-v01'] + +concurrency: + group: gridfs-safety-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true permissions: contents: read @@ -10,6 +14,7 @@ permissions: jobs: gridfs: runs-on: ubuntu-latest + timeout-minutes: 30 strategy: fail-fast: false matrix: @@ -39,4 +44,5 @@ jobs: .generated/test-dist/test/unit/gridfs/gridfs.test.js \ .generated/test-dist/test/integration/gridfs/gridfs.test.js \ .generated/test-dist/test/integration/gridfs/gridfs-scan-regressions.test.js \ - .generated/test-dist/test/integration/gridfs/gridfs-ownership-regressions.test.js + .generated/test-dist/test/integration/gridfs/gridfs-ownership-regressions.test.js \ + .generated/test-dist/test/integration/gridfs/gridfs-stale-recovery.test.js diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 83a6fe0..b421ae3 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -4,11 +4,18 @@ on: push: branches: [main] pull_request: - branches: [main] + branches: [main, 'fix/gridfs-id-ownership-v01'] + +# Superseded PR revisions must not accumulate long-running test jobs. +# Cancelled prior SHAs are never accepted as evidence for the current HEAD. +concurrency: + group: ci-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true jobs: test: runs-on: ubuntu-latest + timeout-minutes: 60 strategy: fail-fast: false matrix: @@ -50,6 +57,7 @@ jobs: release-gate: runs-on: ubuntu-latest + timeout-minutes: 60 steps: - name: Checkout uses: actions/checkout@v5 diff --git a/docs/en/gridfs-recovery.md b/docs/en/gridfs-recovery.md new file mode 100644 index 0000000..07915c8 --- /dev/null +++ b/docs/en/gridfs-recovery.md @@ -0,0 +1,71 @@ +# GridFS crash recovery and no-data-loss upgrade runbook (V0.1 candidate) + +> **Operator-only.** This is a draft recovery design, not a completed or verified release. Never run recovery against a running, paused or network-partitioned GridFS writer. **Age alone is not proof that a writer is dead.** No automated TTL-based owner unlock is permitted. + +## Scope and invariant + +The ownership registry is the per-bucket `.uploads` collection. A reservation excludes ordinary upload, uploadStream, resumable append/finalize/abort, delete, rename and orphan cleanup for the same ObjectId. A missing file document **does not** prove its chunks are orphaned: an ordinary upload may still be writing, or GridFS driver delete may have removed the files document before chunks. Recovering while an old writer can resume can destroy new data, even with a token-CAS. The maintenance process cannot prove cluster/process shutdown on its own. + +The script `scripts/maintenance/gridfs-recover.cjs` uses a default read-only preview, exact state/token/timestamp checks, a persisted `recovering` ownership claim, verified preservation of committed visible chunks, and an append-only-by-intent audit collection `.ownership_recovery_audit`. It never performs a time-based lock steal. + +## Deployment and rollback gates (required) + +1. Stop all HTTP/worker/job/cron processes that can mutate this GridFS bucket, across **all** deployments, driver versions, replica instances, queue consumers and old binaries. Disable automatic restarts and new GridFS requests. Resolve in-flight operations, check for network-partitioned/paused workers, and ensure the old writer fleet is fully drained. A mixed old/new-writer deployment is unsupported because the old version ignores the new registry. +2. Back up `.files`, `.chunks`, `.uploads`, current indexes and relevant deployment version/worker inventory. Record a timestamp, operator, target database/bucket/id, intended delete/finalize outcome, and why no writer can resume. Retain backups and audit logs outside the cluster. +3. Deploy the new version while writes remain disabled. Run `ensureIndexes()` after verifying the old `createdAt_1` TTL and any legacy pending sessions. The migration derives `expiresAt` from the last update for eligible idle sessions, removes the legacy TTL index, and creates `expiresAt_1` with `expireAfterSeconds: 0`. TTL can already have removed sessions before migration; examine orphan chunks separately. Never treat an expiring session as safe while a legacy writer may still run. +4. Inspect every retained owner that must be recovered. Reconcile visible file and chunks, determine its original operation and the desired post-recovery state, and record the evidence. An owner without expiry is not automatically stale. Use the script only for an owner that has been conclusively quiesced. +5. Run dry-run and compare the proposed action with the recorded intent. To apply, supply the exact observed `status`, `updatedAt` ISO timestamp, and `ownerToken`, `activeAppend.token`, or `recovery.token` when present. Any mismatch is a stop signal. An audit document is written before a change; if a step fails the state remains `recovering` and must be re-inspected. +6. Recheck file visibility, stored chunk count, lengths, and any relevant SHA. Complete the latest-HEAD Node 18/20/22 CI, MongoDB 7.0.37/8.0.26 targeted suite, applicable Driver 6/7 compatibility checks and release preflight before unfreezing. Verify no `recovering` or unintended active reservations remain. +7. Only then restart new writers. Rollback to an old binary is **not safe while a new-version writer or reservation is active**; first stop/drain all new writers, reconcile operations, and assess the old binary's inability to honor `.uploads` ownership. Preserve the migrated index configuration; do not recreate `createdAt` TTL blindly. + +## Manual command + +Use a checkout with installed dependencies and a MongoDB driver. Never pass the connection string in CLI arguments or paste credentials into an issue. + +```sh +export MONSQLIZE_RECOVERY_MONGODB_URI='' +node scripts/maintenance/gridfs-recover.cjs \ + --db=my_database --bucket=fs \ + --id=0123456789abcdef01234567 \ + --expected-status=aborting \ + --expected-updated-at=2026-11-12T02:00:00.000Z +``` + +This previews but does not write. Retrieve the exact observed fields from the backed-up `.uploads` collection first. If the document contains a token, append `--expected-token=...`. **Do not copy** the example ObjectId or timestamp. Apply only after external process quiescence is verified and logged: + +```sh +node scripts/maintenance/gridfs-recover.cjs \ + --db=my_database --bucket=fs \ + --id=0123456789abcdef01234567 \ + --expected-status=aborting \ + --expected-updated-at=2026-11-12T02:00:00.000Z \ + --operator=oncall-id --reason=verified-writer-shutdown \ + --quiesced --apply +``` + +For an already `recovering` owner only, verify that the previous recovery executor has stopped and add `--prior-recovery-stopped` to the apply command; this is separate from stopping ordinary GridFS writers. + +Every apply attempt includes an audit entry. A crash can leave a `status: recovering` record with `recovery.token`; after proving **both all writers and the previous recovery executor have stopped**, re-inspect that record and use its new status, updatedAt and recovery token to retry. Supply the additional `--prior-recovery-stopped` flag. The restart acquires a fresh recovery token via atomic CAS. Two operators with the same observed snapshot cannot perform destructive steps simultaneously, but in-flight writes by an unquiesced prior executor still cannot be fenced by MongoDB alone. Never delete an ownership record by ID alone. + +## State-by-state recovery contract + +| Observed owner | After proven quiescence | Critical preservation rule | +|---|---|---| +| `kind:direct,status:uploading` | When a file is visible, verify every chunk, length and any claimed SHA before releasing ownership; otherwise delete orphan chunks and release | A mismatched caller-supplied metadata SHA blocks recovery; visible-file chunks are never deleted | +| `kind:delete,status:deleting` | Complete the interrupted delete of file and chunks before releasing ownership | No new upload may claim the same ID until both steps finish | +| `kind:cleanup,status:cleaning` | Check visible file; delete chunks only if no file exists; release | Never delete a visible file's chunks | +| `kind:rename,status:renaming` | Verify the resulting visible file has complete chunks, then release; missing or corrupt files remain locked | The tool does not choose or replay a rename target; the operator must inspect the intended filename | +| `status:pending,activeAppend` | Recompute the contiguous chunk prefix, clear only the observed append claim, restore seven-day idle expiry | Existing chunks are preserved; caller replays/checks the data | +| `status:finalizing` without file | Restore pending with expiry; retry normal finalize, which revalidates bytes | Never discard chunks to unlock finalize | +| `status:finalizing` with file | Re-hash the **actual** ordered chunks and check expected file length, chunk size and digest before releasing session | Forged metadata SHA is not sufficient | +| `status:aborting` without file | Delete invisible chunks then remove session | No newly created file can take ownership during recovery | +| `status:aborting` with file | **Blocked** for human forensic review | Never silently delete or detach a visible file | +| `status:recovering` | Resume by exact recovery token and reverified quiescence | Keep claim if any destructive step is ambiguous | + +### Known limitations and release criteria + +- MongoDB cannot revoke an already-running Node.js process's in-memory chunk write. `--quiesced` is a mandatory **external operational assertion**, not a technical fencing guarantee. For non-quiesceable workloads, a future generation-aware writer fencing design and driver integration would be needed; do **not** automate unlocking based on age. +- A direct uploadStream `metadata.sha256` is not authenticated at write time. The tool verifies visible chunk structure and length, but does not treat this metadata as a trusted digest for ordinary uploads. +- The recovery code is a source-checkout maintenance utility. It is not an online public `GridFSBucket` API and must not be exposed to unprivileged application callers. +- `drop()` is bucket-wide and ignores per-ID reservations; maintenance must globally stop writers before dropping the bucket. +- Deterministic tests must show all state paths, wrong-token and wrong-timestamp rejection, fail-closed interrupted recovery, no deletion of visible chunks, and idempotent restart after an interrupted abort/delete. **No merge until the code and this runbook pass the latest-HEAD required CI matrices and independent review.** diff --git a/docs/en/gridfs.md b/docs/en/gridfs.md index 1c631b8..d79980a 100644 --- a/docs/en/gridfs.md +++ b/docs/en/gridfs.md @@ -99,6 +99,8 @@ Retrying `finalizeResumableUpload` is idempotent while the session still exists: `.uploads` uses an `expiresAt` TTL index (`expireAfterSeconds: 0`). Pending resumable sessions expire after **7 days without a successful append**; a successful append refreshes their expiry. Active append/finalize/abort claims and direct upload/delete reservations have no TTL and must be recovered explicitly after a crash, otherwise a stale writer could resume after its lock expires. MongoDB TTL deletion is asynchronous and removes only the session document, not chunks. Known orphan chunks can be cleaned with `abortResumableUpload(uploadId)` after an exclusive cleanup claim, or by starting a new resumable upload with the same ID. Existing `createdAt` TTL indexes are migrated to `expiresAt`; legacy pending expiry is derived from their last update. The migration requires stopping/draining old GridFS writers; mixing old and new writer versions is unsafe. +**Crash and migration recovery:** see [the operator runbook](./gridfs-recovery.md). Recovery is dry-run by default and requires verified process quiescence, an exact state/token check and an audit record; do not delete stale-looking reservations manually. + ## Indexes `upload`, `uploadStream` (before the driver stream is opened), `beginResumableUpload`, and `ensureIndexes` create: diff --git a/docs/zh/gridfs-recovery.md b/docs/zh/gridfs-recovery.md new file mode 100644 index 0000000..c5c6139 --- /dev/null +++ b/docs/zh/gridfs-recovery.md @@ -0,0 +1,62 @@ +# GridFS 故障恢复与安全升级操作规程（候选 V0.1） + +> **仅供运维管理员使用，当前未完成 CI 验收。** 不得对仍在运行、暂停或网络分区状态的写入进程执行恢复。所有权记录时间较久，不等于原写入进程已经终止。禁止按超时自动抢占所有权。 + +## 不可破坏的安全约束 + +`.uploads` 持久化同一文件 ID 的上传、追加、完成、取消、删除、重命名和孤儿清理的互斥所有权。**文件记录不存在不代表其 chunks 可删**：上传过程中，files 文档尚未可见；删除过程中，MongoDB Driver 可能先删 files，再删 chunks。 + +任何强制恢复必须先从集群所有节点停止并排空旧版和新版 GridFS 写入进程、队列消费者、定时任务与自动重启，确认没有仍可恢复的旧写入。恢复脚本无法在 MongoDB 层直接阻止旧进程已发出的 chunks 写入；只有外部进程停机隔离后才安全。不得混用忽略 ownership 的旧版与新版本。 + +## 上线和回滚步骤 + +1. 冻结所有 GridFS 写入，禁止新请求和自动重启，核查所有旧版 Writer 已排空；保留相关运行实例、进程、队列、版本和时间证据。 +2. 备份 `.files`、`.chunks`、`.uploads`、索引及配置；备份应与目标集群隔离。 +3. 在冻结状态部署新版本并运行 `ensureIndexes()`。迁移需将 legacy `createdAt_1` TTL 改为 `expiresAt_1`，以最后活动时间计算普通待上传会话的七天有效期，活动所有权和无 TTL 状态不得被自动过期。迁移前就过期的旧会话可能已被 TTL 清除，需要额外核对孤儿 chunks。 +4. 对每一个需要恢复的 ID，核对上传会话状态、ownerToken/activeAppend.token/recovery.token、updatedAt、files 和 chunks 的一致性，确认目标动作并保留审计记录。 +5. **先 dry-run，后 apply**。apply 必须明确声明写入进程已停机、实际操作者、恢复原因，并提供精确原始状态、时间戳和令牌。脚本先写入审计记录，再使用条件更新取得 `recovering` 所有权，失败时保留该所有权，禁止直接删除锁。 +6. 核对实际 bytes、chunks 顺序/大小、可见性、恢复后状态，并完成当前 HEAD 的 Node 18/20/22、MongoDB 7.0.37/8.0.26、对应 Driver 6/7 和 Release Preflight 全矩阵后再恢复写入。 +7. **回滚到旧版并非直接安全**：旧版不会识别新 ownership，必须先停止并排空全部新版写入，审查所有未完成会话和残留锁，再评估可否回滚。不能简单恢复旧的 `createdAt` TTL。 + +## 操作方式 + +从已安装依赖的仓库源码目录执行，MongoDB URI 通过环境变量传递，不要在 shell 命令参数或 Issue 中公开凭证。 + +```sh +export MONSQLIZE_RECOVERY_MONGODB_URI='<从密钥管理系统读取的 MongoDB URI>' + +# 默认仅读取，不做变更；使用数据库里实际查询到的 ID、状态和时间 +node scripts/maintenance/gridfs-recover.cjs \ + --db=database --bucket=fs --id=0123456789abcdef01234567 \ + --expected-status=finalizing \ + --expected-updated-at=2026-11-12T02:00:00.000Z + +# 完成所有进程隔离与审计后，才允许执行 +node scripts/maintenance/gridfs-recover.cjs \ + --db=database --bucket=fs --id=0123456789abcdef01234567 \ + --expected-status=finalizing \ + --expected-updated-at=2026-11-12T02:00:00.000Z \ + --operator=oncall-id --reason=writers-drained \ + --quiesced --apply +``` + +以上 ID、时间戳仅为格式示例。如果记录存在 `ownerToken`、`activeAppend.token` 或 `recovery.token`，必须加 `--expected-token=实际令牌`。任何状态/令牌/时间戳不符都应中止，重新检查。**当 `status: recovering` 时，还必须先确认上一轮恢复执行器确实已停止，执行 apply 时加 `--prior-recovery-stopped`。**恢复续接会通过 CAS 原子轮换令牌，两个持同一旧快照的恢复进程不能同时执行删除；但仍无法隔离未停止的旧进程已经发出的 MongoDB 写入。 + +## 按状态恢复 + +| 状态 | 恢复动作 | 数据保护 | +|---|---|---| +| direct/uploading | 可见文件先验证 chunks 完整性及声明的 SHA 再解锁；没有 files 才清理孤儿 chunks | SHA 元数据不可信，但一旦与实际 bytes 不符就阻断；绝不删除可见文件 chunks | +| delete/deleting | 继续删除 files 和 chunks，之后才释放同 ID 的所有权 | 删除未完成时禁止替换上传 | +| cleanup/cleaning | 可见文件保留 chunks，否则清理孤儿 chunks | 先查 files | +| rename/renaming | 先核验可见文件及完整 chunks，缺失或损坏时保留锁；运维核查文件名 | 工具不猜测目标或重新执行 rename | +| pending + activeAppend | 根据 chunks 的连续前缀恢复 nextChunk，清除中断令牌，重新设置七天闲置过期 | 保留已写 chunks，后续需验证重传 | +| finalizing，文件不可见 | 退回 pending，正常 finalize 重新校验 | 不删已有 chunks | +| finalizing，文件可见 | 核对文件长度、chunk size、实际 SHA 和元数据，再清理遗留会话 | 不信任单独的元数据 SHA | +| aborting，文件不可见 | 清理孤儿 chunks，移除会话 | 无新写入抢占 | +| aborting，文件可见 | **阻断；人工分析** | 严禁删除可见文件 | +| recovering | 必须确认上一轮恢复执行器也已停止，加 `--prior-recovery-stopped` 并通过 CAS 取得新恢复令牌 | 同一旧快照不能双重恢复，失败仍保留锁 | + +## 失败和验收边界 + +恢复脚本位于源码 `scripts/maintenance/gridfs-recover.cjs`，不是对外公开 API。没有绝对可靠的在线“僵尸 Writer 自动判断”，因此不能靠超时解锁。必须以确定性测试覆盖异常中断后状态、错误令牌拒绝、错误时间戳拒绝、可见文件 chunks 绝不误删、finalize 重验 SHA、abort/delete 幂等续接、审计结果及两版 MongoDB。验证结果未全绿前不能把 PR #12 标记 Ready 或合入 main。 \ No newline at end of file diff --git a/docs/zh/gridfs.md b/docs/zh/gridfs.md index d29deb1..f82e31d 100644 --- a/docs/zh/gridfs.md +++ b/docs/zh/gridfs.md @@ -99,6 +99,8 @@ await bucket.abortResumableUpload(session.uploadId); `.uploads` 使用 `{ expiresAt: 1 }` TTL 索引，`expireAfterSeconds: 0`。pending 续传会话在**连续 7 天没有成功 append**时过期，每次成功 append 会刷新过期时间；正在执行的 append/finalize/abort，以及普通上传或删除的所有权记录不设置 TTL，进程异常退出后需确认旧进程停止再显式恢复，不能自动释放导致旧写入复活。MongoDB TTL 是异步清理，且只删除 session，不级联 chunks。已知 uploadId 的孤儿块由 `abortResumableUpload(uploadId)` 在取得独占清理权后回收，或复用同一 id 重新开始续传时清理。旧 `createdAt` TTL 索引会迁移为 `expiresAt`；迁移前必须停止/排空旧版 GridFS 写进程，新旧写入协议混跑不安全。 +**故障恢复与迁移：**参阅[运维恢复规程](./gridfs-recovery.md)。恢复默认只预览，必须在全部写入进程确实排空后，核对精确状态/令牌并留下审计记录；禁止直接删除看似过期的所有权。 + ## 索引 `upload`、`uploadStream`（打开驱动流之前）、`beginResumableUpload` 和 `ensureIndexes` 会创建： diff --git a/scripts/maintenance/gridfs-recover.cjs b/scripts/maintenance/gridfs-recover.cjs new file mode 100644 index 0000000..d55af79 --- /dev/null +++ b/scripts/maintenance/gridfs-recover.cjs @@ -0,0 +1,304 @@ +'use strict'; + +/** + * Operator-only recovery for GridFS ownership records after an externally + * verified writer shutdown. Never run against a live or potentially paused + * writer; MongoDB cannot fence a previously issued driver chunk write. + * + * Dry-run is default. The caller must supply the exact observed state, + * updatedAt timestamp and (where applicable) ownership token to apply. + * A durable "recovering" claim makes interruption/retry fail closed. + */ +const { randomUUID, createHash } = require('node:crypto'); +const { MongoClient, ObjectId } = require('mongodb'); + +const TTL_MS = 604800000; + +function fail(message) { + const error = new Error(message); + error.code = 'GRIDFS_RECOVERY_PRECONDITION_FAILED'; + throw error; +} + +function toBuffer(data) { + if (Buffer.isBuffer(data)) return data; + if (data && typeof data.value === 'function') return Buffer.from(data.value(true)); + if (data instanceof Uint8Array) return Buffer.from(data); + fail('Encountered a chunk with unsupported BSON binary data'); +} + +async function verifyVisibleFile(db, bucket, id, session) { + const files = db.collection(bucket + '.files'); + const doc = await files.findOne({ _id: id }); + if (!doc) return false; + if (!Number.isSafeInteger(doc.length) || doc.length < 0 || + !Number.isSafeInteger(doc.chunkSize) || doc.chunkSize < 1) { + fail('Visible file has invalid length/chunkSize; preserve ownership'); + } + const totalChunks = Math.ceil(doc.length / doc.chunkSize); + const digest = createHash('sha256'); + let n = 0; + let length = 0; + for await (const chunk of db.collection(bucket + '.chunks') + .find({ files_id: id }).sort({ n: 1 })) { + if (chunk.n !== n || n >= totalChunks) fail('Visible file has a missing, extra or out-of-order chunk'); + const bytes = toBuffer(chunk.data); + const want = n === totalChunks - 1 ? doc.length - n * doc.chunkSize : doc.chunkSize; + if (bytes.length !== want) fail('Visible file has a corrupt chunk length'); + digest.update(bytes); + length += bytes.length; + n++; + } + if (length !== doc.length || n !== totalChunks) { + fail('Visible file is incomplete; preserve ownership and chunks'); + } + const actualSha = digest.digest('hex'); + if (session) { + if (session.length !== doc.length || session.chunkSizeBytes !== doc.chunkSize || + doc.metadata?.sha256 !== session.sha256 || actualSha !== session.sha256) { + fail('Finalizing session and visible bytes mismatch; preserve ownership and chunks'); + } + } else if (typeof doc.metadata?.sha256 === 'string') { + // UploadStream SHA is caller-supplied metadata, not proof of integrity. + // A mismatched claimed hash is nevertheless evidence of inconsistency: + // do not unlock an anomalous direct upload as if it had been verified. + if (!/^[0-9a-f]{64}$/i.test(doc.metadata.sha256) || + actualSha !== doc.metadata.sha256.toLowerCase()) { + fail('Visible direct upload metadata SHA mismatch; preserve ownership and chunks'); + } + } + return true; +} + +function snapshotToken(doc) { + return doc.recovery?.token || doc.ownerToken || doc.activeAppend?.token || null; +} + +function classify(doc) { + let state = doc.status === 'recovering' ? doc.recovery?.fromStatus : null; + if (doc.status === 'recovering' && !state) fail('Recovering record lacks its original state'); + if (!state) { + if (doc.kind === 'direct' && doc.status === 'uploading') state = 'uploading'; + else if (doc.kind === 'delete' && doc.status === 'deleting') state = 'deleting'; + else if (doc.kind === 'rename' && doc.status === 'renaming') state = 'renaming'; + else if (doc.kind === 'cleanup' && doc.status === 'cleaning') state = 'cleaning'; + else if ((!doc.kind || doc.kind === 'resumable') && + doc.status === 'pending' && doc.activeAppend) state = 'activeAppend'; + else if ((!doc.kind || doc.kind === 'resumable') && + (doc.status === 'finalizing' || doc.status === 'aborting')) state = doc.status; + } + // Also validate the persisted origin when a crash left status=recovering: + // do not allow malformed/corrupt origin metadata to unlock arbitrary IDs. + const requiredKind = { + uploading: 'direct', deleting: 'delete', renaming: 'rename', cleaning: 'cleanup', + }; + if (!state || (requiredKind[state] && doc.kind !== requiredKind[state]) || + (['activeAppend', 'finalizing', 'aborting'].includes(state) && + doc.kind && doc.kind !== 'resumable') || + (state === 'activeAppend' && !doc.activeAppend?.token)) { + fail('Inconsistent or unsupported owner origin; preserve ownership'); + } + return state; +} + +async function contiguousChunkPrefix(db, bucket, id, session) { + if (!Number.isSafeInteger(session.length) || session.length < 1 || + !Number.isSafeInteger(session.chunkSizeBytes) || session.chunkSizeBytes < 1) { + fail('Pending session has invalid length/chunkSizeBytes; preserve ownership'); + } + const count = Math.ceil(session.length / session.chunkSizeBytes); + let next = 0; + for await (const doc of db.collection(bucket + '.chunks') + .find({ files_id: id }).sort({ n: 1 })) { + if (doc.n !== next || next >= count) { + fail('Interrupted append contains a gap or extra chunk; preserve ownership'); + } + const want = next === count - 1 + ? session.length - next * session.chunkSizeBytes : session.chunkSizeBytes; + if (toBuffer(doc.data).length !== want) { + fail('Interrupted append contains a malformed chunk; preserve ownership'); + } + next++; + } + return next; +} + +async function recoverGridFSOwnership(input) { + const { db, bucket, id, expectedStatus, expectedUpdatedAt, expectedToken, + quiesced, operator, reason, apply = false, resumeQuiesced = false } = input; + if (!db || !bucket || typeof bucket !== 'string' || !(id instanceof ObjectId)) fail('Invalid database, bucket or ObjectId'); + if (!expectedStatus || !expectedUpdatedAt) fail('Exact expected status and updatedAt are required'); + const uploads = db.collection(bucket + '.uploads'); + const doc = await uploads.findOne({ _id: id }); + if (!doc) return { outcome: 'no_owner', modified: false }; + if (!(doc.updatedAt instanceof Date) || + doc.status !== expectedStatus || + doc.updatedAt.toISOString() !== expectedUpdatedAt) { + fail('Owner status or updatedAt changed since inspection'); + } + const token = snapshotToken(doc); + if ((token || null) !== (expectedToken || null)) fail('Owner token changed since inspection'); + const sourceStatus = classify(doc); + const visible = await db.collection(bucket + '.files').findOne({ _id: id }, { projection: { _id: 1 } }); + const plan = { + bucket, uploadId: id.toHexString(), fromStatus: sourceStatus, + fileVisible: !!visible, + action: ({ + uploading: 'verify-visible-or-remove-orphan-chunks', + deleting: 'complete-file-and-chunk-deletion', + renaming: 'release-after-inspecting-rename-outcome', + cleaning: 'remove-invisible-orphan-chunks', + activeAppend: 'restore-pending-and-reconcile-prefix', + finalizing: visible ? 'verify-visible-bytes-and-release-session' : 'restore-pending-for-retry', + aborting: visible ? 'BLOCK-visible-file-conflict' : 'finish-invisible-chunk-removal', + })[sourceStatus], + }; + if (!apply) return { outcome: 'dry_run', modified: false, plan }; + if (quiesced !== true || typeof operator !== 'string' || !operator.trim() || + typeof reason !== 'string' || !reason.trim()) { + fail('Apply requires external writer quiescence, named operator and reason'); + } + if (sourceStatus === 'aborting' && visible) fail('Visible file conflicts with aborting state; manual inspection required'); + // A recovering owner may belong to a running recovery executor. Caller + // must separately prove that executor has stopped before a manual resume. + if (doc.status === 'recovering' && resumeQuiesced !== true) { + fail('Resuming recovery requires verified shutdown of the prior recovery executor'); + } + + const audit = db.collection(bucket + '.ownership_recovery_audit'); + const attemptId = randomUUID(); + await audit.insertOne({ + _id: attemptId, fileId: id, bucket, fromStatus: sourceStatus, + operator, reason, startedAt: new Date(), outcome: 'started', plan, + }); + let recoveryToken; + if (doc.status === 'recovering') { + // Concurrent recovery attempts with one observed snapshot must not share + // the same destructive authority. Atomically replace the old token. + // A prior executor must ALSO have been externally confirmed stopped. + recoveryToken = randomUUID(); + const claim = await uploads.updateOne( + { _id: id, status: 'recovering', updatedAt: doc.updatedAt, + 'recovery.token': doc.recovery.token }, + { $set: { updatedAt: new Date(), 'recovery.token': recoveryToken, + 'recovery.operator': operator, 'recovery.attemptId': attemptId, + 'recovery.startedAt': new Date() } }, + ); + if (claim.matchedCount !== 1) { + await audit.updateOne({ _id: attemptId }, { + $set: { outcome: 'claim_conflict', finishedAt: new Date() }, + }); + fail('Concurrent recovery executor claimed the same owner first'); + } + await audit.updateOne({ _id: attemptId }, { $set: { outcome: 'resumed', recoveryToken } }); + } else { + recoveryToken = randomUUID(); + const filter = { _id: id, status: doc.status, updatedAt: doc.updatedAt }; + if (doc.kind === undefined) filter.kind = { $exists: false }; + else filter.kind = doc.kind; + if (doc.ownerToken !== undefined) filter.ownerToken = doc.ownerToken; + if (doc.activeAppend) filter['activeAppend.token'] = doc.activeAppend.token; + const claim = await uploads.updateOne(filter, { + $set: { + status: 'recovering', updatedAt: new Date(), + recovery: { token: recoveryToken, fromStatus: sourceStatus, operator, attemptId, startedAt: new Date() }, + }, + $unset: { expiresAt: '' }, + }); + if (claim.matchedCount !== 1) { + await audit.updateOne({ _id: attemptId }, { $set: { outcome: 'claim_conflict', finishedAt: new Date() } }); + fail('Concurrent state change prevented recovery claim'); + } + } + const held = { _id: id, status: 'recovering', 'recovery.token': recoveryToken }; + try { + const chunks = db.collection(bucket + '.chunks'); + const files = db.collection(bucket + '.files'); + if (sourceStatus === 'activeAppend') { + const nextChunk = await contiguousChunkPrefix(db, bucket, id, doc); + const changed = await uploads.updateOne(held, { + $set: { status: 'pending', nextChunk, updatedAt: new Date(), + expiresAt: new Date(Date.now() + TTL_MS) }, + $unset: { activeAppend: '', recovery: '' }, + }); + if (changed.matchedCount !== 1) fail('Lost recovery claim while restoring append'); + } else if (sourceStatus === 'finalizing' && !visible) { + const changed = await uploads.updateOne(held, { + $set: { status: 'pending', updatedAt: new Date(), + expiresAt: new Date(Date.now() + TTL_MS) }, + $unset: { recovery: '' }, + }); + if (changed.matchedCount !== 1) fail('Lost recovery claim while restoring finalize'); + } else { + // Verify before ANY deletion. Never delete chunks of a visible file + // unless this is an explicitly quiesced, retained "deleting" owner. + if (sourceStatus === 'finalizing') { + await verifyVisibleFile(db, bucket, id, doc); + } else if ((sourceStatus === 'uploading' || sourceStatus === 'renaming') && visible) { + await verifyVisibleFile(db, bucket, id, null); + } else if (sourceStatus === 'renaming' && !visible) { + fail('Renaming owner has no visible file; preserve ownership for manual inspection'); + } + if (sourceStatus === 'deleting') { + await files.deleteOne({ _id: id }); + await chunks.deleteMany({ files_id: id }); + } else if ((sourceStatus === 'uploading' || sourceStatus === 'cleaning' || + sourceStatus === 'aborting') && !visible) { + await chunks.deleteMany({ files_id: id }); + } + const removed = await uploads.deleteOne(held); + if (removed.deletedCount !== 1) fail('Lost recovery claim while releasing owner'); + } + await audit.updateOne({ _id: attemptId }, { $set: { outcome: 'completed', finishedAt: new Date() } }); + return { outcome: 'completed', modified: true, plan, attemptId }; + } catch (error) { + // Deliberately retain recovering state; retry using its exact token and + // updatedAt only after re-verifying process quiescence. + try { + await audit.updateOne({ _id: attemptId }, + { $set: { outcome: 'blocked', error: String(error), finishedAt: new Date() } }); + } catch { /* The recovery claim remains; preserve original error. */ } + throw error; + } +} + +function readArg(args, name) { + const prefix = '--' + name + '='; + const found = args.find((a) => a.startsWith(prefix)); + return found && found.slice(prefix.length); +} + +async function main() { + const args = process.argv.slice(2); + const uri = process.env.MONSQLIZE_RECOVERY_MONGODB_URI; + if (!uri) fail('MONSQLIZE_RECOVERY_MONGODB_URI is required (do not put credentials in argv)'); + const dbName = readArg(args, 'db'); + const bucket = readArg(args, 'bucket'); + const idString = readArg(args, 'id'); + if (!dbName || !bucket || !idString || !ObjectId.isValid(idString)) fail('--db, --bucket and --id are required'); + const client = new MongoClient(uri); + await client.connect(); + try { + const report = await recoverGridFSOwnership({ + db: client.db(dbName), bucket, id: new ObjectId(idString), + expectedStatus: readArg(args, 'expected-status'), + expectedUpdatedAt: readArg(args, 'expected-updated-at'), + expectedToken: readArg(args, 'expected-token') || null, + operator: readArg(args, 'operator'), reason: readArg(args, 'reason'), + quiesced: args.includes('--quiesced'), + resumeQuiesced: args.includes('--prior-recovery-stopped'), + apply: args.includes('--apply'), + }); + process.stdout.write(JSON.stringify(report, null, 2) + '\n'); + } finally { + await client.close(); + } +} + +module.exports = { recoverGridFSOwnership, verifyVisibleFile }; +if (require.main === module) { + main().catch((error) => { + process.stderr.write(String(error.code || 'RECOVERY_ERROR') + ': ' + error.message + '\n'); + process.exitCode = 1; + }); +} diff --git a/scripts/validation/run-memory-server-matrix.cjs b/scripts/validation/run-memory-server-matrix.cjs index 65474e7..f6a7071 100644 --- a/scripts/validation/run-memory-server-matrix.cjs +++ b/scripts/validation/run-memory-server-matrix.cjs @@ -48,6 +48,7 @@ const integrationSourceSuites = [ 'test/integration/gridfs/gridfs.test.js', 'test/integration/gridfs/gridfs-scan-regressions.test.js', 'test/integration/gridfs/gridfs-ownership-regressions.test.js', + 'test/integration/gridfs/gridfs-stale-recovery.test.js', ]; const integrationSuites = integrationSourceSuites.map((suite) => path.join(testDistRoot, suite)); const compatibilitySuites = [ diff --git a/src/capabilities/gridfs/ownership.ts b/src/capabilities/gridfs/ownership.ts index 1518602..caac89b 100644 --- a/src/capabilities/gridfs/ownership.ts +++ b/src/capabilities/gridfs/ownership.ts @@ -30,10 +30,17 @@ export async function claimOwner( } export async function releaseOwner(ctx: GridFSContext, owner: GridFSOwner, operation: string): Promise { + const status = owner.kind === 'direct' ? 'uploading' + : owner.kind === 'delete' ? 'deleting' + : owner.kind === 'rename' ? 'renaming' : 'cleaning'; try { - await ctx.db.collection(uploadsCollectionName(ctx.bucketName)).deleteOne({ - _id: owner.id, kind: owner.kind, ownerToken: owner.token, + const released = await ctx.db.collection(uploadsCollectionName(ctx.bucketName)).deleteOne({ + _id: owner.id, kind: owner.kind, ownerToken: owner.token, status, }); + // Recovery intentionally retains the original kind/ownerToken while + // changing status to recovering. A late original writer must not be + // able to remove that new recovery fence with its stale ownerToken. + if (released.deletedCount !== 1) throw uploadBusy(operation, ctx, owner.id); } catch (error) { throw mapDriverError(error, operation, ctx, { uploadId: owner.id }); } diff --git a/test/integration/gridfs/gridfs-stale-recovery.test.ts b/test/integration/gridfs/gridfs-stale-recovery.test.ts new file mode 100644 index 0000000..4a19ad8 --- /dev/null +++ b/test/integration/gridfs/gridfs-stale-recovery.test.ts @@ -0,0 +1,437 @@ +import { createHash } from 'node:crypto'; +import { after, before, describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { Collection, MongoClient, ObjectId, type Db } from 'mongodb'; +import { createMemoryServerBootstrap } from '../../bootstrap/memory-server'; +import { releaseOwner } from '../../../src/capabilities/gridfs/ownership'; + +const MonSQLize = require('../../../dist/cjs/index.cjs'); +const { recoverGridFSOwnership } = require(process.cwd() + '/scripts/maintenance/gridfs-recover.cjs'); + +function digest(data: Buffer): string { + return createHash('sha256').update(data).digest('hex'); +} + +describe('GridFS operator recovery: deterministic post-crash state and chunk safety', () => { + const bootstrap = createMemoryServerBootstrap(); + let raw: MongoClient; + let uri: string; + let counter = 0; + + before(async () => { + uri = (await bootstrap.setup()).uri; + raw = new MongoClient(uri); + await raw.connect(); + }); + after(async () => { + await raw.close(); + await bootstrap.teardown(); + MonSQLize.Model._clear(); + }); + + async function withRuntime(fn: (bucket: any, db: Db) => Promise): Promise { + const dbName = 'gridfs_recovery_' + (++counter); + const runtime = new MonSQLize({ type: 'mongodb', databaseName: dbName, config: { uri } }); + await runtime.connect(); + try { + await fn(runtime.gridfs('b', { chunkSizeBytes: 4 }), raw.db(dbName)); + } finally { + await runtime.close(); + await raw.db(dbName).dropDatabase(); + } + } + + async function recover(db: Db, id: ObjectId, overrides: Record = {}) { + const doc = await db.collection('b.uploads').findOne({ _id: id }); + assert.ok(doc, 'recovery owner must exist'); + return recoverGridFSOwnership({ + db, bucket: 'b', id, + expectedStatus: doc.status, + expectedUpdatedAt: doc.updatedAt.toISOString(), + expectedToken: doc.recovery?.token || doc.ownerToken || doc.activeAppend?.token || null, + operator: 'integration-test', reason: 'deterministic process quiescence simulation', + quiesced: true, apply: true, ...overrides, + }); + } + + it('is read-only by default and rejects apply without explicit quiescence', async () => { + await withRuntime(async (_, db) => { + const id = new ObjectId(); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'direct', status: 'uploading', ownerToken: 'direct-1', + createdAt: now, updatedAt: now, + }); + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('ABCD') }); + const snapshot = await db.collection('b.uploads').findOne({ _id: id }); + const dry = await recover(db, id, { apply: false, quiesced: false }); + assert.equal(dry.outcome, 'dry_run'); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + await assert.rejects(recover(db, id, { quiesced: false }), + /external writer quiescence/); + await assert.rejects(recover(db, id, { expectedToken: 'incorrect' }), + /Owner token changed/); + await assert.rejects(recover(db, id, { expectedUpdatedAt: '2020-01-01T00:00:00.000Z' }), + /updatedAt changed/); + assert.deepEqual(await db.collection('b.uploads').findOne({ _id: id }), snapshot); + await recover(db, id); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 0); + assert.equal(await db.collection('b.ownership_recovery_audit').countDocuments({ outcome: 'completed' }), 1); + }); + }); + + it('preserves committed direct upload bytes while releasing a stranded owner', async () => { + await withRuntime(async (bucket, db) => { + const id = new ObjectId(); + const data = Buffer.from('GOODGOOD'); + await bucket.upload('committed.bin', data, { id, sha256: digest(data) }); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'direct', status: 'uploading', ownerToken: 'post-commit', + createdAt: now, updatedAt: now, + }); + await recover(db, id); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 2); + assert.deepEqual(await bucket.download(id), data); + }); + }); + + it('completes interrupted delete without exposing replacement chunks', async () => { + await withRuntime(async (bucket, db) => { + const id = new ObjectId(); + await bucket.upload('old.bin', Buffer.from('ABCD'), { id }); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'delete', status: 'deleting', ownerToken: 'delete-1', + createdAt: now, updatedAt: now, + }); + await db.collection('b.files').deleteOne({ _id: id }); // failure after files deletion + await recover(db, id); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 0); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + await bucket.upload('replacement.bin', Buffer.from('NEW!'), { id }); + assert.deepEqual(await bucket.download(id), Buffer.from('NEW!')); + }); + }); + + it('restores stale activeAppend with a reconciled contiguous nextChunk and expiry', async () => { + await withRuntime(async (bucket, db) => { + const body = Buffer.from('AAAABBBB'); + const id = (await bucket.beginResumableUpload('append.bin', + { length: body.length, sha256: digest(body) })).uploadId; + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('AAAA') }); + await db.collection('b.uploads').updateOne({ _id: id }, { + $set: { activeAppend: { token: 'append-crash', startedAt: new Date() }, updatedAt: new Date() }, + $unset: { expiresAt: '' }, + }); + await recover(db, id); + const session = await db.collection('b.uploads').findOne({ _id: id }); + assert.equal(session?.status, 'pending'); + assert.equal(session?.nextChunk, 1); + assert.equal(session?.activeAppend, undefined); + assert.ok(session?.expiresAt instanceof Date); + await bucket.appendChunks(id, 1, Buffer.from('BBBB')); + await bucket.finalizeResumableUpload(id); + assert.deepEqual(await bucket.download(id), body); + }); + }); + + it('keeps interrupted append fenced when an existing chunk is malformed', async () => { + await withRuntime(async (bucket, db) => { + const body = Buffer.from('AAAABBBB'); + const id = (await bucket.beginResumableUpload('corrupt-append.bin', + { length: body.length, sha256: digest(body) })).uploadId; + await db.collection('b.chunks').insertOne({ + files_id: id, n: 0, data: Buffer.from('A'), + }); + await db.collection('b.uploads').updateOne({ _id: id }, { + $set: { activeAppend: { token: 'corrupt-append', startedAt: new Date() }, + updatedAt: new Date() }, + $unset: { expiresAt: '' }, + }); + await assert.rejects(recover(db, id), /malformed chunk/); + const owner = await db.collection('b.uploads').findOne({ _id: id }); + assert.equal(owner?.status, 'recovering'); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + assert.equal((await db.collection('b.chunks').findOne({ files_id: id }))?.n, 0); + }); + }); + + it('recovers pre-commit finalizing without deleting verified chunks', async () => { + await withRuntime(async (bucket, db) => { + const body = Buffer.from('AAAABBBB'); + const id = (await bucket.beginResumableUpload('finalize.bin', + { length: body.length, sha256: digest(body) })).uploadId; + await bucket.appendChunks(id, 0, body); + await db.collection('b.uploads').updateOne({ _id: id }, { + $set: { status: 'finalizing', updatedAt: new Date() }, + $unset: { expiresAt: '' }, + }); + await recover(db, id); + const doc = await db.collection('b.uploads').findOne({ _id: id }); + assert.equal(doc?.status, 'pending'); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 2); + await bucket.finalizeResumableUpload(id); + assert.deepEqual(await bucket.download(id), body); + }); + }); + + it('recovers post-commit finalizing only after rehash of actual visible bytes', async () => { + await withRuntime(async (bucket, db) => { + const body = Buffer.from('GOOD'); + const hash = digest(body); + const id = (await bucket.beginResumableUpload('done.bin', + { length: body.length, sha256: hash })).uploadId; + await bucket.appendChunks(id, 0, body); + await db.collection('b.uploads').updateOne({ _id: id }, { + $set: { status: 'finalizing', updatedAt: new Date() }, + $unset: { expiresAt: '' }, + }); + await db.collection('b.files').insertOne({ + _id: id, filename: 'done.bin', length: 4, chunkSize: 4, + uploadDate: new Date(), metadata: { sha256: hash }, + }); + await recover(db, id); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + assert.deepEqual(await bucket.download(id), body); + }); + }); + + it('fails closed on forged metadata SHA and leaves chunks/ownership untouched', async () => { + await withRuntime(async (bucket, db) => { + const body = Buffer.from('GOOD'); + const id = (await bucket.beginResumableUpload('wrong.bin', + { length: body.length, sha256: digest(body) })).uploadId; + await bucket.appendChunks(id, 0, Buffer.from('EVIL')); + await db.collection('b.uploads').updateOne({ _id: id }, { + $set: { status: 'finalizing', updatedAt: new Date() }, + $unset: { expiresAt: '' }, + }); + await db.collection('b.files').insertOne({ + _id: id, filename: 'wrong.bin', length: 4, chunkSize: 4, + uploadDate: new Date(), metadata: { sha256: digest(body) }, + }); + await assert.rejects(recover(db, id), /visible bytes mismatch/); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + const owner = await db.collection('b.uploads').findOne({ _id: id }); + assert.equal(owner?.status, 'recovering', 'failed recovery remains fenced'); + assert.ok(owner?.recovery?.token); + assert.equal(await db.collection('b.ownership_recovery_audit').countDocuments({ outcome: 'blocked' }), 1); + }); + }); + + it('completes aborting and is idempotent when the prior owner is gone', async () => { + await withRuntime(async (bucket, db) => { + const body = Buffer.from('GOOD'); + const id = (await bucket.beginResumableUpload('aborting.bin', + { length: body.length, sha256: digest(body) })).uploadId; + await bucket.appendChunks(id, 0, body); + await db.collection('b.uploads').updateOne({ _id: id }, { + $set: { status: 'aborting', updatedAt: new Date() }, + $unset: { expiresAt: '' }, + }); + const receipt = await recover(db, id); + assert.equal(receipt.outcome, 'completed'); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 0); + assert.equal((await recoverGridFSOwnership({ + db, bucket: 'b', id, expectedStatus: 'aborting', + expectedUpdatedAt: new Date().toISOString(), + })).outcome, 'no_owner'); + await bucket.beginResumableUpload('new.bin', { id, length: 4, sha256: digest(body) }); + }); + }); + + it('can resume a recovery crash using the persistent recovery token', async () => { + await withRuntime(async (_, db) => { + const id = new ObjectId(); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'cleanup', status: 'recovering', + ownerToken: 'old-claim', createdAt: now, updatedAt: now, + recovery: { token: 'recovery-claim', fromStatus: 'cleaning', operator: 'original' }, + }); + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('ABCD') }); + await recover(db, id, { resumeQuiesced: true }); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 0); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + }); + }); + + it('rejects corrupt recovery origin metadata rather than unlocking the ID', async () => { + await withRuntime(async (_, db) => { + const id = new ObjectId(); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'delete', status: 'recovering', ownerToken: 'old', + createdAt: now, updatedAt: now, + recovery: { token: 'recovery-token', fromStatus: 'uploading' }, + }); + await db.collection('b.chunks').insertOne({ + files_id: id, n: 0, data: Buffer.from('GOOD'), + }); + await assert.rejects(recover(db, id), /Inconsistent or unsupported owner origin/); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 1); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + }); + }); + + it('refuses to clear aborting ownership if a visible file already owns the chunks', async () => { + await withRuntime(async (bucket, db) => { + const id = new ObjectId(); + await bucket.upload('visible.bin', Buffer.from('GOOD'), { id }); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'resumable', status: 'aborting', + filename: 'bad.bin', length: 4, chunkSizeBytes: 4, sha256: digest(Buffer.from('GOOD')), + createdAt: now, updatedAt: now, + }); + await assert.rejects(recover(db, id), /Visible file conflicts/); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 1); + }); + }); + + it('requires proof the previous recovery process stopped before resuming', async () => { + await withRuntime(async (_, db) => { + const id = new ObjectId(); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'cleanup', status: 'recovering', + ownerToken: 'original', createdAt: now, updatedAt: now, + recovery: { token: 'old-recovery', fromStatus: 'cleaning' }, + }); + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('GOOD') }); + await assert.rejects(recover(db, id), /verified shutdown of the prior recovery executor/); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 1); + }); + }); + + it('atomically rejects two recovery resumptions holding the same snapshot', { timeout: 30_000 }, async () => { + await withRuntime(async (bucket, db) => { + const id = new ObjectId(); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'cleanup', status: 'recovering', + ownerToken: 'original', createdAt: now, updatedAt: now, + recovery: { token: 'old-recovery', fromStatus: 'cleaning' }, + }); + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('GOOD') }); + const original = Collection.prototype.updateOne; + let ready!: () => void; + const bothReached = new Promise((resolve) => { ready = resolve; }); + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + let claims = 0; + (Collection.prototype as any).updateOne = async function (this: Collection, ...args: unknown[]) { + const filter = args[0] as Record; + if (this.collectionName === 'b.uploads' && filter.status === 'recovering' + && filter['recovery.token'] === 'old-recovery') { + claims += 1; + if (claims === 2) ready(); + await gate; + } + return (original as any).apply(this, args); + }; + try { + const input = { + db, bucket: 'b', id, + expectedStatus: 'recovering', expectedUpdatedAt: now.toISOString(), + expectedToken: 'old-recovery', quiesced: true, resumeQuiesced: true, + operator: 'integration-test', reason: 'verified prior executor stop', + apply: true, + }; + const first = recoverGridFSOwnership(input); + const second = recoverGridFSOwnership(input); + await bothReached; + release(); + const outcomes = await Promise.allSettled([first, second]); + assert.equal(outcomes.filter((x) => x.status === 'fulfilled').length, 1); + assert.equal(outcomes.filter((x) => x.status === 'rejected').length, 1); + assert.equal(await db.collection('b.ownership_recovery_audit').countDocuments({ outcome: 'completed' }), 1); + assert.equal(await db.collection('b.ownership_recovery_audit').countDocuments({ outcome: 'claim_conflict' }), 1); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 0); + const payload = Buffer.from('NEW!'); + await bucket.upload('replacement.bin', payload, { id }); + assert.deepEqual(await bucket.download(id), payload); + } finally { + release(); + Collection.prototype.updateOne = original; + } + }); + }); + + + it('does not release a visible direct upload with inconsistent claimed SHA', async () => { + await withRuntime(async (bucket, db) => { + const id = new ObjectId(); + const actual = Buffer.from('EVIL'); + await bucket.upload('direct.bin', actual, { id }); + await db.collection('b.files').updateOne({ _id: id }, { + $set: { 'metadata.sha256': digest(Buffer.from('GOOD')) }, + }); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'direct', status: 'uploading', ownerToken: 'crash', + createdAt: now, updatedAt: now, + }); + await assert.rejects(recover(db, id), /metadata SHA mismatch/); + assert.equal((await db.collection('b.uploads').findOne({ _id: id }))?.status, 'recovering'); + assert.deepEqual(await bucket.download(id), actual); + }); + }); + + it('releases a quiesced rename only after checking a visible complete file', async () => { + await withRuntime(async (bucket, db) => { + const id = new ObjectId(); + const payload = Buffer.from('GOODGOOD'); + await bucket.upload('already-renamed.bin', payload, { id }); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'rename', status: 'renaming', ownerToken: 'rename-crash', + createdAt: now, updatedAt: now, + }); + await recover(db, id); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + assert.deepEqual(await bucket.download(id), payload); + }); + }); + + it('keeps a missing-file rename fenced pending manual inspection', async () => { + await withRuntime(async (_, db) => { + const id = new ObjectId(); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'rename', status: 'renaming', ownerToken: 'rename-crash', + createdAt: now, updatedAt: now, + }); + await assert.rejects(recover(db, id), /no visible file/); + assert.equal((await db.collection('b.uploads').findOne({ _id: id }))?.status, 'recovering'); + }); + }); + + + it('does not let a stale original writer release an active recovery claim', async () => { + await withRuntime(async (_, db) => { + const id = new ObjectId(); + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'direct', status: 'recovering', + ownerToken: 'original-writer', createdAt: now, updatedAt: now, + recovery: { token: 'active-recovery', fromStatus: 'uploading' }, + }); + const ctx = { db, bucketName: 'b' } as any; + await assert.rejects( + releaseOwner(ctx, { id, kind: 'direct', token: 'original-writer' }, 'upload'), + (error: unknown) => (error as { code?: string }).code === 'GRIDFS_UPLOAD_BUSY', + ); + const owner = await db.collection('b.uploads').findOne({ _id: id }); + assert.equal(owner?.status, 'recovering'); + assert.equal(owner?.recovery?.token, 'active-recovery'); + }); + }); + +}); diff --git a/test/run-tests.cjs b/test/run-tests.cjs index e15017e..0627adf 100644 --- a/test/run-tests.cjs +++ b/test/run-tests.cjs @@ -138,6 +138,7 @@ const INTEGRATION = [ 'test/integration/gridfs/gridfs.test.js', 'test/integration/gridfs/gridfs-scan-regressions.test.js', 'test/integration/gridfs/gridfs-ownership-regressions.test.js', + 'test/integration/gridfs/gridfs-stale-recovery.test.js', 'test/integration/mongodb/review-repair-regressions.test.js', 'test/integration/model/review-model-regressions.test.js', 'test/integration/slow-query-log/review-storage-regressions.test.js',