diff --git a/.github/workflows/gridfs-fix-validation.yml b/.github/workflows/gridfs-fix-validation.yml index 48b72b8..c139c92 100644 --- a/.github/workflows/gridfs-fix-validation.yml +++ b/.github/workflows/gridfs-fix-validation.yml @@ -42,7 +42,10 @@ jobs: run: | node --test --test-concurrency=1 \ .generated/test-dist/test/unit/gridfs/gridfs.test.js \ + .generated/test-dist/test/unit/gridfs/stream-chunks.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-stale-recovery.test.js + .generated/test-dist/test/integration/gridfs/gridfs-stale-recovery.test.js \ + .generated/test-dist/test/integration/gridfs/gridfs-remaining-issues.test.js \ + .generated/test-dist/test/integration/gridfs/gridfs-inspect-orphans.test.js diff --git a/config/package-budget.json b/config/package-budget.json index 19f781d..c46e87d 100644 --- a/config/package-budget.json +++ b/config/package-budget.json @@ -1,8 +1,8 @@ { - "comment": "The 3.5.3 bundle ceilings were 881096 CJS and 878096 ESM. The GridFS cross-operation ownership, durable session CAS/abort fencing, delete protection and idle-TTL migration add ~4.4/4.7 KiB over that ceiling (CI build 2026-11-12: 885536 CJS / 882819 ESM). Raise to 890000 / 887000, keeping ~4 KiB reviewed headroom; no dependency change. Enforce these ceilings in release preflight.", + "comment": "GridFS session identity checks bring the 2026-11-13 build to 897830 CJS and 895065 ESM bytes, with no dependency change. Release preflight enforces these ceilings.", "artifacts": { - "dist/cjs/index.cjs": 890000, - "dist/esm/index.mjs": 887000, + "dist/cjs/index.cjs": 898000, + "dist/esm/index.mjs": 895500, "dist/cjs/cli/data-task.cjs": 50000 } } diff --git a/docs/en/gridfs-recovery.md b/docs/en/gridfs-recovery.md index 5409cf5..888f912 100644 --- a/docs/en/gridfs-recovery.md +++ b/docs/en/gridfs-recovery.md @@ -6,7 +6,14 @@ 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. +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. Dry-run plans may use a pre-claim visibility snapshot; **apply always re-reads `files` after the exclusive claim**. If visibility changed, or `verifyVisibleFile` cannot confirm a visible file that the plan required, the tool fails closed and leaves the `recovering` owner in place. + +To survey leftover chunks with ObjectId file IDs, use the inspect-only reporter `scripts/maintenance/gridfs-inspect-orphans.cjs`. It walks `.chunks` in index order in bounded batches, re-checks `files` after finding neither a file nor an owner for a `files_id`, and reports `visible_file`, `owned`, or `candidate_orphan`. It has no apply mode, does not delete, and does not use document age as a deletion criterion. It is not part of the public `GridFSBucket` API. The CLI emits one NDJSON finding per line plus a final summary object. `--max-findings` (default 1000, cap 100000) bounds memory; a truncated scan returns `resumeAfter.filesId` for the next `--after-files-id`. Summary counters accumulate incrementally. + +```sh +export MONSQLIZE_RECOVERY_MONGODB_URI='' +node scripts/maintenance/gridfs-inspect-orphans.cjs --db=my_database --bucket=fs --page-size=200 --max-findings=1000 +``` ## Deployment and rollback gates (required) @@ -54,6 +61,7 @@ Every apply attempt includes an audit entry. A completed visible-file verificati | `kind:direct,status:uploading` | Verify chunk sequence, per-chunk length and total length. Enforce SHA only when the owner records a pre-verified `upload()` expected digest; otherwise record any metadata mismatch as an audit warning | `uploadStream()` and arbitrary `metadata.sha256` are untrusted claims; 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:resumable,status:initializing` | Preserve visible chunks or remove invisible leftovers, then release the initialization claim | Retry `beginResumableUpload` after recovery if no file is visible | | `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 | diff --git a/docs/en/gridfs.md b/docs/en/gridfs.md index 85d6d5f..45e85ef 100644 --- a/docs/en/gridfs.md +++ b/docs/en/gridfs.md @@ -46,7 +46,7 @@ Calling either method before `connect()` throws `NOT_CONNECTED`. `upload` defaults to `ifExists: 'error'`. - When `sha256` is set, the conflict key is `metadata.sha256`. The digest must be 64 lowercase hex characters. `Buffer` and `Uint8Array` payloads are hashed before any GridFS write. A mismatch throws `GRIDFS_INTEGRITY_MISMATCH` and leaves no visible file. -- A `Readable` passed with `sha256` is buffered completely, then hashed, before any write. Large files should use the resumable API instead. +- A `Readable` passed with `sha256` is buffered completely, then hashed, before any write. That is required to honor the contract that a failed verification leaves no visible file. `upload()` without `sha256` can stream, but the whole payload is still a single commit. For large files, use the resumable API with `appendChunks` and a `Readable` whose `highWaterMark` stays modest; do not pass a multi-gigabyte `Buffer` or a fully-buffered `upload({ sha256 })`. - `skip` with `sha256` returns the newest visible file with that digest. `skip` without `sha256` returns the newest visible file with the same filename. `duplicate` always creates another file. - A `metadata.sha256` value supplied inside `metadata` is caller data and is not verified. To request byte verification with `upload()`, use the top-level `sha256` option. - `uploadStream` stores its top-level `sha256` option in metadata only. It does not check the bytes and it does not accept `ifExists`. @@ -75,13 +75,13 @@ const info = await bucket.finalizeResumableUpload(session.uploadId); await bucket.abortResumableUpload(session.uploadId); ``` -`beginResumableUpload` requires `length` and `sha256`. An id that already exists in `files` or `uploads` throws `GRIDFS_FILE_EXISTS`. +`beginResumableUpload` requires `length` and `sha256`. An id already present in `files` or `uploads` throws `GRIDFS_FILE_EXISTS`. The insert claims the id in `initializing` without expiry. It rechecks `files` and cleans old chunks before making the session `pending`. While initialization runs, other writes to that id are blocked. If a file became visible before the recheck, the call preserves its chunks and throws `GRIDFS_FILE_EXISTS`. An interrupted initialization can be cleared through operator recovery after the writer has stopped. Chunk bytes must follow the session size: - Every chunk except the last has length `chunkSizeBytes`. - The last chunk has length `length % chunkSizeBytes`, or `chunkSizeBytes` when the length divides evenly. -- A payload that does not split into those sizes throws `INVALID_ARGUMENT`. The whole payload is checked before any chunk is inserted. +- A payload that does not split into those sizes throws `INVALID_ARGUMENT`. For a `Buffer` or `Uint8Array`, the whole payload is planned and checked before any chunk is inserted. For a `Readable`, complete chunks are written as they arrive; a trailing partial or oversize remainder is rejected during consumption, after any complete prefix chunks have been stored. A failed append throws and does not return `nextChunk`. Resume from the last successfully acknowledged position; if the stored prefix has moved, `GRIDFS_RESUME_MISMATCH` includes `expectedNextChunk`. A source error during automatic destruction after EOF also rejects the append. Memory use is bounded by the current stream buffer, the partial chunk, and a fixed-size batch of stored chunk indexes. Use a modest `highWaterMark` for large files. `appendChunks` is addressed by `fromChunk`: @@ -92,9 +92,11 @@ Chunk bytes must follow the session size: - A payload that contains any new chunk must start at `nextChunk`. Starting earlier throws `GRIDFS_RESUME_MISMATCH` and writes nothing. - A duplicate `{ files_id, n }` insert is accepted only when the stored bytes match. -`finalizeResumableUpload` requires a contiguous prefix, a total length equal to the session length, and a sha256 equal to the session digest. It then inserts the files document, sets `metadata.sha256`, and deletes the session. Missing chunk indexes throw `GRIDFS_UPLOAD_INCOMPLETE` with `details[0].missingChunks`. A bad digest or a chunk of the wrong size throws `GRIDFS_INTEGRITY_MISMATCH`. The files document is not inserted in those cases. +`finalizeResumableUpload` requires exactly one chunk at each expected index starting at zero, a total length equal to the session length, and a sha256 equal to the session digest. It then inserts the files document, sets `metadata.sha256`, and deletes the session. Missing chunk indexes throw `GRIDFS_UPLOAD_INCOMPLETE`. `details[0].missingChunks` lists the first 1,024 missing indexes at most. When truncated, `details[0].missingChunkCount` gives the total. An invalid, duplicate, or out-of-range chunk index, a bad digest, or a chunk of the wrong size throws `GRIDFS_INTEGRITY_MISMATCH`. The files document is not inserted in those cases. -Retrying `finalizeResumableUpload` is idempotent while the session still exists: if a file with the same id is already visible and its `metadata.sha256` and length match the session, the call returns that file and deletes the session. A file with a different sha256 or length still throws `GRIDFS_FILE_EXISTS`. Once the session is gone (a finalize that fully succeeded, an abort, or the TTL), there is no digest to compare, so a retry throws `GRIDFS_FILE_NOT_FOUND`; use `getFile(uploadId)` to check the result. +Retrying `finalizeResumableUpload` is idempotent while the session still exists: if a file with the same id is already visible, the library re-hashes the actual stored chunks and returns the file only when that digest, the file length, and the chunk size match the session. A different sha256 or length throws `GRIDFS_FILE_EXISTS`. A chunk-size mismatch throws `GRIDFS_INTEGRITY_MISMATCH` and retains the existing finalizing session. A retry that observes a different upload under the same ID returns `GRIDFS_UPLOAD_BUSY`. Once the session is gone (a finalize that fully succeeded, an abort, or the TTL), there is no digest to compare, so a retry throws `GRIDFS_FILE_NOT_FOUND`. Callers must persist `uploadId`, `length`, and `sha256` themselves. After a vanished session, use `getFile(uploadId)` to see whether a file is visible, then re-hash the **actual** bytes (for example `download(uploadId)` or a walk of `.chunks`) and compare them to the persisted length and digest. Never treat `metadata.sha256` alone as proof: `uploadStream()` and a raw `metadata.sha256` can store an unverified claim. This package does not add a separate public verification API. + +A `finalizeResumableUpload` that fails with `GRIDFS_FILE_EXISTS` or `GRIDFS_INTEGRITY_MISMATCH` after it has claimed `finalizing` rolls the session back to `pending` so `abortResumableUpload` (and then a delete of a conflicting visible file) can proceed through the public API. Ambiguous driver errors during the files insert still leave `finalizing` for explicit recovery. `abortResumableUpload` atomically claims a pending session before deleting chunks. It fails with `GRIDFS_UPLOAD_BUSY` while append or finalize owns the session. It deletes the session and, when no files document is visible, that upload's chunks. A missing or already aborted session resolves, and any chunks left under that id without a visible file are still deleted. Aborting an id that already has a visible file does not delete that file's chunks. @@ -136,7 +138,7 @@ A blocked call throws `INVALID_OPERATION` before any GridFS write. `details[0]` |------|------| | `GRIDFS_FILE_NOT_FOUND` | Download, delete, rename, or resumable session target is missing | | `GRIDFS_FILE_EXISTS` | `ifExists: 'error'` finds a conflict, or the chosen id is already present | -| `GRIDFS_INTEGRITY_MISMATCH` | Bytes do not match `sha256`, or a finalized chunk has the wrong size | +| `GRIDFS_INTEGRITY_MISMATCH` | Bytes do not match `sha256`, or finalize found an invalid chunk index or size | | `GRIDFS_UPLOAD_INCOMPLETE` | Finalize found gaps. `details[0].missingChunks` lists them | | `GRIDFS_RESUME_MISMATCH` | `fromChunk` is ahead of the contiguous prefix, or a replayed chunk differs | | `GRIDFS_UPLOAD_BUSY` | Another append/finalize/abort/upload/delete operation owns the ID; do not retry until it finishes or is explicitly recovered | diff --git a/docs/en/support-matrix.md b/docs/en/support-matrix.md index 7c1a944..e3f1d69 100644 --- a/docs/en/support-matrix.md +++ b/docs/en/support-matrix.md @@ -18,7 +18,7 @@ | MongoDB Driver | 6.x / 7.x | `npm run test:server-matrix` | | MongoDB Server | 7.x / 8.x | strict `mongodb-memory-server` 7.0.37 / 8.0.26 single + replica set matrix | -GridFS is a client convention on `.files` and `.chunks`. Its integration tests run on `mongodb-memory-server` 7.0.37. That does not add MongoDB 4.x, or any server outside 7.x / 8.x, to this matrix. +GridFS is a client convention on `.files` and `.chunks`. Its integration tests run on `mongodb-memory-server` 7.0.37 and 8.0.26 in the current GridFS safety / release-gate jobs. That does not add MongoDB 4.x, or any server outside 7.x / 8.x, to this matrix. Search and Vector Search against Atlas, atlas-local, or mongot remain UNVERIFIED in this checkout; those deployments are not implied by the memory-server matrix. ## Default verification method diff --git a/docs/zh/gridfs-recovery.md b/docs/zh/gridfs-recovery.md index 3ef801a..7370b0e 100644 --- a/docs/zh/gridfs-recovery.md +++ b/docs/zh/gridfs-recovery.md @@ -8,6 +8,16 @@ 任何强制恢复必须先从集群所有节点停止并排空旧版和新版 GridFS 写入进程、队列消费者、定时任务与自动重启，确认没有仍可恢复的旧写入。恢复脚本无法在 MongoDB 层直接阻止旧进程已发出的 chunks 写入；只有外部进程停机隔离后才安全。不得混用忽略 ownership 的旧版与新版本。 +dry-run 可以使用领取所有权之前的可见性快照；**apply 必须在独占 claim 之后重新读取 `files`**。若可见性已经变化，或在需要可见文件时 `verifyVisibleFile` 无法确认，脚本失败并保留 `recovering` 所有权。 + +排查使用 ObjectId 的残留 chunks 时，使用只读巡检脚本 `scripts/maintenance/gridfs-inspect-orphans.cjs`。它按已有 `{ files_id, n }` 索引分批扫描 `.chunks`。分类时若未查到 `files` 和 `uploads` 记录，会再次核对 `files`，报告 `visible_file`、`owned` 或 `candidate_orphan`。没有 apply 模式，不会删除，也不会按文档年龄删除。它不是公开的 `GridFSBucket` API。CLI 按行输出 NDJSON 结果及一条摘要。`--max-findings`（默认 1000，上限 100000）限制单次内存；截断时用 `resumeAfter.filesId` 作为下次的 `--after-files-id`。摘要计数在扫描过程中增量累计。 + +```sh +export MONSQLIZE_RECOVERY_MONGODB_URI='<从密钥管理系统读取的 MongoDB URI>' +node scripts/maintenance/gridfs-inspect-orphans.cjs --db=database --bucket=fs --page-size=200 --max-findings=1000 +``` + + ## 上线和回滚步骤 1. 冻结所有 GridFS 写入，禁止新请求和自动重启，核查所有旧版 Writer 已排空；保留相关运行实例、进程、队列、版本和时间证据。 @@ -49,6 +59,7 @@ node scripts/maintenance/gridfs-recover.cjs \ | direct/uploading | 验证 chunks 连续性、每块长度和总长度；只有 owner 记录了 upload() 写入前验证过的 expected SHA 时才强制校验摘要，否则将 metadata 不一致记为审计 warning | uploadStream() 与任意 metadata.sha256 都是不可信声明；绝不删除可见文件 chunks | | delete/deleting | 继续删除 files 和 chunks，之后才释放同 ID 的所有权 | 删除未完成时禁止替换上传 | | cleanup/cleaning | 可见文件保留 chunks，否则清理孤儿 chunks | 先查 files | +| resumable/initializing | 保留可见文件的分片，或清理不可见的残留分片，再释放初始化占位 | 文件不可见时，恢复完成后可重试 `beginResumableUpload` | | rename/renaming | 先核验可见文件及完整 chunks，缺失或损坏时保留锁；运维核查文件名 | 工具不猜测目标或重新执行 rename | | pending + activeAppend | 根据 chunks 的连续前缀恢复 nextChunk，清除中断令牌，重新设置七天闲置过期 | 保留已写 chunks，后续需验证重传 | | finalizing，文件不可见 | 退回 pending，正常 finalize 重新校验 | 不删已有 chunks | @@ -61,4 +72,4 @@ Direct owner 会保存摘要来源：`upload()` 的 `options.sha256` 在写入 ## 失败和验收边界 -恢复脚本位于源码 `scripts/maintenance/gridfs-recover.cjs`，不是对外公开 API。没有绝对可靠的在线“僵尸 Writer 自动判断”，因此不能靠超时解锁。必须以确定性测试覆盖异常中断后状态、错误令牌拒绝、错误时间戳拒绝、可见文件 chunks 绝不误删、finalize 重验 SHA、abort/delete 幂等续接、审计结果及两版 MongoDB。验证结果未全绿前不能把 PR #12 标记 Ready 或合入 main。 \ No newline at end of file +恢复脚本位于源码 `scripts/maintenance/gridfs-recover.cjs`，不是对外公开 API。没有绝对可靠的在线“僵尸 Writer 自动判断”，因此不能靠超时解锁。必须以确定性测试覆盖异常中断后状态、错误令牌拒绝、错误时间戳拒绝、可见文件 chunks 绝不误删、finalize 重验 SHA、abort/delete 幂等续接、审计结果及两版 MongoDB。验证结果未全绿前不能把 PR #12 标记 Ready 或合入 main。 diff --git a/docs/zh/gridfs.md b/docs/zh/gridfs.md index e08df52..02cceaf 100644 --- a/docs/zh/gridfs.md +++ b/docs/zh/gridfs.md @@ -46,7 +46,7 @@ const modelBucket = model.gridfs(bucketName?: string, opts?: { `upload` 的 `ifExists` 默认为 `'error'`。 - 提供 `sha256` 时，冲突键是 `metadata.sha256`。摘要必须是 64 位小写十六进制。`Buffer` 与 `Uint8Array` 会在任何 GridFS 写入之前完成哈希。不一致时抛出 `GRIDFS_INTEGRITY_MISMATCH`，并且不会留下可见文件。 -- 带 `sha256` 的 `Readable` 会先完整缓冲再哈希，然后才写入。大文件应改用可恢复上传。 +- 带 `sha256` 的 `Readable` 会先完整缓冲再哈希，然后才写入。这是为了遵守“校验失败不会留下可见文件”的契约。不带 `sha256` 的 `upload()` 可以边读边写，但整段 payload 仍是一次提交。大文件应使用可恢复上传，用 `appendChunks` 配合 `highWaterMark` 较小的 `Readable`，不要传入数 GB 的 `Buffer`，也不要走会全量缓冲的 `upload({ sha256 })`。 - 带 `sha256` 的 `skip` 返回该摘要最新的可见文件。不带 `sha256` 的 `skip` 返回同名最新可见文件。`duplicate` 总是再创建一份。 - 通过 `metadata` 对象传入的 `metadata.sha256` 是调用方数据，不会被验证。`upload()` 如需校验字节，必须使用顶层 `sha256` 选项。 - `uploadStream` 的顶层 `sha256` 只写入 metadata，不校验字节，也不接受 `ifExists`。 @@ -75,13 +75,13 @@ const info = await bucket.finalizeResumableUpload(session.uploadId); await bucket.abortResumableUpload(session.uploadId); ``` -`beginResumableUpload` 必须提供 `length` 和 `sha256`。`files` 或 `uploads` 中已经存在该 id 时抛出 `GRIDFS_FILE_EXISTS`。 +`beginResumableUpload` 必须提供 `length` 和 `sha256`。`files` 或 `uploads` 已存在该 id 时抛出 `GRIDFS_FILE_EXISTS`。插入会以 `initializing` 状态独占该 id，且没有过期时间。再次检查 `files` 并清理旧分片后，会话才变为 `pending`。初始化期间，针对同一 id 的其他写入会被阻止。若复查时文件已可见，会保留其分片并抛出 `GRIDFS_FILE_EXISTS`。初始化中断后，运维人员可在确认写入进程已停止后执行恢复。 分片长度必须符合会话： - 除最后一片外，每片长度都是 `chunkSizeBytes`。 - 最后一片长度是 `length % chunkSizeBytes`；能够整除时仍是 `chunkSizeBytes`。 -- 无法按这些长度切开的 payload 抛出 `INVALID_ARGUMENT`。整段 payload 会在插入任何分片之前完成检查。 +- 无法按这些长度切开的 payload 抛出 `INVALID_ARGUMENT`。对 `Buffer` / `Uint8Array`，整段 payload 会在插入任何分片之前完成规划与检查。对 `Readable`，完整分片会随读随写；末尾不足或超出的残余会在消费过程中被拒绝，此前已经写入的完整前缀分片会保留。失败的 append 会抛错，不会返回 `nextChunk`。应从上一次已成功确认的位置继续；若已写入前缀已前进，以 `GRIDFS_RESUME_MISMATCH.expectedNextChunk` 为准。正常读到 EOF 后，若输入流在自动销毁期间报错，追加仍会失败。内存占用受当前流缓冲、未凑齐的分片和固定大小的分片索引批次限制。大文件应使用较小的 `highWaterMark`。 `appendChunks` 以 `fromChunk` 定位： @@ -92,9 +92,11 @@ await bucket.abortResumableUpload(session.uploadId); - 只要 payload 含有新分片，`fromChunk` 就必须等于 `nextChunk`。从更早的序号开始会抛出 `GRIDFS_RESUME_MISMATCH`，并且不会写入。 - `{ files_id, n }` 唯一索引冲突只有在已存字节相同的时候才算成功。 -`finalizeResumableUpload` 要求分片从 0 连续、总长度等于会话长度，且 sha256 等于会话摘要。然后插入 files 文档、写入 `metadata.sha256`，并删除会话。缺少的分片序号抛出 `GRIDFS_UPLOAD_INCOMPLETE`，`details[0].missingChunks` 列出缺口。摘要不符或分片长度不对抛出 `GRIDFS_INTEGRITY_MISMATCH`。这些失败都不会插入 files 文档。 +`finalizeResumableUpload` 要求从 0 开始的每个预期序号恰好对应一个分片，总长度等于会话长度，且 sha256 等于会话摘要。然后插入 files 文档、写入 `metadata.sha256`，并删除会话。缺少分片时抛出 `GRIDFS_UPLOAD_INCOMPLETE`。`details[0].missingChunks` 最多列出前 1,024 个缺失序号，截断时 `details[0].missingChunkCount` 给出缺失总数。序号无效、重复或越界，以及摘要或分片长度不符，均抛出 `GRIDFS_INTEGRITY_MISMATCH`。这些失败都不会插入 files 文档。 -会话仍在时，重试 `finalizeResumableUpload` 是幂等的：若同 id 的文件已可见，且其 `metadata.sha256` 与长度都和会话一致，调用直接返回该文件并删除会话。sha256 或长度不同的文件仍抛 `GRIDFS_FILE_EXISTS`。会话已不存在时（上一次 finalize 已完整成功、已 abort 或 TTL 过期），没有可比对的摘要，重试抛 `GRIDFS_FILE_NOT_FOUND`，可用 `getFile(uploadId)` 确认结果。 +会话仍在时，重试 `finalizeResumableUpload` 是幂等的：若同 id 的文件已可见，库会重新哈希**实际存储的 chunks**，只有摘要、文件长度和分片大小都与会话一致时才返回该文件并删除会话。sha256 或长度不一致时抛出 `GRIDFS_FILE_EXISTS`。分片大小不一致时抛出 `GRIDFS_INTEGRITY_MISMATCH`，并保留已有的 `finalizing` 会话。重试期间若同一 ID 已由另一次上传接管，则返回 `GRIDFS_UPLOAD_BUSY`。会话已不存在时（上一次 finalize 已完整成功、已 abort 或 TTL 过期），没有可比对的摘要，重试抛 `GRIDFS_FILE_NOT_FOUND`。调用方必须自己持久化 `uploadId`、`length` 和 `sha256`。会话消失后，先用 `getFile(uploadId)` 看文件是否可见，再对**实际字节**重新哈希（例如 `download(uploadId)` 或遍历 `.chunks`），与自己保存的长度和摘要比对。不要只信任 `metadata.sha256`：`uploadStream()` 和调用方写入的 `metadata.sha256` 可以是未校验的声明。本包不为此再提供新的公开校验 API。 + +`finalizeResumableUpload` 在已经占有 `finalizing` 之后，若因 `GRIDFS_FILE_EXISTS` 或 `GRIDFS_INTEGRITY_MISMATCH` 失败，会把会话 CAS 回 `pending`，以便通过公开 API 执行 `abortResumableUpload`（然后再删除冲突的可见文件）。files 插入过程中含义不明的驱动错误仍会保留 `finalizing`，需显式恢复。 `abortResumableUpload` 先原子取得 pending 会话的所有权；append 或 finalize 仍在执行时返回 `GRIDFS_UPLOAD_BUSY`，绝不删除其 chunks。取得所有权后删除会话；如果还没有可见的 files 文档，同时删除该上传的 chunks。会话不存在或已经中止时直接成功，并且仍会删除该 id 下没有可见文件的残留 chunks。对已经可见的文件调用 abort 不会删除该文件的 chunks。 @@ -136,7 +138,7 @@ GridFS 写入的类别是 `gridfs`。命名空间是 `.files`，匹配 |--------|------| | `GRIDFS_FILE_NOT_FOUND` | 下载、删除、重命名或可恢复会话的目标不存在 | | `GRIDFS_FILE_EXISTS` | `ifExists: 'error'` 发现冲突，或指定 id 已经存在 | -| `GRIDFS_INTEGRITY_MISMATCH` | 字节与 `sha256` 不符，或 finalize 时分片长度不对 | +| `GRIDFS_INTEGRITY_MISMATCH` | 字节与 `sha256` 不符，或 finalize 时发现无效的分片序号或长度 | | `GRIDFS_UPLOAD_INCOMPLETE` | finalize 发现缺口。`details[0].missingChunks` 列出缺少的序号 | | `GRIDFS_RESUME_MISMATCH` | `fromChunk` 超前于连续前缀，或重放的分片内容不同 | | `GRIDFS_UPLOAD_BUSY` | 其他 append/finalize/abort/upload/delete 操作正在占用该 ID，必须等待完成或在确认旧进程停止后显式恢复 | diff --git a/docs/zh/support-matrix.md b/docs/zh/support-matrix.md index f74886c..ccc1cca 100644 --- a/docs/zh/support-matrix.md +++ b/docs/zh/support-matrix.md @@ -18,7 +18,7 @@ | MongoDB Driver | 6.x / 7.x | `npm run test:server-matrix` | | MongoDB Server | 7.x / 8.x | `mongodb-memory-server` 7.0.37 / 8.0.26 single + replica set 严格矩阵 | -GridFS 是 `.files` 与 `.chunks` 上的客户端约定。它的集成测试跑在 `mongodb-memory-server` 7.0.37 上。这不会把 MongoDB 4.x，或 7.x / 8.x 以外的服务器，写进本矩阵。 +GridFS 是 `.files` 与 `.chunks` 上的客户端约定。它的集成测试跑在当前 GridFS safety / release-gate 任务中的 `mongodb-memory-server` 7.0.37 和 8.0.26 上。这不会把 MongoDB 4.x，或 7.x / 8.x 以外的服务器，写进本矩阵。本检出内针对 Atlas、atlas-local 或 mongot 的 Search / Vector Search 仍为 UNVERIFIED；memory-server 矩阵不能外推到这些部署。 ## 默认验证方式 diff --git a/scripts/maintenance/gridfs-inspect-orphans.cjs b/scripts/maintenance/gridfs-inspect-orphans.cjs new file mode 100644 index 0000000..bb974b5 --- /dev/null +++ b/scripts/maintenance/gridfs-inspect-orphans.cjs @@ -0,0 +1,200 @@ +'use strict'; + +/** + * Operator-only, inspect-only reporter for ObjectId files_id groups. + * Walks .chunks via the existing { files_id, n } index in bounded batches. + * Each files_id is classified only after a unified re-check of files and uploads: + * - visible_file: a files document exists + * - owned: no files document, but an uploads owner/session exists + * - candidate_orphan: no files document and no uploads owner at re-check time + * + * Findings are streamed (onFinding / CLI NDJSON) and capped by maxFindings. + * Use resumeAfter.filesId (exclusive) to continue a truncated scan. Counters + * are accumulated incrementally for the current invocation only. This tool + * never deletes, never applies, and never uses document age as a deletion + * criterion. It is not part of the public GridFSBucket API. + * + * Usage: + * MONSQLIZE_RECOVERY_MONGODB_URI=... node scripts/maintenance/gridfs-inspect-orphans.cjs \ + * --db=my_database --bucket=fs [--page-size=200] [--max-findings=1000] \ + * [--after-files-id=0123456789abcdef01234567] + */ +const { MongoClient, ObjectId } = require('mongodb'); + +const DEFAULT_MAX_FINDINGS = 1000; +const MAX_FINDINGS_CAP = 100000; + +function fail(message) { + const error = new Error(message); + error.code = 'GRIDFS_INSPECT_PRECONDITION_FAILED'; + throw error; +} + +async function classifyFilesId(db, bucket, group) { + const files = db.collection(bucket + '.files'); + const uploads = db.collection(bucket + '.uploads'); + const fileFilter = { _id: group.id }; + const fileProjection = { projection: { _id: 1 }, readPreference: 'primary' }; + let file = await files.findOne(fileFilter, fileProjection); + const owner = await uploads.findOne( + { _id: group.id }, + { projection: { _id: 1, kind: 1, status: 1 }, readPreference: 'primary' }, + ); + // A successful upload publishes the file before it releases its owner. + if (!file && !owner) file = await files.findOne(fileFilter, fileProjection); + let classification; + if (file) classification = 'visible_file'; + else if (owner) classification = 'owned'; + else classification = 'candidate_orphan'; + const finding = { + filesId: group.id.toHexString(), + chunkCount: group.count, + minN: group.minN, + maxN: group.maxN, + classification, + }; + if (owner && !file) { + finding.ownerKind = owner.kind ?? null; + finding.ownerStatus = owner.status ?? null; + } + return finding; +} + +function parseAfter(after) { + if (after == null) return null; + const raw = after.filesId ?? after.files_id; + if (raw instanceof ObjectId) return raw; + if (typeof raw !== 'string' || !ObjectId.isValid(raw)) fail('after.filesId must be an ObjectId'); + return new ObjectId(raw); +} + +async function inspectGridFSOrphans(input) { + const { db, bucket, pageSize = 200, onFinding, collectFindings = true } = input; + if (!db || !bucket || typeof bucket !== 'string') fail('Invalid database or bucket'); + if (!Number.isInteger(pageSize) || pageSize < 1 || pageSize > 10000) { + fail('pageSize must be an integer between 1 and 10000'); + } + const maxFindings = input.maxFindings === undefined ? DEFAULT_MAX_FINDINGS : input.maxFindings; + if (!Number.isInteger(maxFindings) || maxFindings < 1 || maxFindings > MAX_FINDINGS_CAP) { + fail('maxFindings must be an integer between 1 and ' + MAX_FINDINGS_CAP); + } + if (input.apply === true) fail('This inspector has no apply mode and never deletes'); + + const afterId = parseAfter(input.after || input.resumeAfter); + const chunks = db.collection(bucket + '.chunks'); + const findings = collectFindings ? [] : undefined; + let current = null; + let rows = 0; + let filesIdsScanned = 0; + let visibleFiles = 0; + let owned = 0; + let candidateOrphans = 0; + let truncated = false; + let resumeAfter = null; + + const flush = async (group) => { + const finding = await classifyFilesId(db, bucket, group); + filesIdsScanned += 1; + if (finding.classification === 'visible_file') visibleFiles += 1; + else if (finding.classification === 'owned') owned += 1; + else candidateOrphans += 1; + if (typeof onFinding === 'function') onFinding(finding); + if (findings) findings.push(finding); + return finding.filesId; + }; + + const cursor = chunks.find({ files_id: { $type: 'objectId', ...(afterId ? { $gt: afterId } : {}) } }, + { readPreference: 'primary' }) + .sort({ files_id: 1, n: 1 }) + .project({ _id: 0, files_id: 1, n: 1 }) + .batchSize(pageSize); + for await (const doc of cursor) { + rows += 1; + // The query type predicate also matches arrays containing ObjectIds. + if (!(doc.files_id instanceof ObjectId)) continue; + if (!current || !current.id.equals(doc.files_id)) { + if (current) { + const filesId = await flush(current); + if (filesIdsScanned === maxFindings) { + truncated = true; + resumeAfter = { filesId }; + break; + } + } + current = { id: doc.files_id, minN: doc.n, maxN: doc.n, count: 1 }; + } else { + current.count += 1; + if (doc.n < current.minN) current.minN = doc.n; + if (doc.n > current.maxN) current.maxN = doc.n; + } + } + if (current && !truncated) await flush(current); + + return { + bucket, + pageSize, + maxFindings, + pages: Math.max(1, Math.ceil(rows / pageSize)), + filesIdsScanned, + visibleFiles, + owned, + candidateOrphans, + truncated, + resumeAfter, + ...(findings ? { findings } : {}), + }; +} + +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); + if (args.includes('--apply')) fail('This inspector has no apply mode and never deletes'); + const uri = process.env.MONSQLIZE_RECOVERY_MONGODB_URI || process.env.MONSQLIZE_MAINTENANCE_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'); + if (!dbName || !bucket) fail('--db and --bucket are required'); + const pageSizeArg = readArg(args, 'page-size'); + const pageSize = pageSizeArg === undefined ? 200 : Number(pageSizeArg); + if (pageSizeArg !== undefined && (!Number.isInteger(pageSize) || pageSize < 1)) { + fail('--page-size must be a positive integer'); + } + const maxArg = readArg(args, 'max-findings'); + const maxFindings = maxArg === undefined ? DEFAULT_MAX_FINDINGS : Number(maxArg); + if (maxArg !== undefined && (!Number.isInteger(maxFindings) || maxFindings < 1)) { + fail('--max-findings must be a positive integer'); + } + const afterFilesId = readArg(args, 'after-files-id'); + const after = afterFilesId === undefined ? undefined : { filesId: parseAfter({ filesId: afterFilesId }) }; + const client = new MongoClient(uri); + await client.connect(); + try { + const report = await inspectGridFSOrphans({ + db: client.db(dbName), + bucket, + pageSize, + maxFindings, + after, + collectFindings: false, + onFinding: (finding) => { + process.stdout.write(JSON.stringify({ type: 'finding', ...finding }) + '\n'); + }, + }); + process.stdout.write(JSON.stringify({ type: 'summary', ...report }) + '\n'); + } finally { + await client.close(); + } +} + +module.exports = { inspectGridFSOrphans, DEFAULT_MAX_FINDINGS }; +if (require.main === module) { + main().catch((error) => { + process.stderr.write(String(error.code || 'INSPECT_ERROR') + ': ' + error.message + '\n'); + process.exitCode = 1; + }); +} diff --git a/scripts/maintenance/gridfs-recover.cjs b/scripts/maintenance/gridfs-recover.cjs index d80f669..70a7a06 100644 --- a/scripts/maintenance/gridfs-recover.cjs +++ b/scripts/maintenance/gridfs-recover.cjs @@ -13,6 +13,16 @@ const { randomUUID, createHash } = require('node:crypto'); const { MongoClient, ObjectId } = require('mongodb'); const TTL_MS = 604800000; +const RECOVERY_ORIGIN_KINDS = new Map([ + ['uploading', 'direct'], + ['deleting', 'delete'], + ['renaming', 'rename'], + ['cleaning', 'cleanup'], + ['initializing', 'resumable'], + ['activeAppend', 'resumable'], + ['finalizing', 'resumable'], + ['aborting', 'resumable'], +]); function fail(message) { const error = new Error(message); @@ -29,7 +39,7 @@ function toBuffer(data) { async function verifyVisibleFile(db, bucket, id, session, owner) { const files = db.collection(bucket + '.files'); - const doc = await files.findOne({ _id: id }); + const doc = await files.findOne({ _id: id }, { readPreference: 'primary' }); if (!doc) return null; if (!Number.isSafeInteger(doc.length) || doc.length < 0 || !Number.isSafeInteger(doc.chunkSize) || doc.chunkSize < 1) { @@ -40,7 +50,7 @@ async function verifyVisibleFile(db, bucket, id, session, owner) { let n = 0; let length = 0; for await (const chunk of db.collection(bucket + '.chunks') - .find({ files_id: id }).sort({ n: 1 })) { + .find({ files_id: id }, { readPreference: 'primary' }).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; @@ -110,6 +120,7 @@ function classify(doc) { 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 === 'resumable' && doc.status === 'initializing') state = 'initializing'; else if ((!doc.kind || doc.kind === 'resumable') && doc.status === 'pending' && doc.activeAppend) state = 'activeAppend'; else if ((!doc.kind || doc.kind === 'resumable') && @@ -117,13 +128,11 @@ function classify(doc) { } // 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)) { + const requiredKind = RECOVERY_ORIGIN_KINDS.get(state); + if (!requiredKind || + (doc.kind !== requiredKind && !(requiredKind === 'resumable' && !doc.kind)) || + (state === 'activeAppend' && !doc.activeAppend?.token) || + (state === 'initializing' && (doc.kind !== 'resumable' || !doc.ownerToken))) { fail('Inconsistent or unsupported owner origin; preserve ownership'); } return state; @@ -137,7 +146,7 @@ async function contiguousChunkPrefix(db, bucket, id, session) { 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 })) { + .find({ files_id: id }, { readPreference: 'primary' }).sort({ n: 1 })) { if (doc.n !== next || next >= count) { fail('Interrupted append contains a gap or extra chunk; preserve ownership'); } @@ -157,7 +166,7 @@ async function recoverGridFSOwnership(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 }); + const doc = await uploads.findOne({ _id: id }, { readPreference: 'primary' }); if (!doc) return { outcome: 'no_owner', modified: false }; if (!(doc.updatedAt instanceof Date) || doc.status !== expectedStatus || @@ -167,7 +176,8 @@ async function recoverGridFSOwnership(input) { 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 fileProjection = { projection: { _id: 1 }, readPreference: 'primary' }; + const visible = await db.collection(bucket + '.files').findOne({ _id: id }, fileProjection); const plan = { bucket, uploadId: id.toHexString(), fromStatus: sourceStatus, fileVisible: !!visible, @@ -176,6 +186,7 @@ async function recoverGridFSOwnership(input) { deleting: 'complete-file-and-chunk-deletion', renaming: 'release-after-inspecting-rename-outcome', cleaning: 'remove-invisible-orphan-chunks', + initializing: 'remove-invisible-orphan-chunks-and-release-initialization', 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', @@ -243,6 +254,13 @@ async function recoverGridFSOwnership(input) { try { const chunks = db.collection(bucket + '.chunks'); const files = db.collection(bucket + '.files'); + // Dry-run may use the pre-claim snapshot. Apply must re-read visibility + // after the exclusive claim and fail closed if it changed. + const visibleAfter = await files.findOne({ _id: id }, fileProjection); + const visibleNow = !!visibleAfter; + if (visibleNow !== !!visible) { + fail('File visibility changed after recovery claim; preserve recovering ownership'); + } if (sourceStatus === 'activeAppend') { const nextChunk = await contiguousChunkPrefix(db, bucket, id, doc); const changed = await uploads.updateOne(held, { @@ -251,11 +269,11 @@ async function recoverGridFSOwnership(input) { $unset: { activeAppend: '', recovery: '' }, }); if (changed.matchedCount !== 1) fail('Lost recovery claim while restoring append'); - } else if (sourceStatus === 'finalizing' && !visible) { + } else if (sourceStatus === 'finalizing' && !visibleNow) { const changed = await uploads.updateOne(held, { $set: { status: 'pending', updatedAt: new Date(), expiresAt: new Date(Date.now() + TTL_MS) }, - $unset: { recovery: '' }, + $unset: { recovery: '', ownerToken: '' }, }); if (changed.matchedCount !== 1) fail('Lost recovery claim while restoring finalize'); } else { @@ -263,16 +281,18 @@ async function recoverGridFSOwnership(input) { // unless this is an explicitly quiesced, retained "deleting" owner. if (sourceStatus === 'finalizing') { verification = await verifyVisibleFile(db, bucket, id, doc, doc); - } else if ((sourceStatus === 'uploading' || sourceStatus === 'renaming') && visible) { + if (!verification) fail('Visible file could not be verified after claim; preserve recovering ownership'); + } else if ((sourceStatus === 'uploading' || sourceStatus === 'renaming') && visibleNow) { verification = await verifyVisibleFile(db, bucket, id, null, doc); - } else if (sourceStatus === 'renaming' && !visible) { + if (!verification) fail('Visible file could not be verified after claim; preserve recovering ownership'); + } else if (sourceStatus === 'renaming' && !visibleNow) { 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) { + } else if ((sourceStatus === 'uploading' || sourceStatus === 'cleaning' || sourceStatus === 'initializing' || + sourceStatus === 'aborting') && !visibleNow) { await chunks.deleteMany({ files_id: id }); } const removed = await uploads.deleteOne(held); diff --git a/scripts/validation/run-memory-server-matrix.cjs b/scripts/validation/run-memory-server-matrix.cjs index f6a7071..09396e4 100644 --- a/scripts/validation/run-memory-server-matrix.cjs +++ b/scripts/validation/run-memory-server-matrix.cjs @@ -49,6 +49,8 @@ const integrationSourceSuites = [ '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/gridfs/gridfs-remaining-issues.test.js', + 'test/integration/gridfs/gridfs-inspect-orphans.test.js', ]; const integrationSuites = integrationSourceSuites.map((suite) => path.join(testDistRoot, suite)); const compatibilitySuites = [ diff --git a/src/capabilities/gridfs/errors.ts b/src/capabilities/gridfs/errors.ts index 2d5801e..3042165 100644 --- a/src/capabilities/gridfs/errors.ts +++ b/src/capabilities/gridfs/errors.ts @@ -12,6 +12,7 @@ export interface GridFSErrorDetails { expected?: unknown; actual?: unknown; missingChunks?: number[]; + missingChunkCount?: number; fromChunk?: number; expectedNextChunk?: number; start?: number; @@ -92,11 +93,12 @@ export function uploadBusy(operation: string, ctx: Pick missingChunks.length ? { missingChunkCount: missingCount } : {}) }, ); } diff --git a/src/capabilities/gridfs/operations.ts b/src/capabilities/gridfs/operations.ts index f425cd8..40ffa66 100644 --- a/src/capabilities/gridfs/operations.ts +++ b/src/capabilities/gridfs/operations.ts @@ -68,7 +68,7 @@ export async function uploadFile( } await ctx.indexes.ensure(); - if (options.id && await getFile(ctx, options.id)) { + if (options.id && await getFile(ctx, options.id, 'primary')) { if (source) releaseReadable(source); throw fileExists('upload', ctx, { id: options.id, filename, sha256: options.sha256 }); } @@ -97,14 +97,14 @@ export async function uploadFile( throw error; } try { - if (await getFile(ctx, id)) { + if (await getFile(ctx, id, 'primary')) { if (source) releaseReadable(source); throw fileExists('upload', ctx, { id, filename, sha256: options.sha256 }); } const writeOptions = { chunkSizeBytes, metadata, contentType: options.contentType }; if (payload) await writeBuffer(ctx, id, filename, payload, writeOptions); else if (source) await writeReadable(ctx, id, filename, source, writeOptions); - const info = await getFile(ctx, id); + const info = await getFile(ctx, id, 'primary'); if (!info) throw fileNotFound('upload', ctx, { id, filename }); // A successful upload is committed even if removing its reservation fails. // Keep the durable reservation for explicit recovery rather than deleting the file. @@ -150,7 +150,7 @@ export function openUploadStream(ctx: GridFSContext, filename: string, opts?: Gr ...(typeof metadata?.sha256 === 'string' ? { claimedSha256: metadata.sha256 } : {}), }); // Check again after the ID is reserved, in case an older unreserved file exists. - if (await getFile(ctx, id)) { + if (await getFile(ctx, id, 'primary')) { throw fileExists('uploadStream', ctx, { id, filename, sha256: options.sha256 }); } target = ctx.driver.openUploadStream(id, filename, { @@ -160,7 +160,7 @@ export function openUploadStream(ctx: GridFSContext, filename: string, opts?: Gr }); const { pipeline } = await import('node:stream/promises'); await pipeline(input, target); - const info = await getFile(ctx, id); + const info = await getFile(ctx, id, 'primary'); if (!info) throw fileNotFound('uploadStream', ctx, { id, filename }); try { await releaseOwner(ctx, owner, 'uploadStream'); @@ -224,10 +224,10 @@ export function openDownloadStream(ctx: GridFSContext, idOrName: GridFSIdOrName, return output; } -export async function getFile(ctx: GridFSContext, id: ObjectId): Promise { +export async function getFile(ctx: GridFSContext, id: ObjectId, readPreference?: 'primary'): Promise { assertObjectId(id, 'getFile', 'id'); try { - const doc = await ctx.db.collection(filesCollectionName(ctx.bucketName)).findOne({ _id: id }); + const doc = await ctx.db.collection(filesCollectionName(ctx.bucketName)).findOne({ _id: id }, { readPreference }); return doc ? toFileInfo(doc as Record) : null; } catch (error) { throw mapDriverError(error, 'getFile', ctx, { id }); @@ -254,7 +254,7 @@ export async function deleteFile(ctx: GridFSContext, id: ObjectId, opts?: GridFS const owner = await claimOwner(ctx, id, 'delete', 'delete'); let driverInvoked = false; try { - const existing = await getFile(ctx, id); + const existing = await getFile(ctx, id, 'primary'); if (!existing) { await releaseOwner(ctx, owner, 'delete'); if (opts?.ignoreMissing) return; @@ -284,16 +284,26 @@ export async function renameFile(ctx: GridFSContext, id: ObjectId, filename: str assertFilename(filename, 'rename'); const owner = await claimOwner(ctx, id, 'rename', 'rename'); try { - const existing = await getFile(ctx, id); + const existing = await getFile(ctx, id, 'primary'); if (!existing) throw fileNotFound('rename', ctx, { id, filename }); try { await ctx.driver.rename(id, filename); } catch (error) { throw mapDriverError(error, 'rename', ctx, { id, filename }); } - } finally { - // Rename never removes chunks and is safe to unlock after driver settles. - await releaseOwner(ctx, owner, 'rename'); + // A successful rename is committed even if removing its reservation fails. + try { + await releaseOwner(ctx, owner, 'rename'); + } catch (error) { + ctx.logger?.warn?.('[gridfs] rename committed but reservation cleanup failed', { bucket: ctx.bucketName, id: String(id), error }); + } + } catch (error) { + try { + await releaseOwner(ctx, owner, 'rename'); + } catch (releaseError) { + ctx.logger?.warn?.('[gridfs] rename reservation retained after cleanup failure', { bucket: ctx.bucketName, id: String(id), error: releaseError }); + } + throw error; } } @@ -323,7 +333,7 @@ async function findConflict(ctx: GridFSContext, filename: string, sha256: string const filter = sha256 !== undefined ? { 'metadata.sha256': sha256 } : { filename }; let docs; try { - docs = await files.find(filter).sort({ uploadDate: -1 }).limit(1).toArray(); + docs = await files.find(filter, { readPreference: 'primary' }).sort({ uploadDate: -1 }).limit(1).toArray(); } catch (error) { throw mapDriverError(error, 'upload', ctx, { filename }); } @@ -404,6 +414,6 @@ async function cleanupFailedWrite(ctx: GridFSContext, id: ObjectId, stream: Grid if (typeof stream.abort === 'function') await stream.abort(); } catch { /* A finished driver stream refuses abort; delete the visible file below. */ } try { - if (await getFile(ctx, id)) await ctx.driver.delete(id); + if (await getFile(ctx, id, 'primary')) await ctx.driver.delete(id); } catch { /* The caller still receives the original write error. */ } } diff --git a/src/capabilities/gridfs/ownership.ts b/src/capabilities/gridfs/ownership.ts index d26d531..e364a31 100644 --- a/src/capabilities/gridfs/ownership.ts +++ b/src/capabilities/gridfs/ownership.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto'; import type { ObjectId } from 'mongodb'; import type { GridFSContext } from './types'; -import { fileExists, isDuplicateKey, mapDriverError, uploadBusy } from './errors'; +import { isDuplicateKey, mapDriverError, uploadBusy } from './errors'; import { chunksCollectionName, filesCollectionName, uploadsCollectionName } from './normalize'; /** @@ -81,7 +81,7 @@ export async function releaseOwner(ctx: GridFSContext, owner: GridFSOwner, opera export async function cleanupFailedDirect(ctx: GridFSContext, owner: GridFSOwner): Promise { try { const visible = await ctx.db.collection(filesCollectionName(ctx.bucketName)) - .findOne({ _id: owner.id }, { projection: { _id: 1 } }); + .findOne({ _id: owner.id }, { projection: { _id: 1 }, readPreference: 'primary' }); if (!visible) { await ctx.db.collection(chunksCollectionName(ctx.bucketName)) .deleteMany({ files_id: owner.id }); @@ -103,7 +103,7 @@ export async function cleanupOrphanById(ctx: GridFSContext, id: ObjectId): Promi let safeToRelease = false; try { const visible = await ctx.db.collection(filesCollectionName(ctx.bucketName)) - .findOne({ _id: id }, { projection: { _id: 1 } }); + .findOne({ _id: id }, { projection: { _id: 1 }, readPreference: 'primary' }); if (!visible) await ctx.db.collection(chunksCollectionName(ctx.bucketName)).deleteMany({ files_id: id }); safeToRelease = true; } catch (error) { diff --git a/src/capabilities/gridfs/resumable.ts b/src/capabilities/gridfs/resumable.ts index bb3245a..a15694e 100644 --- a/src/capabilities/gridfs/resumable.ts +++ b/src/capabilities/gridfs/resumable.ts @@ -2,7 +2,7 @@ import { createHash, randomUUID } from 'node:crypto'; import type { Readable } from 'node:stream'; import { ObjectId } from 'mongodb'; import type { AppendChunksResult, BeginResumableUploadOptions, BeginResumableUploadResult, GridFSContext, GridFSFileInfo, GridFSUploadSession } from './types'; -import { fileExists, fileNotFound, integrityMismatch, invalidArgument, isDuplicateKey, mapDriverError, resumeMismatch, uploadBusy, uploadIncomplete } from './errors'; +import { fileExists, fileNotFound, integrityMismatch, invalidArgument, isDuplicateKey, isGridFSDomainError, mapDriverError, resumeMismatch, uploadBusy, uploadIncomplete } from './errors'; import { cleanupOrphanById } from './ownership'; import { binaryToBuffer, chunksCollectionName, filesCollectionName, sameBytes, toFileInfo, uploadsCollectionName } from './normalize'; import { @@ -20,7 +20,9 @@ import { toNodeBuffer, withContentType, } from './validate'; -import { readReadable } from './bytes'; +import { consumeReadableChunks, watchReadable } from './stream-chunks'; + +const MAX_REPORTED_MISSING_CHUNKS = 1024; export async function beginResumableUpload(ctx: GridFSContext, filename: string, opts: BeginResumableUploadOptions): Promise { assertFilename(filename, 'beginResumableUpload'); @@ -39,37 +41,49 @@ export async function beginResumableUpload(ctx: GridFSContext, filename: string, const uploadId = opts.id ?? new ObjectId(); const files = ctx.db.collection(filesCollectionName(ctx.bucketName)); const uploads = ctx.db.collection(uploadsCollectionName(ctx.bucketName)); - if (await files.findOne({ _id: uploadId }, { projection: { _id: 1 } })) { + if (await files.findOne({ _id: uploadId }, { projection: { _id: 1 }, readPreference: 'primary' })) { throw fileExists('beginResumableUpload', ctx, { id: uploadId, filename, sha256: opts.sha256 }); } - if (await uploads.findOne({ _id: uploadId }, { projection: { _id: 1 } })) { + if (await uploads.findOne({ _id: uploadId }, { projection: { _id: 1 }, readPreference: 'primary' })) { throw fileExists('beginResumableUpload', ctx, { id: uploadId, filename, sha256: opts.sha256 }); } const now = new Date(); + const ownerToken = randomUUID(); + const initializing = { _id: uploadId, kind: 'resumable', status: 'initializing', ownerToken }; try { await uploads.insertOne({ - _id: uploadId, + ...initializing, + sessionToken: ownerToken, filename, length: opts.length, sha256: opts.sha256, chunkSizeBytes, ...(metadata ? { metadata } : {}), ...(opts.contentType !== undefined ? { contentType: opts.contentType } : {}), - kind: 'resumable', - status: 'pending', nextChunk: 0, createdAt: now, updatedAt: now, - expiresAt: new Date(now.getTime() + 604800 * 1000), }); } catch (error) { throw mapDriverError(error, 'beginResumableUpload', ctx, { uploadId, filename }); } - // Chunks can outlive an expired session (the TTL only removes the session document). This call now owns - // the id and no file is visible, so leftovers would only break the new upload; remove them. + // Only publish a pending session after its orphan cleanup has settled. + // Initializing has no TTL and excludes append, finalize, abort and ID reuse. try { + if (await files.findOne({ _id: uploadId }, { projection: { _id: 1 }, readPreference: 'primary' })) { + try { + await uploads.deleteOne(initializing); + } catch { /* Best-effort session cleanup; chunks of the visible file stay. */ } + throw fileExists('beginResumableUpload', ctx, { id: uploadId, filename, sha256: opts.sha256 }); + } await ctx.db.collection(chunksCollectionName(ctx.bucketName)).deleteMany({ files_id: uploadId }); + const updatedAt = new Date(); + const ready = await uploads.updateOne(initializing, { + $set: { status: 'pending', updatedAt, expiresAt: new Date(updatedAt.getTime() + 604800 * 1000) }, + $unset: { ownerToken: '' }, + }); + if (!ready.matchedCount) throw uploadBusy('beginResumableUpload', ctx, uploadId); } catch (error) { throw mapDriverError(error, 'beginResumableUpload', ctx, { uploadId, filename }); } @@ -84,8 +98,29 @@ export async function appendChunks( ): Promise { assertObjectId(uploadId, 'appendChunks', 'uploadId'); assertNonNegativeInteger(fromChunk, 'appendChunks', 'fromChunk'); - const payload = await payloadBytes(data); + if (isReadableStream(data)) { + const stream = watchReadable(data); + try { + if (stream.error) throw stream.error; + return await appendToSession(ctx, uploadId, fromChunk, stream); + } finally { + stream.release(); + } + } + if (!Buffer.isBuffer(data) && !(data instanceof Uint8Array)) { + throw invalidArgument('appendChunks', 'data', 'data must be a Buffer, Uint8Array, or Readable'); + } + const payload = toNodeBuffer(data); if (payload.length === 0) throw badChunkSize(); + return appendToSession(ctx, uploadId, fromChunk, payload); +} + +async function appendToSession( + ctx: GridFSContext, + uploadId: ObjectId, + fromChunk: number, + data: Buffer | ReturnType, +): Promise { const uploads = ctx.db.collection(uploadsCollectionName(ctx.bucketName)); const token = randomUUID(); // A durable single-append claim excludes finalize, abort, and concurrent @@ -101,24 +136,40 @@ export async function appendChunks( throw mapDriverError(error, 'appendChunks', ctx, { uploadId }); } if (!claimed) { - const existing = await uploads.findOne({ _id: uploadId }); + const existing = await uploads.findOne({ _id: uploadId }, { readPreference: 'primary' }); if (!existing) throw fileNotFound('appendChunks', ctx, { id: uploadId }); throw uploadBusy('appendChunks', ctx, uploadId); } try { + if (!Buffer.isBuffer(data) && data.error) throw data.error; const session = await requireSession(ctx, uploadId, 'appendChunks'); const nextChunk = await contiguousNext(ctx, uploadId, session.nextChunk); if (fromChunk > nextChunk) { throw resumeMismatch('appendChunks', ctx.bucketName, { uploadId, fromChunk, expectedNextChunk: nextChunk }); } - const slices = planChunkSlices(payload, fromChunk, session.length, session.chunkSizeBytes); - const lastIndex = slices[slices.length - 1].index; - if (lastIndex >= nextChunk && fromChunk !== nextChunk) { - throw resumeMismatch('appendChunks', ctx.bucketName, { uploadId, fromChunk, expectedNextChunk: nextChunk }); - } - for (const slice of slices) { - if (slice.index < nextChunk) await assertStoredChunk(ctx, uploadId, slice.index, slice.bytes, nextChunk); - else await writeChunk(ctx, uploadId, slice.index, slice.bytes, nextChunk); + const writeSlice = async (slice: { index: number; bytes: Buffer }): Promise => { + if (slice.index < nextChunk) { + await assertStoredChunk(ctx, uploadId, slice.index, slice.bytes, nextChunk); + return; + } + if (fromChunk !== nextChunk) { + throw resumeMismatch('appendChunks', ctx.bucketName, { uploadId, fromChunk, expectedNextChunk: nextChunk }); + } + await writeChunk(ctx, uploadId, slice.index, slice.bytes, nextChunk); + }; + let bytesAccepted: number; + if (Buffer.isBuffer(data)) { + const slices = planChunkSlices(data, fromChunk, session.length, session.chunkSizeBytes); + const lastIndex = slices[slices.length - 1].index; + if (lastIndex >= nextChunk && fromChunk !== nextChunk) { + throw resumeMismatch('appendChunks', ctx.bucketName, { uploadId, fromChunk, expectedNextChunk: nextChunk }); + } + for (const slice of slices) await writeSlice(slice); + bytesAccepted = data.length; + } else { + const streamed = await consumeReadableChunks(data.source, fromChunk, session.length, + session.chunkSizeBytes, writeSlice, undefined, data); + bytesAccepted = streamed.bytesAccepted; } const advanced = await contiguousNext(ctx, uploadId, nextChunk); const completed = await uploads.updateOne( @@ -130,7 +181,7 @@ export async function appendChunks( }, ); if (!completed.matchedCount) throw uploadBusy('appendChunks', ctx, uploadId); - return { nextChunk: advanced, bytesAccepted: payload.length }; + return { nextChunk: advanced, bytesAccepted }; } catch (error) { // The worker has stopped writing. Releasing only its own token allows // the next append to reconcile partial chunks idempotently. If release @@ -148,7 +199,7 @@ export async function appendChunks( export async function finalizeResumableUpload(ctx: GridFSContext, uploadId: ObjectId): Promise { assertObjectId(uploadId, 'finalizeResumableUpload', 'uploadId'); const uploads = ctx.db.collection(uploadsCollectionName(ctx.bucketName)); - let session = await uploads.findOne({ _id: uploadId }) as unknown as GridFSUploadSession | null; + let session = await uploads.findOne({ _id: uploadId }, { readPreference: 'primary' }) as unknown as GridFSUploadSession | null; if (!session || (session.kind && session.kind !== 'resumable')) { throw fileNotFound('finalizeResumableUpload', ctx, { id: uploadId }); } @@ -160,35 +211,60 @@ export async function finalizeResumableUpload(ctx: GridFSContext, uploadId: Obje throw uploadBusy('finalizeResumableUpload', ctx, uploadId); } if (session.status !== 'pending') throw uploadBusy('finalizeResumableUpload', ctx, uploadId); + const ownerToken = randomUUID(); + const owner = finalizingOwner(uploadId, ownerToken); const transitioned = await uploads.updateOne( - { _id: uploadId, status: 'pending', activeAppend: { $exists: false } }, - { $set: { status: 'finalizing', updatedAt: new Date() }, $unset: { expiresAt: '' } }, + { ...sessionIdentity(session), status: 'pending', activeAppend: { $exists: false } }, + { $set: { status: 'finalizing', ownerToken, updatedAt: new Date() }, $unset: { expiresAt: '' } }, ); if (!transitioned.matchedCount) throw uploadBusy('finalizeResumableUpload', ctx, uploadId); // Only the winner of this CAS may inspect/commit files. Abort will now // fail its pending -> aborting CAS instead of deleting validated chunks. session = await requireSession(ctx, uploadId, 'finalizeResumableUpload'); + if (session.status !== 'finalizing' || session.ownerToken !== ownerToken) { + throw uploadBusy('finalizeResumableUpload', ctx, uploadId); + } const files = ctx.db.collection(filesCollectionName(ctx.bucketName)); - const committed = await committedFile(ctx, uploadId, session); - if (committed) return committed; + + const restorePending = async (): Promise => { + try { + await uploads.updateOne(owner, { + $set: { status: 'pending', updatedAt: new Date(), expiresAt: new Date(Date.now() + 604800 * 1000) }, + $unset: { ownerToken: '' }, + }); + } catch { /* The retained finalizing state must be recovered explicitly. */ } + }; + + try { + const already = await committedFile(ctx, uploadId, session); + if (already) return already; + } catch (error) { + if (isClosedFinalizeFailure(error)) await restorePending(); + throw error; + } try { const expected = chunkCount(session.length, session.chunkSizeBytes); const hash = createHash('sha256'); let total = 0; const missing: number[] = []; - const cursor = ctx.db.collection(chunksCollectionName(ctx.bucketName)).find({ files_id: uploadId }).sort({ n: 1 }); + let missingCount = 0; + const cursor = ctx.db.collection(chunksCollectionName(ctx.bucketName)) + .find({ files_id: uploadId }, { readPreference: 'primary' }).sort({ n: 1 }); let seen = 0; + const recordGap = (end: number): void => { + missingCount += end - seen; + for (let n = seen; n < end && missing.length < MAX_REPORTED_MISSING_CHUNKS; n++) missing.push(n); + seen = end; + }; for await (const doc of cursor) { - const n = typeof doc.n === 'number' ? doc.n : -1; - while (seen < n && seen < expected) { - missing.push(seen); - seen += 1; - } - if (n !== seen || n < 0 || n >= expected) { - if (n >= 0) missing.push(n); - continue; + const n = doc.n; + if (!Number.isSafeInteger(n) || n < seen || n >= expected) { + throw integrityMismatch('finalizeResumableUpload', ctx.bucketName, { + uploadId, expected: `chunk index between ${seen} and ${expected - 1}`, actual: n, + }); } + recordGap(n); const bytes = binaryToBuffer(doc.data, 'finalizeResumableUpload'); const want = chunkLengthAt(n, session.length, session.chunkSizeBytes); if (bytes.length !== want) { @@ -198,11 +274,8 @@ export async function finalizeResumableUpload(ctx: GridFSContext, uploadId: Obje total += bytes.length; seen += 1; } - while (seen < expected) { - missing.push(seen); - seen += 1; - } - if (missing.length > 0) throw uploadIncomplete(ctx.bucketName, uploadId, missing); + recordGap(expected); + if (missingCount > 0) throw uploadIncomplete(ctx.bucketName, uploadId, missing, missingCount); if (total !== session.length) { throw integrityMismatch('finalizeResumableUpload', ctx.bucketName, { uploadId, expected: session.length, actual: total }); } @@ -213,12 +286,7 @@ export async function finalizeResumableUpload(ctx: GridFSContext, uploadId: Obje } catch (error) { // Validation failed before any files insert. Return to pending so the // caller can supply missing/correct chunks; do not erase those chunks. - try { - await uploads.updateOne( - { _id: uploadId, status: 'finalizing' }, - { $set: { status: 'pending', updatedAt: new Date(), expiresAt: new Date(Date.now() + 604800 * 1000) } }, - ); - } catch { /* The retained finalizing state must be recovered explicitly. */ } + await restorePending(); throw error; } @@ -226,37 +294,41 @@ export async function finalizeResumableUpload(ctx: GridFSContext, uploadId: Obje buildMetadata(session.metadata, session.sha256, 'finalizeResumableUpload'), session.contentType, ); + const file = { + _id: uploadId, + length: session.length, + chunkSize: session.chunkSizeBytes, + uploadDate: new Date(), + filename: session.filename, + ...(metadata ? { metadata } : {}), + }; try { - await files.insertOne({ - _id: uploadId, - length: session.length, - chunkSize: session.chunkSizeBytes, - uploadDate: new Date(), - filename: session.filename, - ...(metadata ? { metadata } : {}), - }); + await files.insertOne(file); } catch (error) { if (isDuplicateKey(error)) { - const raced = await committedFile(ctx, uploadId, session); - if (raced) return raced; + try { + const raced = await committedFile(ctx, uploadId, session); + if (raced) return raced; + } catch (conflict) { + if (isClosedFinalizeFailure(conflict)) await restorePending(); + throw conflict; + } + } + const mapped = mapDriverError(error, 'finalizeResumableUpload', ctx, { uploadId, filename: session.filename }); + if (isClosedFinalizeFailure(mapped)) { + await restorePending(); } - // Commit outcome may be ambiguous; hold finalizing ownership rather than + // Ambiguous commit outcomes keep finalizing ownership rather than // allowing an abort or another writer to delete/reuse the same chunks. - throw mapDriverError(error, 'finalizeResumableUpload', ctx, { uploadId, filename: session.filename }); + throw mapped; } try { - await uploads.deleteOne({ _id: uploadId, status: 'finalizing' }); + await uploads.deleteOne(owner); } catch (error) { ctx.logger?.warn?.('[gridfs] finalizeResumableUpload could not delete the upload session', { bucket: ctx.bucketName, uploadId: String(uploadId), error }); } - let info; - try { - info = await files.findOne({ _id: uploadId }); - } catch (error) { - throw mapDriverError(error, 'finalizeResumableUpload', ctx, { uploadId, filename: session.filename }); - } - if (!info) throw fileNotFound('finalizeResumableUpload', ctx, { id: uploadId, filename: session.filename }); - return toFileInfo(info as Record); + // The ID may be reused as soon as the finalize owner is released. + return toFileInfo(file); } export async function abortResumableUpload(ctx: GridFSContext, uploadId: ObjectId): Promise { @@ -264,7 +336,7 @@ export async function abortResumableUpload(ctx: GridFSContext, uploadId: ObjectI const uploads = ctx.db.collection(uploadsCollectionName(ctx.bucketName)); let session: GridFSUploadSession | null; try { - session = await uploads.findOne({ _id: uploadId }) as unknown as GridFSUploadSession | null; + session = await uploads.findOne({ _id: uploadId }, { readPreference: 'primary' }) as unknown as GridFSUploadSession | null; } catch (error) { throw mapDriverError(error, 'abortResumableUpload', ctx, { uploadId }); } @@ -277,8 +349,9 @@ export async function abortResumableUpload(ctx: GridFSContext, uploadId: ObjectI if (session.kind && session.kind !== 'resumable') throw uploadBusy('abortResumableUpload', ctx, uploadId); if (session.status === 'finalized') return; if (session.status !== 'pending') throw uploadBusy('abortResumableUpload', ctx, uploadId); + const identity = sessionIdentity(session); const transitioned = await uploads.updateOne( - { _id: uploadId, status: 'pending', activeAppend: { $exists: false } }, + { ...identity, status: 'pending', activeAppend: { $exists: false } }, { $set: { status: 'aborting', updatedAt: new Date() }, $unset: { expiresAt: '' } }, ); if (!transitioned.matchedCount) throw uploadBusy('abortResumableUpload', ctx, uploadId); @@ -286,33 +359,53 @@ export async function abortResumableUpload(ctx: GridFSContext, uploadId: ObjectI // aborting record that prevents a second writer from claiming this ID. try { await removeInvisibleChunks(ctx, uploadId, 'abortResumableUpload'); - await uploads.deleteOne({ _id: uploadId, status: 'aborting' }); + await uploads.deleteOne({ ...identity, status: 'aborting' }); } catch (error) { throw mapDriverError(error, 'abortResumableUpload', ctx, { uploadId }); } } -/** - * Resolve a finalize retry against an already visible file under `uploadId`. - * Returns null when no file exists. A file whose stored sha256 and length match the session is this upload's own - * earlier commit: delete the leftover session and return it. Any other file is a real conflict (GRIDFS_FILE_EXISTS). - */ +function sessionIdentity(session: GridFSUploadSession) { + if (session.sessionToken !== undefined) return { _id: session._id, sessionToken: session.sessionToken }; + // Older sessions have no token. In particular, they must never match a new upload. + return { _id: session._id, sessionToken: { $exists: false }, createdAt: session.createdAt ?? { $exists: false } }; +} + +function finalizingOwner(uploadId: ObjectId, ownerToken?: string) { + // Legacy finalizing records have no token. Their retries can only clear legacy records. + return { _id: uploadId, status: 'finalizing', ownerToken: ownerToken ?? { $exists: false } }; +} + +/** Verify an already visible file against the session that still owns this ID. */ async function committedFile(ctx: GridFSContext, uploadId: ObjectId, session: GridFSUploadSession): Promise { const target = { uploadId, filename: session.filename }; + const uploads = ctx.db.collection(uploadsCollectionName(ctx.bucketName)); + const owner = finalizingOwner(uploadId, session.ownerToken); try { - const doc = await ctx.db.collection(filesCollectionName(ctx.bucketName)).findOne({ _id: uploadId }); + const doc = await ctx.db.collection(filesCollectionName(ctx.bucketName)) + .findOne({ _id: uploadId }, { readPreference: 'primary' }); if (!doc) return null; + // After capturing the file, verify that the same finalize still owns + // this ID. Identical bytes can belong to a later upload with this ID. + if (!await uploads.findOne(owner, { projection: { _id: 1 }, readPreference: 'primary' })) { + throw uploadBusy('finalizeResumableUpload', ctx, uploadId); + } const info = toFileInfo(doc as Record); if (info.metadata?.sha256 !== session.sha256 || info.length !== session.length) { throw fileExists('finalizeResumableUpload', ctx, { id: uploadId, filename: session.filename, sha256: session.sha256 }); } + if (info.chunkSize !== session.chunkSizeBytes) { + throw integrityMismatch('finalizeResumableUpload', ctx.bucketName, { + uploadId, expected: session.chunkSizeBytes, actual: info.chunkSize, + }); + } // Metadata-only SHA can be forged by uploadStream(). Never treat it as // proof of a previous resumable commit without verifying actual chunks. const hash = createHash('sha256'); let total = 0; let index = 0; for await (const chunk of ctx.db.collection(chunksCollectionName(ctx.bucketName)) - .find({ files_id: uploadId }).sort({ n: 1 })) { + .find({ files_id: uploadId }, { readPreference: 'primary' }).sort({ n: 1 })) { const bytes = binaryToBuffer(chunk.data, 'finalizeResumableUpload'); const expectedSize = chunkLengthAt(index, session.length, session.chunkSizeBytes); if (chunk.n !== index || bytes.length !== expectedSize) { @@ -325,7 +418,8 @@ async function committedFile(ctx: GridFSContext, uploadId: ObjectId, session: Gr if (index !== chunkCount(session.length, session.chunkSizeBytes) || total !== session.length || hash.digest('hex') !== session.sha256) { throw integrityMismatch('finalizeResumableUpload', ctx.bucketName, { uploadId, expected: session.sha256, actual: 'committed chunks do not match' }); } - await ctx.db.collection(uploadsCollectionName(ctx.bucketName)).deleteOne({ _id: uploadId, status: 'finalizing' }); + const released = await uploads.deleteOne(owner); + if (!released.deletedCount) throw uploadBusy('finalizeResumableUpload', ctx, uploadId); return info; } catch (error) { // GridFS domain errors (the conflict above) pass through unchanged. @@ -338,7 +432,7 @@ async function removeInvisibleChunks(ctx: GridFSContext, uploadId: ObjectId, ope try { const visible = await ctx.db.collection(filesCollectionName(ctx.bucketName)).findOne( { _id: uploadId }, - { projection: { _id: 1 } }, + { projection: { _id: 1 }, readPreference: 'primary' }, ); if (!visible) await ctx.db.collection(chunksCollectionName(ctx.bucketName)).deleteMany({ files_id: uploadId }); } catch (error) { @@ -346,10 +440,9 @@ async function removeInvisibleChunks(ctx: GridFSContext, uploadId: ObjectId, ope } } -async function payloadBytes(data: Buffer | Uint8Array | Readable): Promise { - if (Buffer.isBuffer(data) || data instanceof Uint8Array) return toNodeBuffer(data); - if (isReadableStream(data)) return readReadable(data); - throw invalidArgument('appendChunks', 'data', 'data must be a Buffer, Uint8Array, or Readable'); +function isClosedFinalizeFailure(error: unknown): boolean { + return isGridFSDomainError(error) + && (error.code === 'GRIDFS_FILE_EXISTS' || error.code === 'GRIDFS_INTEGRITY_MISMATCH'); } function badChunkSize() { @@ -373,7 +466,7 @@ function planChunkSlices(payload: Buffer, fromChunk: number, length: number, chu } async function requireSession(ctx: GridFSContext, uploadId: ObjectId, operation: string): Promise { - const doc = await ctx.db.collection(uploadsCollectionName(ctx.bucketName)).findOne({ _id: uploadId }); + const doc = await ctx.db.collection(uploadsCollectionName(ctx.bucketName)).findOne({ _id: uploadId }, { readPreference: 'primary' }); if (!doc || (doc.kind && doc.kind !== 'resumable')) throw fileNotFound(operation, ctx, { id: uploadId }); return doc as unknown as GridFSUploadSession; } @@ -384,26 +477,26 @@ async function requireSession(ctx: GridFSContext, uploadId: ObjectId, operation: */ async function contiguousNext(ctx: GridFSContext, uploadId: ObjectId, from: number): Promise { const start = Number.isInteger(from) && from > 0 ? from : 0; - let docs; + let expected = start; try { - docs = await ctx.db.collection(chunksCollectionName(ctx.bucketName)) - .find({ files_id: uploadId, n: { $gte: start } }) + const cursor = ctx.db.collection(chunksCollectionName(ctx.bucketName)) + .find({ files_id: uploadId, n: { $gte: start } }, { readPreference: 'primary' }) .sort({ n: 1 }) - .project({ n: 1 }) - .toArray(); + .project({ _id: 0, n: 1 }) + .batchSize(256); + for await (const doc of cursor) { + if (doc.n !== expected) break; + expected += 1; + } } catch (error) { throw mapDriverError(error, 'appendChunks', ctx, { uploadId }); } - let expected = start; - for (const doc of docs) { - if (doc.n !== expected) break; - expected += 1; - } return expected; } async function assertStoredChunk(ctx: GridFSContext, uploadId: ObjectId, n: number, slice: Buffer, expectedNextChunk: number): Promise { - const existing = await ctx.db.collection(chunksCollectionName(ctx.bucketName)).findOne({ files_id: uploadId, n }); + const existing = await ctx.db.collection(chunksCollectionName(ctx.bucketName)) + .findOne({ files_id: uploadId, n }, { readPreference: 'primary' }); if (!existing || !sameBytes(binaryToBuffer(existing.data, 'appendChunks'), slice)) { throw resumeMismatch('appendChunks', ctx.bucketName, { uploadId, fromChunk: n, expectedNextChunk }); } diff --git a/src/capabilities/gridfs/stream-chunks.ts b/src/capabilities/gridfs/stream-chunks.ts new file mode 100644 index 0000000..cc2a903 --- /dev/null +++ b/src/capabilities/gridfs/stream-chunks.ts @@ -0,0 +1,243 @@ +import type { Readable } from 'node:stream'; +import { invalidArgument } from './errors'; +import { chunkCount, chunkLengthAt } from './validate'; + +const DESTROY_CHECK_MS = 50; + +export interface StreamedChunk { + index: number; + bytes: Buffer; +} + +function badChunkSize(): never { + throw invalidArgument('appendChunks', 'data', 'chunk payload must match the session chunk size'); +} + +function streamClosedError(): Error { + return invalidArgument('appendChunks', 'data', 'readable closed before the upload completed'); +} + +function isPrematureClose(source: Readable, ended: boolean): boolean { + return source.destroyed && !ended && !source.readableEnded; +} + +/** Capture errors before the append claim and keep handling errors from async destruction. */ +export function watchReadable(source: Readable) { + let error: Error | undefined; + let released = false; + let closeCheck: ReturnType | undefined; + const detach = (): void => { + clearInterval(closeCheck); + source.removeListener('error', onError); + source.removeListener('close', detach); + }; + const onError = (cause: Error): void => { + error ??= cause; + if (released) detach(); + }; + source.on('error', onError); + return { + source, + get error() { return error ?? source.errored; }, + release(consumed = false): void { + if (released) return; + released = true; + if (!source.destroyed && (!consumed || source.readableEnded)) { + detach(); + return; + } + if (source.closed) { + // A synchronous _destroy may have already queued an error. + setImmediate(detach); + return; + } + source.once('close', detach); + closeCheck = setInterval(() => { + if (source.closed) detach(); + }, DESTROY_CHECK_MS); + closeCheck.unref(); + if (!source.destroyed) source.destroy(); + }, + }; +} + +/** + * Consume a Readable into GridFS-sized slices without buffering the whole payload. + * Retention is limited to a residual shorter than the next expected slice plus the + * current incoming `read()` buffer. The stream is not pulled again until the + * current complete slices have been awaited, so Node backpressure applies. + * + * A premature `close` (destroy without error) or an already-destroyed source + * rejects in finite time and settles once. An in-flight `onChunk` write is + * awaited before settlement, and any remaining buffered slices are discarded. + */ +export async function consumeReadableChunks( + source: Readable, + fromChunk: number, + length: number, + chunkSizeBytes: number, + onChunk: (slice: StreamedChunk) => Promise, + onBufferedBytes?: (bytes: number) => void, + watched?: ReturnType, +): Promise<{ bytesAccepted: number }> { + const stream = watched ?? watchReadable(source); + const total = chunkCount(length, chunkSizeBytes); + let residual = Buffer.alloc(0); + let index = fromChunk; + let bytesAccepted = 0; + let ended = false; + let aborted = false; + let queuedError: unknown; + let busy = false; + let settled = false; + let pending = false; + + const report = (incoming = 0): void => { + onBufferedBytes?.(residual.length + incoming); + }; + + const emit = async (slice: StreamedChunk): Promise => { + report(); + await onChunk(slice); + bytesAccepted += slice.bytes.length; + index += 1; + }; + + const takeFrom = async (incoming: Buffer): Promise => { + let offset = 0; + report(incoming.length); + while (offset < incoming.length) { + if (aborted || stream.error || isPrematureClose(source, ended)) return; + if (index >= total) badChunkSize(); + const want = chunkLengthAt(index, length, chunkSizeBytes); + const have = residual.length + (incoming.length - offset); + if (have < want) { + const rest = incoming.subarray(offset); + residual = residual.length === 0 + ? Buffer.from(rest) + : Buffer.concat([residual, rest]); + report(); + return; + } + let bytes: Buffer; + if (residual.length === 0) { + bytes = Buffer.from(incoming.subarray(offset, offset + want)); + offset += want; + } else { + const need = want - residual.length; + bytes = Buffer.concat([residual, incoming.subarray(offset, offset + need)]); + residual = Buffer.alloc(0); + offset += need; + } + await emit({ index, bytes }); + } + report(); + }; + + await new Promise((resolve, reject) => { + const detach = (): void => { + clearInterval(checkDestroyed); + source.removeListener('readable', onReadable); + source.removeListener('end', onEnd); + source.removeListener('error', onError); + source.removeListener('close', onClose); + }; + + const finish = (error?: unknown): void => { + if (settled) return; + settled = true; + detach(); + if (error) reject(error); + else resolve(); + }; + + const abort = (error: unknown): void => { + aborted = true; + queuedError = queuedError ?? error; + if (busy) { + pending = true; + return; + } + finish(queuedError); + }; + + const onReadable = (): void => run(); + const onEnd = (): void => { + ended = true; + run(); + }; + const onError = (error: unknown): void => abort(error); + const onClose = (): void => { + if (ended || source.readableEnded) run(); + else abort(streamClosedError()); + }; + + const pump = async (): Promise => { + if (settled || busy) return; + busy = true; + try { + let incoming: Buffer | string | null; + while (!aborted && !stream.error && !isPrematureClose(source, ended) && + (incoming = source.read() as Buffer | string | null) !== null) { + await takeFrom(Buffer.isBuffer(incoming) ? incoming : Buffer.from(incoming)); + } + if (queuedError || stream.error) throw queuedError ?? stream.error; + if (isPrematureClose(source, ended)) throw streamClosedError(); + if (ended || source.readableEnded) { + ended = true; + if (residual.length > 0) badChunkSize(); + } + } finally { + busy = false; + } + }; + + const run = (): void => { + if (settled) return; + if (busy) { + pending = true; + return; + } + if (queuedError || stream.error) { + finish(queuedError ?? stream.error); + return; + } + if (isPrematureClose(source, ended)) { + finish(streamClosedError()); + return; + } + pump().then(() => { + if (settled) return; + if (pending || queuedError || stream.error || isPrematureClose(source, ended)) { + pending = false; + run(); + return; + } + // Node starts auto-destruction after end listeners return. + // Once it starts, wait for its result before accepting the append. + if (ended && (!source.destroyed || source.closed)) finish(); + }, (error: unknown) => finish(error)); + }; + + source.on('readable', onReadable); + source.on('end', onEnd); + source.on('error', onError); + source.on('close', onClose); + // emitClose:false suppresses the notification even when an idle source is destroyed. + const checkDestroyed = setInterval(() => { + if (isPrematureClose(source, ended)) abort(stream.error ?? streamClosedError()); + else if (ended && source.closed) run(); + }, DESTROY_CHECK_MS); + checkDestroyed.unref(); + run(); + if (!ended && source.readableEnded) { + ended = true; + run(); + } + }).finally(() => { + stream.release(true); + }); + + if (bytesAccepted === 0) badChunkSize(); + return { bytesAccepted }; +} diff --git a/src/capabilities/gridfs/types.ts b/src/capabilities/gridfs/types.ts index baaf27f..411acd1 100644 --- a/src/capabilities/gridfs/types.ts +++ b/src/capabilities/gridfs/types.ts @@ -85,7 +85,9 @@ export interface GridFSUploadSession { metadata?: Record; contentType?: string; kind?: 'resumable'; - status: 'pending' | 'finalizing' | 'aborting' | 'finalized' | 'aborted'; + status: 'initializing' | 'pending' | 'finalizing' | 'aborting' | 'finalized' | 'aborted'; + sessionToken?: string; + ownerToken?: string; activeAppend?: { token: string; startedAt: Date }; expiresAt?: Date; nextChunk: number; diff --git a/test/integration/gridfs/gridfs-inspect-orphans.test.ts b/test/integration/gridfs/gridfs-inspect-orphans.test.ts new file mode 100644 index 0000000..7e58b81 --- /dev/null +++ b/test/integration/gridfs/gridfs-inspect-orphans.test.ts @@ -0,0 +1,261 @@ +import { after, before, describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { execFile } from 'node:child_process'; +import { promisify } from 'node:util'; +import { Collection, MongoClient, ObjectId, type Db } from 'mongodb'; +import { createMemoryServerBootstrap } from '../../bootstrap/memory-server'; + +const { inspectGridFSOrphans } = require(process.cwd() + '/scripts/maintenance/gridfs-inspect-orphans.cjs'); + +describe('GridFS inspect-only orphan reporter', () => { + 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(); + }); + + async function withDb(fn: (db: Db) => Promise): Promise { + const db = raw.db('gridfs_inspect_' + (++counter)); + try { + await db.collection('b.chunks').createIndex({ files_id: 1, n: 1 }, { unique: true }); + await fn(db); + } finally { + await db.dropDatabase(); + } + } + + it('rejects an explicitly empty CLI continuation but accepts an omitted or valid one', { timeout: 15_000 }, async () => { + await withDb(async (db) => { + const first = new ObjectId(); + const second = new ObjectId(); + await db.collection('b.chunks').insertMany([{ files_id: first, n: 0 }, { files_id: second, n: 0 }]); + const runCli = (...args: string[]) => promisify(execFile)(process.execPath, [ + 'scripts/maintenance/gridfs-inspect-orphans.cjs', `--db=${db.databaseName}`, '--bucket=b', ...args, + ], { env: { ...process.env, MONSQLIZE_RECOVERY_MONGODB_URI: uri } }); + + await assert.rejects(runCli('--after-files-id='), { + code: 1, stdout: '', stderr: 'GRIDFS_INSPECT_PRECONDITION_FAILED: after.filesId must be an ObjectId\n', + }); + const { stdout } = await runCli('--max-findings=1'); + const [finding, summary] = stdout.trim().split('\n').map((line) => JSON.parse(line)); + assert.equal(finding.filesId, first.toHexString()); + assert.equal(summary.resumeAfter.filesId, first.toHexString()); + const continued = await runCli(`--after-files-id=${summary.resumeAfter.filesId}`); + const [next, end] = continued.stdout.trim().split('\n').map((line) => JSON.parse(line)); + assert.equal(next.filesId, second.toHexString()); + assert.equal(end.truncated, false); + }); + }); + + it('classifies visible, owned, and candidate orphan files_id groups without deleting', async () => { + await withDb(async (db) => { + const visible = new ObjectId(); + const owned = new ObjectId(); + const orphan = new ObjectId(); + const ancient = new Date('2018-01-01T00:00:00.000Z'); + await db.collection('b.files').insertOne({ + _id: visible, filename: 'ok.bin', length: 4, chunkSize: 4, uploadDate: new Date(), + }); + await db.collection('b.chunks').insertMany([ + { files_id: visible, n: 0, data: Buffer.from('ABCD') }, + { files_id: owned, n: 0, data: Buffer.from('PEND') }, + { files_id: orphan, n: 0, data: Buffer.from('ORPH') }, + { files_id: orphan, n: 1, data: Buffer.from('ANED') }, + ]); + await db.collection('b.uploads').insertOne({ + _id: owned, kind: 'resumable', status: 'pending', createdAt: ancient, updatedAt: ancient, + }); + + const originalDeleteMany = Collection.prototype.deleteMany; + let deletes = 0; + (Collection.prototype as any).deleteMany = async function (...args: unknown[]) { + deletes += 1; + return (originalDeleteMany as any).apply(this, args); + }; + try { + await assert.rejects( + () => inspectGridFSOrphans({ db, bucket: 'b', apply: true }), + /no apply mode/, + ); + const report = await inspectGridFSOrphans({ db, bucket: 'b', pageSize: 2 }); + assert.equal(report.filesIdsScanned, 3); + assert.equal(report.visibleFiles, 1); + assert.equal(report.owned, 1); + assert.equal(report.candidateOrphans, 1); + assert.ok(report.pages >= 2); + const byId = Object.fromEntries(report.findings.map((row: { filesId: string }) => [row.filesId, row])); + assert.equal(byId[visible.toHexString()].classification, 'visible_file'); + assert.equal(byId[owned.toHexString()].classification, 'owned'); + assert.equal(byId[owned.toHexString()].ownerStatus, 'pending'); + assert.equal(byId[orphan.toHexString()].classification, 'candidate_orphan'); + assert.equal(byId[orphan.toHexString()].chunkCount, 2); + assert.equal(deletes, 0); + assert.equal(await db.collection('b.chunks').countDocuments(), 4); + assert.equal(await db.collection('b.uploads').countDocuments(), 1); + } finally { + Collection.prototype.deleteMany = originalDeleteMany; + } + }); + }); + + it('re-checks ownership after pagination and does not classify a late-visible id as an orphan', async () => { + await withDb(async (db) => { + const first = new ObjectId(); + const second = new ObjectId(); + const third = new ObjectId(); + // ObjectIds are time-ordered; insert so walk order is first, second, third. + const ids = [first, second, third].sort((a, b) => a.toString().localeCompare(b.toString())); + await db.collection('b.chunks').insertMany(ids.map((id) => ({ + files_id: id, n: 0, data: Buffer.from('DATA'), + }))); + + const originalFindOne = Collection.prototype.findOne; + let published = false; + (Collection.prototype as any).findOne = async function (...args: unknown[]) { + // After the first files_id is classified, make the last id visible before its re-check. + if (this.collectionName === 'b.files') { + const filter = args[0] as { _id?: ObjectId }; + if (!published && filter?._id && filter._id.equals(ids[0])) { + published = true; + await db.collection('b.files').insertOne({ + _id: ids[2], filename: 'late.bin', length: 4, chunkSize: 4, uploadDate: new Date(), + }); + } + } + return (originalFindOne as any).apply(this, args); + }; + try { + const report = await inspectGridFSOrphans({ db, bucket: 'b', pageSize: 1 }); + const byId = Object.fromEntries(report.findings.map((row: { filesId: string }) => [row.filesId, row])); + assert.equal(byId[ids[0].toHexString()].classification, 'candidate_orphan'); + assert.equal(byId[ids[2].toHexString()].classification, 'visible_file'); + assert.equal(await db.collection('b.chunks').countDocuments(), 3); + } finally { + Collection.prototype.findOne = originalFindOne; + } + }); + }); + + it('recognizes a file published between the files and owner lookups', async () => { + await withDb(async (db) => { + const id = new ObjectId(); + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('DATA') }); + await db.collection('b.uploads').insertOne({ _id: id, status: 'uploading' }); + const original = Collection.prototype.findOne; + Collection.prototype.findOne = async function (this: Collection, ...args: unknown[]) { + const filter = args[0] as { _id?: ObjectId } | undefined; + if (this.collectionName === 'b.uploads' && id.equals(filter?._id)) { + await db.collection('b.files').insertOne({ + _id: id, filename: 'published.bin', length: 4, chunkSize: 4, uploadDate: new Date(), + }); + await db.collection('b.uploads').deleteOne({ _id: id }); + } + return Reflect.apply(original, this, args); + } as typeof original; + try { + const report = await inspectGridFSOrphans({ db, bucket: 'b' }); + assert.equal(report.findings[0].classification, 'visible_file'); + assert.equal(report.candidateOrphans, 0); + } finally { + Collection.prototype.findOne = original; + } + }); + }); + + it('counts a group split across batches starting with a negative chunk index', async () => { + await withDb(async (db) => { + const id = new ObjectId(); + const nextId = new ObjectId(); + await db.collection('b.chunks').insertMany([ + ...[-1, 0, 1].map((n) => ({ files_id: id, n, data: Buffer.from('DATA') })), + { files_id: nextId, n: 0, data: Buffer.from('NEXT') }, + ]); + const first = await inspectGridFSOrphans({ db, bucket: 'b', pageSize: 1, maxFindings: 1 }); + assert.deepEqual(first.findings, [{ + filesId: id.toHexString(), chunkCount: 3, minN: -1, maxN: 1, classification: 'candidate_orphan', + }]); + assert.equal(first.truncated, true); + const resumed = await inspectGridFSOrphans({ db, bucket: 'b', pageSize: 1, maxFindings: 1, after: first.resumeAfter }); + assert.equal(resumed.truncated, false); + assert.equal(resumed.findings[0].filesId, nextId.toHexString()); + assert.equal(await db.collection('b.chunks').countDocuments(), 4); + }); + }); + + it('skips foreign IDs and preserves an ObjectId continuation on mixed collections', async () => { + await withDb(async (db) => { + const firstId = new ObjectId(); + const secondId = new ObjectId(); + await db.collection('b.chunks').insertMany([ + { files_id: 3, n: 0 }, + { files_id: 'foreign', n: 0 }, + { files_id: firstId.toHexString(), n: 0 }, + { files_id: firstId, n: 0 }, + { files_id: secondId, n: 0 }, + ]); + const first = await inspectGridFSOrphans({ db, bucket: 'b', pageSize: 1, maxFindings: 1 }); + assert.equal(first.findings[0].filesId, firstId.toHexString()); + assert.deepEqual(first.resumeAfter, { filesId: firstId.toHexString() }); + const next = await inspectGridFSOrphans({ db, bucket: 'b', pageSize: 1, after: first.resumeAfter }); + assert.deepEqual(next.findings.map((row: { filesId: string }) => row.filesId), [secondId.toHexString()]); + assert.equal(next.truncated, false); + await assert.rejects(inspectGridFSOrphans({ db, bucket: 'b', after: { filesId: 'foreign' } }), { + code: 'GRIDFS_INSPECT_PRECONDITION_FAILED', + }); + assert.equal(await db.collection('b.chunks').countDocuments(), 5); + }); + }); + + it('caps findings, streams results, and resumes a large-cardinality scan without deleting', async () => { + await withDb(async (db) => { + const total = 250; + const ids = Array.from({ length: total }, () => new ObjectId()); + ids.sort((left, right) => left.toString().localeCompare(right.toString())); + await db.collection('b.chunks').insertMany(ids.map((id) => ({ + files_id: id, n: 0, data: Buffer.from('DATA'), + }))); + const streamed: string[] = []; + const first = await inspectGridFSOrphans({ + db, bucket: 'b', pageSize: 11, maxFindings: 40, + onFinding: (row: { filesId: string }) => { streamed.push(row.filesId); }, + }); + assert.equal(first.filesIdsScanned, 40); + assert.equal(first.findings.length, 40); + assert.equal(first.truncated, true); + assert.ok(first.resumeAfter?.filesId); + assert.equal(streamed.length, 40); + assert.equal(first.candidateOrphans, 40); + const second = await inspectGridFSOrphans({ + db, bucket: 'b', pageSize: 11, maxFindings: 40, after: first.resumeAfter, + collectFindings: false, + onFinding: (row: { filesId: string }) => { streamed.push(row.filesId); }, + }); + assert.equal(second.findings, undefined); + assert.equal(second.filesIdsScanned, 40); + assert.equal(second.truncated, true); + const rest = await inspectGridFSOrphans({ + db, bucket: 'b', pageSize: 17, maxFindings: 200, after: second.resumeAfter, + }); + assert.equal(rest.truncated, false); + assert.equal(rest.resumeAfter, null); + assert.equal(first.filesIdsScanned + second.filesIdsScanned + rest.filesIdsScanned, total); + // streamed includes first+second pages; rest.findings is the remainder + const all = new Set([ + ...first.findings.map((row: { filesId: string }) => row.filesId), + ...streamed.slice(40), + ...rest.findings.map((row: { filesId: string }) => row.filesId), + ]); + assert.equal(all.size, total); + assert.equal(await db.collection('b.chunks').countDocuments(), total); + }); + }); +}); diff --git a/test/integration/gridfs/gridfs-remaining-issues.test.ts b/test/integration/gridfs/gridfs-remaining-issues.test.ts new file mode 100644 index 0000000..dae32ae --- /dev/null +++ b/test/integration/gridfs/gridfs-remaining-issues.test.ts @@ -0,0 +1,769 @@ +import { createHash } from 'node:crypto'; +import { Readable } from 'node:stream'; +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'; + +const MonSQLize = require('../../../dist/cjs/index.cjs'); + +function sha(data: Buffer): string { + return createHash('sha256').update(data).digest('hex'); +} + +function delayed(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function gate(): { wait: Promise; open: () => void } { + let open = (): void => {}; + const wait = new Promise((resolve) => { open = resolve; }); + return { wait, open }; +} + +describe('GridFS remaining-issue regressions (begin race, finalize rollback, streaming append)', () => { + const bootstrap = createMemoryServerBootstrap(); + let uri = ''; + let raw: MongoClient; + + 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(databaseName: string, fn: (runtime: any, db: Db) => Promise): Promise { + const runtime = new MonSQLize({ type: 'mongodb', databaseName, config: { uri } }); + await runtime.connect(); + try { + await fn(runtime, raw.db(databaseName)); + } finally { + await runtime.close(); + await raw.db(databaseName).dropDatabase(); + } + } + + it('beginResumableUpload does not delete chunks of a file that became visible after precheck', { timeout: 30_000 }, async () => { + await withRuntime('gridfs_begin_precheck_race', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const id = new ObjectId(); + const payload = Buffer.from('ABCD'); + const original = Collection.prototype.insertOne; + (Collection.prototype as any).insertOne = async function (this: Collection, ...args: unknown[]) { + const document = args[0] as Record; + if (this.collectionName === 'b.uploads' && document.kind === 'resumable' && (document._id as ObjectId).equals(id)) { + await original.call(db.collection('b.files'), { + _id: id, + filename: 'visible.bin', + length: 4, + chunkSize: 4, + uploadDate: new Date(), + metadata: { sha256: sha(payload) }, + }); + await original.call(db.collection('b.chunks'), { files_id: id, n: 0, data: payload }); + } + return (original as any).apply(this, args); + }; + try { + await assert.rejects( + bucket.beginResumableUpload('raced.bin', { id, length: 4, sha256: sha(payload) }), + (error: unknown) => (error as { code?: string }).code === 'GRIDFS_FILE_EXISTS', + ); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + assert.deepEqual(await bucket.download(id), payload); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: id }), 0); + } finally { + Collection.prototype.insertOne = original; + } + }); + }); + + it('excludes writes and reuse of an explicit ID until begin cleanup completes', { timeout: 10_000 }, async () => { + await withRuntime('gridfs_initializing', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const id = new ObjectId(); + const payload = Buffer.from('GOOD'); + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('OLD!') }); + const entered = gate(); + const proceed = gate(); + const original = Collection.prototype.deleteMany; + let paused = false; + Collection.prototype.deleteMany = async function (this: Collection, filter, options) { + if (!paused && this.collectionName === 'b.chunks' && id.equals(filter?.files_id)) { + paused = true; + entered.open(); + await proceed.wait; + } + return original.call(this, filter, options); + }; + const starting = bucket.beginResumableUpload('starting.bin', { id, length: 4, sha256: sha(payload) }); + try { + await entered.wait; + const busy = { code: 'GRIDFS_UPLOAD_BUSY' }; + await assert.rejects(bucket.appendChunks(id, 0, payload), busy); + await assert.rejects(bucket.finalizeResumableUpload(id), busy); + await assert.rejects(bucket.abortResumableUpload(id), busy); + await assert.rejects(bucket.upload('replacement.bin', payload, { id }), busy); + const owner = await db.collection('b.uploads').findOne({ _id: id }); + assert.equal(owner?.status, 'initializing'); + assert.equal(owner?.expiresAt, undefined); + } finally { + proceed.open(); + Collection.prototype.deleteMany = original; + await starting; + } + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 0); + await bucket.appendChunks(id, 0, payload); + await bucket.finalizeResumableUpload(id); + assert.deepEqual(await bucket.download(id), payload); + }); + }); + + it('finalize FILE_EXISTS rolls back to pending so public abort works and visible chunks stay', { timeout: 30_000 }, async () => { + await withRuntime('gridfs_finalize_rollback', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('GOOD'); + const session = await bucket.beginResumableUpload('conflict.bin', { + length: 4, sha256: sha(payload), + }); + await bucket.appendChunks(session.uploadId, 0, payload); + await db.collection('b.files').insertOne({ + _id: session.uploadId, + filename: 'other.bin', + length: 4, + chunkSize: 4, + uploadDate: new Date(), + metadata: { sha256: 'a'.repeat(64) }, + }); + await assert.rejects( + bucket.finalizeResumableUpload(session.uploadId), + (error: unknown) => (error as { code?: string }).code === 'GRIDFS_FILE_EXISTS', + ); + assert.equal((await db.collection('b.uploads').findOne({ _id: session.uploadId }))?.status, 'pending'); + await bucket.abortResumableUpload(session.uploadId); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: session.uploadId }), 0); + assert.ok(await db.collection('b.files').findOne({ _id: session.uploadId })); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: session.uploadId }), 1); + }); + }); + + it('retains the finalizing session when the visible file declares a different chunk size', async () => { + await withRuntime('gridfs_finalize_wrong_size', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('aaaabbbbcccc'); + const { uploadId } = await bucket.beginResumableUpload('retry.bin', { length: 12, sha256: sha(payload) }); + await bucket.appendChunks(uploadId, 0, payload); + const uploads = db.collection('b.uploads'); + const files = db.collection('b.files'); + await uploads.updateOne({ _id: uploadId }, { $set: { status: 'finalizing' }, $unset: { expiresAt: '' } }); + await files.insertOne({ + _id: uploadId, filename: 'retry.bin', length: 12, chunkSize: 8, + uploadDate: new Date(), metadata: { sha256: sha(payload) }, + }); + const owner = await uploads.findOne({ _id: uploadId }); + await assert.rejects(bucket.finalizeResumableUpload(uploadId), { code: 'GRIDFS_INTEGRITY_MISMATCH' }); + assert.deepEqual(await uploads.findOne({ _id: uploadId }), owner); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: uploadId }), 3); + + await files.updateOne({ _id: uploadId }, { $set: { chunkSize: 4 } }); + assert.equal((await bucket.finalizeResumableUpload(uploadId)).chunkSize, 4); + assert.deepEqual(await bucket.download(uploadId), payload); + }); + }); + + it('returns the file committed by this finalize after a concurrent deletion and ID reuse', { timeout: 10_000 }, async () => { + await withRuntime('gridfs_finalize_result_reuse', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const id = new ObjectId(); + const firstBytes = Buffer.from('OLD!'); + const replacementBytes = Buffer.from('NEW!NEW!'); + await bucket.beginResumableUpload('first.bin', { + id, length: firstBytes.length, sha256: sha(firstBytes), + contentType: 'application/octet-stream', metadata: { label: 'first' }, + }); + await bucket.appendChunks(id, 0, firstBytes); + + const released = gate(); + const returnFirst = gate(); + const original = Collection.prototype.deleteOne; + let pauseFirst = true; + Collection.prototype.deleteOne = async function (this: Collection, filter, options) { + const result = await original.call(this, filter, options); + if (pauseFirst && this.collectionName === 'b.uploads' && filter?.status === 'finalizing' && id.equals(filter._id)) { + pauseFirst = false; + released.open(); + await returnFirst.wait; + } + return result; + }; + const first = bucket.finalizeResumableUpload(id); + try { + await released.wait; + const committed = await bucket.getFile(id); + assert.equal(committed?.filename, 'first.bin'); + assert.equal(committed?.length, firstBytes.length); + assert.equal(committed?.metadata?.sha256, sha(firstBytes)); + assert.equal(committed?.contentType, 'application/octet-stream'); + + await bucket.delete(id); + await bucket.beginResumableUpload('replacement.bin', { + id, length: replacementBytes.length, sha256: sha(replacementBytes), + chunkSizeBytes: 8, metadata: { label: 'replacement' }, + }); + await bucket.appendChunks(id, 0, replacementBytes); + const replacement = await bucket.finalizeResumableUpload(id); + + returnFirst.open(); + assert.deepEqual(await first, committed); + assert.deepEqual(await bucket.getFile(id), replacement); + assert.deepEqual(await bucket.download(id), replacementBytes); + assert.equal(await db.collection('b.uploads').findOne({ _id: id }), null); + } finally { + returnFirst.open(); + Collection.prototype.deleteOne = original; + await Promise.allSettled([first]); + } + }); + }); + + it('rejects a stale finalize retry that sees a replacement with the same bytes', { timeout: 10_000 }, async () => { + await withRuntime('gridfs_finalize_stale_retry', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const id = new ObjectId(); + const payload = Buffer.from('SAME'); + const opts = { id, length: payload.length, sha256: sha(payload) }; + await bucket.beginResumableUpload('first.bin', opts); + await bucket.appendChunks(id, 0, payload); + + const inserted = gate(); + const reading = gate(); + const releaseFirst = gate(); + const resumeRetry = gate(); + const originalDelete = Collection.prototype.deleteOne; + const originalFind = Collection.prototype.findOne; + let pauseRelease = true; + let pauseRead = false; + Collection.prototype.deleteOne = async function (this: Collection, filter, options) { + if (pauseRelease && this.collectionName === 'b.uploads' && filter?.status === 'finalizing') { + pauseRelease = false; + inserted.open(); + await releaseFirst.wait; + } + return originalDelete.call(this, filter, options); + }; + Collection.prototype.findOne = async function (this: Collection, ...args: unknown[]) { + if (this.collectionName === 'b.files' && pauseRead) { + pauseRead = false; + reading.open(); + await resumeRetry.wait; + } + return Reflect.apply(originalFind, this, args); + } as typeof originalFind; + + const first = bucket.finalizeResumableUpload(id); + const pending = [first]; + try { + await inserted.wait; + pauseRead = true; + const retry = assert.rejects(bucket.finalizeResumableUpload(id), { code: 'GRIDFS_UPLOAD_BUSY' }); + pending.push(retry); + await reading.wait; + releaseFirst.open(); + await first; + await bucket.delete(id); + await bucket.beginResumableUpload('replacement.bin', { ...opts, metadata: { label: 'replacement' } }); + await bucket.appendChunks(id, 0, payload); + await bucket.finalizeResumableUpload(id); + resumeRetry.open(); + await retry; + assert.equal((await bucket.getFile(id))?.filename, 'replacement.bin'); + assert.deepEqual(await bucket.download(id), payload); + assert.equal(await db.collection('b.uploads').findOne({ _id: id }), null); + } finally { + releaseFirst.open(); + resumeRetry.open(); + Collection.prototype.findOne = originalFind; + Collection.prototype.deleteOne = originalDelete; + await Promise.allSettled(pending); + } + }); + }); + + for (const operation of ['finalizeResumableUpload', 'abortResumableUpload'] as const) { + for (const legacy of [false, true]) { + it(`${operation} does not take over a replacement pending session after reading a ${legacy ? 'legacy' : 'tokenized'} session`, { timeout: 10_000 }, async () => { + await withRuntime('gridfs_pending_session_reuse', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const id = new ObjectId(); + const payload = Buffer.from('DATA'); + const options = { id, length: payload.length, sha256: sha(payload) }; + await bucket.beginResumableUpload('old.bin', options); + await bucket.appendChunks(id, 0, payload); + const uploads = db.collection('b.uploads'); + if (legacy) await uploads.updateOne({ _id: id }, { $unset: { sessionToken: '' } }); + const oldSession = await uploads.findOne({ _id: id }); + assert.ok(oldSession?.createdAt instanceof Date); + + const claiming = gate(); + const resume = gate(); + const original = Collection.prototype.updateOne; + const status = operation === 'finalizeResumableUpload' ? 'finalizing' : 'aborting'; + let pause = true; + Collection.prototype.updateOne = async function (this: Collection, filter, update, opts) { + if (pause && this.collectionName === 'b.uploads' && id.equals(filter?._id) && + !Array.isArray(update) && update.$set?.status === status) { + pause = false; + claiming.open(); + await resume.wait; + } + return original.call(this, filter, update, opts); + }; + + const stale = bucket[operation](id); + try { + await claiming.wait; + await bucket.abortResumableUpload(id); + await bucket.beginResumableUpload('new.bin', options); + await bucket.appendChunks(id, 0, payload); + await uploads.updateOne({ _id: id }, { $set: { createdAt: oldSession.createdAt } }); + const replacement = await uploads.findOne({ _id: id }); + assert.equal(replacement?.status, 'pending'); + + resume.open(); + await assert.rejects(stale, { code: 'GRIDFS_UPLOAD_BUSY' }); + assert.deepEqual(await uploads.findOne({ _id: id }), replacement); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + assert.equal(await bucket.getFile(id), null); + assert.equal((await bucket.finalizeResumableUpload(id)).filename, 'new.bin'); + assert.deepEqual(await bucket.download(id), payload); + } finally { + resume.open(); + Collection.prototype.updateOne = original; + await Promise.allSettled([stale]); + } + }); + }); + } + } + + it('bounds missing-chunk diagnostics and counts the complete gap', { timeout: 10_000 }, async () => { + await withRuntime('gridfs_finalize_gap_bound', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 1 }); + const length = 10_000_000_000; + const { uploadId } = await bucket.beginResumableUpload('large.bin', { + length, sha256: sha(Buffer.from('unused')), + }); + await db.collection('b.chunks').insertMany([ + { files_id: uploadId, n: 0, data: Buffer.from('A') }, + { files_id: uploadId, n: length - 2, data: Buffer.from('B') }, + ]); + await assert.rejects(bucket.finalizeResumableUpload(uploadId), (error: unknown) => { + const incomplete = error as { code: string; details: { missingChunks: number[]; missingChunkCount: number }[] }; + assert.equal(incomplete.code, 'GRIDFS_UPLOAD_INCOMPLETE'); + assert.equal(incomplete.details[0].missingChunkCount, length - 2); + assert.deepEqual(incomplete.details[0].missingChunks, Array.from({ length: 1024 }, (_, index) => index + 1)); + return true; + }); + assert.equal((await db.collection('b.uploads').findOne({ _id: uploadId }))?.status, 'pending'); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: uploadId }), 2); + assert.equal(await db.collection('b.files').countDocuments({ _id: uploadId }), 0); + }); + }); + + it('uses primary reads for visibility checks before orphan cleanup', async () => { + await withRuntime('gridfs_primary_cleanup', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('LIVE'); + const { id } = await bucket.upload('visible.bin', payload); + const original = Collection.prototype.findOne; + Collection.prototype.findOne = async function (this: Collection, ...args: unknown[]) { + const options = args[1] as { readPreference?: string } | undefined; + // Model a secondary that has not yet observed the file commit. + if (this.collectionName === 'b.files' && options?.readPreference !== 'primary') return null; + return Reflect.apply(original, this, args); + } as typeof original; + try { + await bucket.abortResumableUpload(id); + await assert.rejects(bucket.beginResumableUpload('duplicate.bin', { + id, length: payload.length, sha256: sha(payload), + }), { code: 'GRIDFS_FILE_EXISTS' }); + } finally { + Collection.prototype.findOne = original; + } + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + assert.deepEqual(await bucket.download(id), payload); + }); + }); + + it('keeps the new finalize owner when an older retry completes after ID reuse', { timeout: 10_000 }, async () => { + await withRuntime('gridfs_finalize_owner_reuse', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const id = new ObjectId(); + const oldBytes = Buffer.from('OLD!'); + const newBytes = Buffer.from('NEW!'); + const firstCleanup = gate(); + const retryCleanup = gate(); + const newInsert = gate(); + const finishFirst = gate(); + const finishRetry = gate(); + const finishNew = gate(); + const originalDelete = Collection.prototype.deleteOne; + const originalInsert = Collection.prototype.insertOne; + await bucket.beginResumableUpload('old.bin', { id, length: 4, sha256: sha(oldBytes) }); + await bucket.appendChunks(id, 0, oldBytes); + let cleanups = 0; + Collection.prototype.deleteOne = async function (this: Collection, filter, options) { + if (this.collectionName === 'b.uploads' && filter?.status === 'finalizing' && id.equals(filter._id)) { + cleanups += 1; + if (cleanups === 1) { + firstCleanup.open(); + await finishFirst.wait; + } else if (cleanups === 2) { + retryCleanup.open(); + await finishRetry.wait; + } + } + return originalDelete.call(this, filter, options); + }; + Collection.prototype.insertOne = async function (this: Collection, doc, options) { + if (this.collectionName === 'b.files' && doc.filename === 'new.bin') { + newInsert.open(); + await finishNew.wait; + } + return originalInsert.call(this, doc, options); + }; + const pending = []; + try { + const first = bucket.finalizeResumableUpload(id); + pending.push(first); + await firstCleanup.wait; + const oldRetry = assert.rejects(bucket.finalizeResumableUpload(id), { code: 'GRIDFS_UPLOAD_BUSY' }); + pending.push(oldRetry); + await retryCleanup.wait; + finishFirst.open(); + await first; + await bucket.delete(id); + await bucket.beginResumableUpload('new.bin', { id, length: 4, sha256: sha(newBytes) }); + await bucket.appendChunks(id, 0, newBytes); + const current = bucket.finalizeResumableUpload(id); + pending.push(current); + await newInsert.wait; + const uploads = db.collection('b.uploads'); + const owner = await uploads.findOne({ _id: id }); + assert.equal(owner?.status, 'finalizing'); + + finishRetry.open(); + await oldRetry; + assert.deepEqual(await uploads.findOne({ _id: id }), owner); + await assert.rejects(bucket.abortResumableUpload(id), { code: 'GRIDFS_UPLOAD_BUSY' }); + + finishNew.open(); + assert.equal((await current).filename, 'new.bin'); + assert.deepEqual(await bucket.download(id), newBytes); + assert.equal(await uploads.findOne({ _id: id }), null); + } finally { + finishFirst.open(); + finishRetry.open(); + finishNew.open(); + Collection.prototype.deleteOne = originalDelete; + Collection.prototype.insertOne = originalInsert; + await Promise.allSettled(pending); + } + }); + }); + + it('rejects malformed or extra chunk indexes even when the expected bytes are present', async () => { + await withRuntime('gridfs_finalize_invalid_index', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('GOOD'); + for (const n of [-1, 'invalid', 0.5, 1, Number.MAX_SAFE_INTEGER + 1]) { + const { uploadId } = await bucket.beginResumableUpload(`bad-index-${n}.bin`, { + length: payload.length, sha256: sha(payload), + }); + await bucket.appendChunks(uploadId, 0, payload); + const chunks = db.collection('b.chunks'); + const extra = await chunks.insertOne({ files_id: uploadId, n, data: Buffer.from('BAD!') }); + + await assert.rejects(bucket.finalizeResumableUpload(uploadId), { + code: 'GRIDFS_INTEGRITY_MISMATCH', + }, `chunk index ${n}`); + assert.equal(await db.collection('b.files').findOne({ _id: uploadId }), null); + assert.equal((await db.collection('b.uploads').findOne({ _id: uploadId }))?.status, 'pending'); + assert.equal(await chunks.countDocuments({ files_id: uploadId }), 2); + + await chunks.deleteOne({ _id: extra.insertedId }); + const info = await bucket.finalizeResumableUpload(uploadId); + assert.deepEqual(await bucket.download(info.id), payload); + } + }); + }); + + it('streams a Readable append with ownership, idempotent replay, and recoverable partial failure', { timeout: 30_000 }, async () => { + await withRuntime('gridfs_stream_append', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('aaaabbbbcccc'); + const otherId = new ObjectId(); + await db.collection('b.chunks').insertOne({ files_id: otherId, n: 0, data: Buffer.from('KEEP') }); + + const session = await bucket.beginResumableUpload('stream.bin', { + length: payload.length, sha256: sha(payload), + }); + const first = await bucket.appendChunks( + session.uploadId, + 0, + Readable.from([payload.subarray(0, 4), payload.subarray(4, 8)]), + ); + assert.equal(first.nextChunk, 2); + await assert.rejects( + bucket.appendChunks(session.uploadId, 2, Readable.from([Buffer.from('cc')])), + (error: unknown) => (error as { code?: string }).code === 'INVALID_ARGUMENT', + ); + assert.deepEqual( + (await db.collection('b.chunks').find({ files_id: session.uploadId }).sort({ n: 1 }).toArray()).map((chunk) => chunk.n), + [0, 1], + ); + const rest = await bucket.appendChunks(session.uploadId, 2, Readable.from([payload.subarray(8)])); + assert.equal(rest.nextChunk, 3); + const replay = await bucket.appendChunks(session.uploadId, 0, Readable.from([payload])); + assert.equal(replay.nextChunk, 3); + const info = await bucket.finalizeResumableUpload(session.uploadId); + assert.deepEqual(await bucket.download(info.id), payload); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: otherId }), 1); + }); + }); + + it('releases activeAppend after a mid-write stream close so retry and abort work', { timeout: 30_000 }, async () => { + await withRuntime('gridfs_stream_close', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('aaaabbbb'); + const session = await bucket.beginResumableUpload('close.bin', { + length: payload.length, sha256: sha(payload), + }); + let pushed = false; + const source = new Readable({ + read() { + if (pushed) return; + pushed = true; + this.push(payload.subarray(0, 4)); + }, + }); + const original = Collection.prototype.insertOne; + let writeStarted: () => void = () => undefined; + const inFlight = new Promise((resolve) => { writeStarted = resolve; }); + (Collection.prototype as any).insertOne = async function (this: Collection, ...args: unknown[]) { + const document = args[0] as Record; + const filesId = document.files_id as ObjectId | undefined; + if (this.collectionName.endsWith('.chunks') && filesId && String(filesId) === String(session.uploadId)) { + writeStarted(); + await delayed(40); + } + return (original as any).apply(this, args); + }; + const appending = bucket.appendChunks(session.uploadId, 0, source); + try { + await Promise.race([inFlight, delayed(2000)]); + source.destroy(); + await Promise.race([ + assert.rejects( + appending, + (error: unknown) => (error as { code?: string }).code === 'INVALID_ARGUMENT', + ), + delayed(5000).then(() => { + throw new Error('appendChunks did not reject within 5s after stream.destroy()'); + }), + ]); + } finally { + Collection.prototype.insertOne = original; + } + const owner = await db.collection('b.uploads').findOne({ _id: session.uploadId }); + assert.equal(owner?.status, 'pending'); + assert.equal(owner?.activeAppend, undefined); + const stored = await db.collection('b.chunks').countDocuments({ files_id: session.uploadId }); + if (stored === 0) { + const resumed = await bucket.appendChunks(session.uploadId, 0, payload); + assert.equal(resumed.nextChunk, 2); + } else { + await assert.rejects( + bucket.appendChunks(session.uploadId, stored + 1, payload.subarray(stored * 4)), + (error: unknown) => (error as { code?: string }).code === 'GRIDFS_RESUME_MISMATCH' + && (error as { details?: { expectedNextChunk?: number }[] }).details?.[0]?.expectedNextChunk === stored, + ); + const resumed = await bucket.appendChunks(session.uploadId, stored, payload.subarray(stored * 4)); + assert.equal(resumed.nextChunk, 2); + } + await bucket.abortResumableUpload(session.uploadId); + assert.equal(await db.collection('b.uploads').countDocuments({ _id: session.uploadId }), 0); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: session.uploadId }), 0); + }); + }); + + it('handles a source failure while acquiring the append claim', { timeout: 10_000 }, async () => { + await withRuntime('gridfs_append_claim_error', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('GOOD'); + const { uploadId } = await bucket.beginResumableUpload('claim.bin', { length: 4, sha256: sha(payload) }); + const claiming = gate(); + const proceed = gate(); + const original = Collection.prototype.updateOne; + Collection.prototype.updateOne = async function (this: Collection, filter, update, options) { + if (this.collectionName === 'b.uploads' && !Array.isArray(update) && update.$set?.activeAppend) { + claiming.open(); + await proceed.wait; + } + return original.call(this, filter, update, options); + }; + const source = new Readable({ read() {} }); + const error = new Error('source failed before claim'); + const rejected = assert.rejects(bucket.appendChunks(uploadId, 0, source), (actual) => actual === error); + try { + await claiming.wait; + source.destroy(error); + await new Promise((resolve) => setImmediate(resolve)); + } finally { + proceed.open(); + Collection.prototype.updateOne = original; + await rejected; + } + assert.equal((await db.collection('b.uploads').findOne({ _id: uploadId }))?.activeAppend, undefined); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: uploadId }), 0); + assert.equal((await bucket.appendChunks(uploadId, 0, payload)).nextChunk, 1); + }); + }); + + it('rejects an append when automatic destruction fails after EOF', async () => { + await withRuntime('gridfs_append_eof_error', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('GOOD'); + const { uploadId } = await bucket.beginResumableUpload('eof.bin', { length: 4, sha256: sha(payload) }); + const destroying = gate(); + const complete = gate(); + const failure = new Error('asynchronous EOF cleanup failed'); + const source = new Readable({ + read() { this.push(payload); this.push(null); }, + destroy(_error, callback) { + destroying.open(); + void complete.wait.then(() => callback(failure)); + }, + }); + const rejected = assert.rejects(bucket.appendChunks(uploadId, 0, source), (error) => error === failure); + try { + await destroying.wait; + await assert.rejects(bucket.abortResumableUpload(uploadId), { code: 'GRIDFS_UPLOAD_BUSY' }); + } finally { + complete.open(); + await rejected; + } + const owner = await db.collection('b.uploads').findOne({ _id: uploadId }); + assert.equal(owner?.nextChunk, 0); + assert.equal(owner?.activeAppend, undefined); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: uploadId }), 1); + assert.equal((await bucket.appendChunks(uploadId, 0, payload)).nextChunk, 1); + }); + }); + + it('scans the stored prefix in bounded batches after a large stream append', async () => { + await withRuntime('gridfs_append_batch_scan', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 1 }); + const payload = Buffer.alloc(600, 7); + const { uploadId } = await bucket.beginResumableUpload('batched.bin', { length: payload.length, sha256: sha(payload) }); + const original = Collection.prototype.find; + const batchSizes: number[] = []; + Collection.prototype.find = function (this: Collection, ...args: unknown[]) { + const cursor = Reflect.apply(original, this, args); + const filter = args[0] as { files_id?: ObjectId; n?: { $gte?: number } } | undefined; + if (this.collectionName === 'b.chunks' && uploadId.equals(filter?.files_id) && filter?.n?.$gte !== undefined) { + const batchSize = cursor.batchSize.bind(cursor); + cursor.batchSize = (size: number) => { batchSizes.push(size); return batchSize(size); }; + cursor.toArray = async () => { throw new Error('prefix scan materialized all matching chunk indexes'); }; + } + return cursor; + } as typeof original; + try { + const result = await bucket.appendChunks(uploadId, 0, Readable.from([payload])); + assert.deepEqual(result, { nextChunk: 600, bytesAccepted: 600 }); + assert.deepEqual(batchSizes, [256, 256]); + } finally { + Collection.prototype.find = original; + } + await bucket.finalizeResumableUpload(uploadId); + assert.deepEqual(await bucket.download(uploadId), payload); + }); + }); + + it('keeps an unread source reusable after pre-consumption rejections', async () => { + await withRuntime('gridfs_append_unread_retry', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('GOOD'); + const { uploadId } = await bucket.beginResumableUpload('retry.bin', { length: 4, sha256: sha(payload) }); + const uploads = db.collection('b.uploads'); + let reads = 0; + const source = new Readable({ + read() { + reads += 1; + this.push(payload); + this.push(null); + }, + }); + await assert.rejects(bucket.appendChunks(new ObjectId(), 0, source), { code: 'GRIDFS_FILE_NOT_FOUND' }); + await uploads.updateOne({ _id: uploadId }, { $set: { activeAppend: { token: 'other-writer' } } }); + await assert.rejects(bucket.appendChunks(uploadId, 0, source), { code: 'GRIDFS_UPLOAD_BUSY' }); + await uploads.updateOne({ _id: uploadId }, { $unset: { activeAppend: '' } }); + await assert.rejects(bucket.appendChunks(uploadId, 1, source), (error: unknown) => { + const mismatch = error as { code: string; details: { expectedNextChunk: number }[] }; + return mismatch.code === 'GRIDFS_RESUME_MISMATCH' && mismatch.details[0].expectedNextChunk === 0; + }); + assert.equal(reads, 0); + assert.equal(source.destroyed, false); + assert.equal(source.listenerCount('error'), 0); + assert.equal((await bucket.appendChunks(uploadId, 0, source)).nextChunk, 1); + await bucket.finalizeResumableUpload(uploadId); + assert.deepEqual(await bucket.download(uploadId), payload); + }); + }); + + it('releases the append claim when an idle source is destroyed without a close event', { timeout: 10_000 }, async () => { + await withRuntime('gridfs_append_silent_close', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('GOOD'); + const { uploadId } = await bucket.beginResumableUpload('idle.bin', { length: 4, sha256: sha(payload) }); + const reading = gate(); + const source = new Readable({ emitClose: false, read() { reading.open(); } }); + const rejected = assert.rejects(bucket.appendChunks(uploadId, 0, source), { code: 'INVALID_ARGUMENT' }); + await reading.wait; + await new Promise((resolve) => setImmediate(resolve)); + source.destroy(); + await rejected; + const owner = await db.collection('b.uploads').findOne({ _id: uploadId }); + assert.equal(owner?.activeAppend, undefined); + assert.ok(owner?.expiresAt instanceof Date); + await bucket.abortResumableUpload(uploadId); + }); + }); + + it('resumes after a streaming append writes one full chunk then rejects a trailing partial', { timeout: 30_000 }, async () => { + await withRuntime('gridfs_stream_resume_partial', async (runtime, db) => { + const bucket = runtime.gridfs('b', { chunkSizeBytes: 4 }); + const payload = Buffer.from('aaaabbbb'); + const session = await bucket.beginResumableUpload('partial-resume.bin', { + length: payload.length, sha256: sha(payload), + }); + await assert.rejects( + bucket.appendChunks(session.uploadId, 0, Readable.from([payload.subarray(0, 4), Buffer.from('bb')])), + (error: unknown) => (error as { code?: string }).code === 'INVALID_ARGUMENT', + ); + assert.deepEqual( + (await db.collection('b.chunks').find({ files_id: session.uploadId }).sort({ n: 1 }).toArray()).map((chunk) => chunk.n), + [0], + ); + const resumed = await bucket.appendChunks(session.uploadId, 1, payload.subarray(4)); + assert.equal(resumed.nextChunk, 2); + assert.deepEqual(await bucket.download((await bucket.finalizeResumableUpload(session.uploadId)).id), payload); + }); + }); +}); diff --git a/test/integration/gridfs/gridfs-stale-recovery.test.ts b/test/integration/gridfs/gridfs-stale-recovery.test.ts index 0f95c7f..e247523 100644 --- a/test/integration/gridfs/gridfs-stale-recovery.test.ts +++ b/test/integration/gridfs/gridfs-stale-recovery.test.ts @@ -261,6 +261,29 @@ describe('GridFS operator recovery: deterministic post-crash state and chunk saf }); }); + it('recovers interrupted initialization after quiescence without deleting a visible file', async () => { + await withRuntime(async (bucket, db) => { + const payload = Buffer.from('GOOD'); + const uploads = db.collection('b.uploads'); + const chunks = db.collection('b.chunks'); + for (const visible of [false, true]) { + const id = new ObjectId(); + if (visible) await bucket.upload('visible.bin', payload, { id }); + else await chunks.insertOne({ files_id: id, n: 0, data: payload }); + await uploads.insertOne({ _id: id, kind: 'resumable', status: 'initializing', + updatedAt: new Date(), ownerToken: 'initialization-token' }); + + await assert.rejects(recover(db, id, { quiesced: false }), /requires external writer quiescence/); + const receipt = await recover(db, id); + assert.equal(receipt.outcome, 'completed'); + assert.equal(await uploads.findOne({ _id: id }), null); + assert.equal(await chunks.countDocuments({ files_id: id }), visible ? 1 : 0); + if (visible) assert.deepEqual(await bucket.download(id), payload); + else await bucket.beginResumableUpload('retry.bin', { id, length: 4, sha256: digest(payload) }); + } + }); + }); + it('can resume a recovery crash using the persistent recovery token', async () => { await withRuntime(async (_, db) => { const id = new ObjectId(); @@ -295,6 +318,34 @@ describe('GridFS operator recovery: deterministic post-crash state and chunk saf }); }); + it('rejects an unknown recovery origin before changing ownership or chunks', async () => { + await withRuntime(async (bucket, db) => { + const payload = Buffer.from('GOOD'); + const { uploadId: id } = await bucket.beginResumableUpload('interrupted.bin', { + length: payload.length, sha256: digest(payload), + }); + await bucket.appendChunks(id, 0, payload); + const uploads = db.collection('b.uploads'); + await uploads.updateOne({ _id: id }, { + $set: { + status: 'recovering', updatedAt: new Date(), + recovery: { token: 'recovery-token', fromStatus: 'unknown' }, + }, + $unset: { expiresAt: '' }, + }); + const owner = await uploads.findOne({ _id: id }); + const chunks = db.collection('b.chunks'); + const originalChunks = await chunks.find({ files_id: id }).toArray(); + + await assert.rejects(recover(db, id, { resumeQuiesced: true }), { + code: 'GRIDFS_RECOVERY_PRECONDITION_FAILED', + }); + assert.deepEqual(await uploads.findOne({ _id: id }), owner); + assert.deepEqual(await chunks.find({ files_id: id }).toArray(), originalChunks); + assert.equal(await db.collection('b.ownership_recovery_audit').countDocuments({ fileId: id }), 0); + }); + }); + it('refuses to clear aborting ownership if a visible file already owns the chunks', async () => { await withRuntime(async (bucket, db) => { const id = new ObjectId(); @@ -547,4 +598,70 @@ describe('GridFS operator recovery: deterministic post-crash state and chunk saf }); }); + + it('fails closed when visibility changes after the 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: 'uploading', ownerToken: 'direct-race', + createdAt: now, updatedAt: now, + }); + await db.collection('b.chunks').insertOne({ files_id: id, n: 0, data: Buffer.from('ABCD') }); + const original = Collection.prototype.updateOne; + (Collection.prototype as any).updateOne = async function (this: Collection, ...args: unknown[]) { + const result = await (original as any).apply(this, args); + const update = args[1] as { $set?: { status?: string } }; + if (this.collectionName === 'b.uploads' && update?.$set?.status === 'recovering') { + await db.collection('b.files').insertOne({ + _id: id, filename: 'appeared.bin', length: 4, chunkSize: 4, uploadDate: new Date(), + }); + } + return result; + }; + try { + await assert.rejects(recover(db, id), /visibility changed after recovery claim/); + assert.equal((await db.collection('b.uploads').findOne({ _id: id }))?.status, 'recovering'); + assert.equal(await db.collection('b.chunks').countDocuments({ files_id: id }), 1); + assert.ok(await db.collection('b.files').findOne({ _id: id })); + } finally { + Collection.prototype.updateOne = original; + } + }); + }); + + it('fails closed when a required visible file cannot be verified after the claim', async () => { + await withRuntime(async (bucket, db) => { + const id = new ObjectId(); + const payload = Buffer.from('ABCD'); + await bucket.upload('final.bin', payload, { id, sha256: digest(payload) }); + // upload already released; implant a leftover finalizing session on the visible id + const now = new Date(); + await db.collection('b.uploads').insertOne({ + _id: id, kind: 'resumable', status: 'finalizing', filename: 'final.bin', + length: 4, sha256: digest(payload), chunkSizeBytes: 4, + createdAt: now, updatedAt: now, + }); + const original = Collection.prototype.findOne; + let verifiedLookups = 0; + (Collection.prototype as any).findOne = async function (this: Collection, ...args: unknown[]) { + const filter = args[0] as { _id?: ObjectId }; + const options = args[1] as { projection?: { _id?: number } } | undefined; + if (this.collectionName === 'b.files' && filter?._id?.equals?.(id) && !options?.projection) { + verifiedLookups += 1; + return null; + } + return (original as any).apply(this, args); + }; + try { + await assert.rejects(recover(db, id), /could not be verified after claim/); + assert.equal(verifiedLookups, 1); + assert.equal((await db.collection('b.uploads').findOne({ _id: id }))?.status, 'recovering'); + assert.ok(await db.collection('b.chunks').countDocuments({ files_id: id }) >= 1); + } finally { + Collection.prototype.findOne = original; + } + assert.ok(await db.collection('b.files').findOne({ _id: id })); + }); + }); }); diff --git a/test/run-tests.cjs b/test/run-tests.cjs index 0627adf..7714ead 100644 --- a/test/run-tests.cjs +++ b/test/run-tests.cjs @@ -54,6 +54,7 @@ const UNIT = [ 'test/unit/expression/operators.test.js', 'test/unit/errors/errors.test.js', 'test/unit/gridfs/gridfs.test.js', + 'test/unit/gridfs/stream-chunks.test.js', 'test/unit/search-indexes/search-indexes.test.js', 'test/unit/management/management.test.js', 'test/unit/writes/batch.test.js', @@ -139,6 +140,8 @@ const INTEGRATION = [ '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/gridfs/gridfs-remaining-issues.test.js', + 'test/integration/gridfs/gridfs-inspect-orphans.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', diff --git a/test/unit/gridfs/gridfs.test.ts b/test/unit/gridfs/gridfs.test.ts index 2810bcf..50c3275 100644 --- a/test/unit/gridfs/gridfs.test.ts +++ b/test/unit/gridfs/gridfs.test.ts @@ -80,15 +80,20 @@ class MemoryCollection { find(filter: Doc) { let rows = this.matched(filter); + const countRow = () => { this.rowsRead += 1; }; const count = (out: Doc[]) => { this.rowsRead += out.length; return out; }; const cursor = { sort: (spec: Doc) => { rows = sortRows(rows, spec); return cursor; }, skip: (count: number) => { rows = rows.slice(count); return cursor; }, limit: (count: number) => { rows = rows.slice(0, count); return cursor; }, project: () => cursor, + batchSize: () => cursor, toArray: async () => count(rows).map((row) => ({ ...row })), async *[Symbol.asyncIterator]() { - for (const row of rows) yield { ...row }; + for (const row of rows) { + countRow(); + yield { ...row }; + } }, }; return cursor; @@ -204,6 +209,7 @@ function bucketFor(options: { driver?: boolean; instanceId?: string; poolName?: string; + logger?: { warn: (...args: unknown[]) => void }; } = {}) { const db = new MemoryDb(); const calls: string[] = []; @@ -267,6 +273,7 @@ function bucketFor(options: { source: options.source ?? 'db', policy: options.policy, driver: options.driver === false ? undefined : driver as never, + logger: options.logger, }); return { bucket, db, calls, bytes }; } @@ -639,4 +646,145 @@ describe('gridfs bucket', () => { return true; }); }); + + it('does not delete a visible file\'s chunks when beginResumableUpload races after precheck', async () => { + const { bucket, db } = bucketFor({ driver: false }); + const id = new ObjectId(); + const payload = Buffer.from('ABCD'); + const uploads = db.collection('fs.uploads'); + const original = uploads.insertOne.bind(uploads); + uploads.insertOne = async (doc: Doc) => { + if (doc.kind === 'resumable' && sameId(doc._id, id)) { + await db.collection('fs.files').insertOne({ + _id: id, filename: 'visible.bin', length: 4, chunkSize: 4, uploadDate: new Date(), + }); + await db.collection('fs.chunks').insertOne({ files_id: id, n: 0, data: payload }); + } + return original(doc); + }; + await assert.rejects( + () => bucket.beginResumableUpload('raced.bin', { id, length: 4, sha256: sha(payload), chunkSizeBytes: 4 }), + (error: unknown) => codeOf(error) === 'GRIDFS_FILE_EXISTS', + ); + assert.equal((await db.collection('fs.chunks').find({ files_id: id }).toArray()).length, 1); + assert.ok(await db.collection('fs.files').findOne({ _id: id })); + assert.equal(await db.collection('fs.uploads').findOne({ _id: id }), null); + }); + + it('rolls finalizing back to pending after FILE_EXISTS or INTEGRITY_MISMATCH so abort works', async () => { + const { bucket, db } = bucketFor({ driver: false }); + const payload = Buffer.from('abcd'); + const conflict = await bucket.beginResumableUpload('conflict.bin', { + length: 4, sha256: sha(payload), chunkSizeBytes: 4, + }); + await bucket.appendChunks(conflict.uploadId, 0, payload); + await db.collection('fs.files').insertOne({ + _id: conflict.uploadId, + filename: 'other.bin', + length: 4, + chunkSize: 4, + uploadDate: new Date(), + metadata: { sha256: 'a'.repeat(64) }, + }); + await assert.rejects( + () => bucket.finalizeResumableUpload(conflict.uploadId), + (error: unknown) => codeOf(error) === 'GRIDFS_FILE_EXISTS', + ); + assert.equal((await db.collection('fs.uploads').findOne({ _id: conflict.uploadId }))?.status, 'pending'); + await bucket.abortResumableUpload(conflict.uploadId); + assert.equal(await db.collection('fs.uploads').findOne({ _id: conflict.uploadId }), null); + assert.ok(await db.collection('fs.files').findOne({ _id: conflict.uploadId })); + assert.equal((await db.collection('fs.chunks').find({ files_id: conflict.uploadId }).toArray()).length, 1); + + const bad = await bucket.beginResumableUpload('bad.bin', { + length: 4, sha256: sha(Buffer.from('nope')), chunkSizeBytes: 4, + }); + await bucket.appendChunks(bad.uploadId, 0, payload); + await assert.rejects( + () => bucket.finalizeResumableUpload(bad.uploadId), + (error: unknown) => codeOf(error) === 'GRIDFS_INTEGRITY_MISMATCH', + ); + assert.equal((await db.collection('fs.uploads').findOne({ _id: bad.uploadId }))?.status, 'pending'); + await bucket.abortResumableUpload(bad.uploadId); + assert.equal((await db.collection('fs.chunks').find({ files_id: bad.uploadId }).toArray()).length, 0); + assert.equal(await db.collection('fs.uploads').findOne({ _id: bad.uploadId }), null); + }); + + it('returns a successful rename even when reservation release fails, and keeps the original error otherwise', async () => { + const warnings: string[] = []; + const { bucket, db } = bucketFor({ + logger: { warn: (message: unknown) => { warnings.push(String(message)); } }, + }); + const info = await bucket.upload('old.bin', Buffer.from('ok')); + const uploads = db.collection('fs.uploads'); + const original = uploads.deleteOne.bind(uploads); + uploads.deleteOne = async (filter: Doc) => { + if (filter.kind === 'rename') return { deletedCount: 0 }; + return original(filter); + }; + await bucket.rename(info.id, 'new.bin'); + assert.equal((await bucket.getFile(info.id))?.filename, 'new.bin'); + assert.ok(warnings.some((message) => message.includes('rename committed but reservation cleanup failed'))); + + warnings.length = 0; + uploads.deleteOne = async () => { + throw driverFailure('release failed', 91); + }; + const missing = new ObjectId(); + await assert.rejects( + () => bucket.rename(missing, 'gone.bin'), + (error: unknown) => codeOf(error) === 'GRIDFS_FILE_NOT_FOUND', + ); + assert.ok(warnings.some((message) => message.includes('rename reservation retained after cleanup failure'))); + assert.ok(await db.collection('fs.uploads').findOne({ _id: missing, kind: 'rename' })); + }); + + it('streams Readable appends, rejects a trailing partial after writing complete chunks, and remains abortable', async () => { + const { bucket, db } = bucketFor({ driver: false }); + const payload = Buffer.from('abcdefghijkl'); + const started = await bucket.beginResumableUpload('stream.bin', { + length: payload.length, sha256: sha(payload), chunkSizeBytes: 4, + }); + const first = await bucket.appendChunks(started.uploadId, 0, Readable.from([payload.subarray(0, 4), payload.subarray(4, 8)])); + assert.equal(first.nextChunk, 2); + await assert.rejects( + () => bucket.appendChunks(started.uploadId, 2, Readable.from([Buffer.from('ij')])), + (error: unknown) => codeOf(error) === 'INVALID_ARGUMENT', + ); + assert.deepEqual( + (await db.collection('fs.chunks').find({ files_id: started.uploadId }).sort({ n: 1 }).toArray()).map((chunk) => chunk.n), + [0, 1], + ); + assert.equal((await db.collection('fs.uploads').findOne({ _id: started.uploadId }))?.activeAppend, undefined); + const rest = await bucket.appendChunks(started.uploadId, 2, Readable.from([payload.subarray(8)])); + assert.equal(rest.nextChunk, 3); + const replay = await bucket.appendChunks(started.uploadId, 0, Readable.from([payload.subarray(0, 8)])); + assert.equal(replay.nextChunk, 3); + const info = await bucket.finalizeResumableUpload(started.uploadId); + assert.equal(info.length, payload.length); + }); + + it('resumes a Readable append from expectedNextChunk after one full chunk then a trailing partial', async () => { + const { bucket, db } = bucketFor({ driver: false }); + const payload = Buffer.from('aaaabbbb'); + const started = await bucket.beginResumableUpload('resume.bin', { + length: payload.length, sha256: sha(payload), chunkSizeBytes: 4, + }); + await assert.rejects( + () => bucket.appendChunks(started.uploadId, 0, Readable.from([payload.subarray(0, 4), Buffer.from('bb')])), + (error: unknown) => codeOf(error) === 'INVALID_ARGUMENT', + ); + assert.deepEqual( + (await db.collection('fs.chunks').find({ files_id: started.uploadId }).sort({ n: 1 }).toArray()).map((chunk) => chunk.n), + [0], + ); + assert.equal((await db.collection('fs.uploads').findOne({ _id: started.uploadId }))?.activeAppend, undefined); + await assert.rejects( + () => bucket.appendChunks(started.uploadId, 2, payload.subarray(4)), + (error: unknown) => codeOf(error) === 'GRIDFS_RESUME_MISMATCH' && detailsOf(error).expectedNextChunk === 1, + ); + const resumed = await bucket.appendChunks(started.uploadId, 1, payload.subarray(4)); + assert.equal(resumed.nextChunk, 2); + assert.equal((await bucket.finalizeResumableUpload(started.uploadId)).length, 8); + }); }); diff --git a/test/unit/gridfs/stream-chunks.test.ts b/test/unit/gridfs/stream-chunks.test.ts new file mode 100644 index 0000000..776b5eb --- /dev/null +++ b/test/unit/gridfs/stream-chunks.test.ts @@ -0,0 +1,261 @@ +import { Readable } from 'node:stream'; +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { consumeReadableChunks } from '../../../src/capabilities/gridfs/stream-chunks'; + +function codeOf(error: unknown): string | undefined { + return error && typeof error === 'object' && 'code' in error ? String((error as { code: unknown }).code) : undefined; +} + +function delayed(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function settles(promise: Promise, ms = 1000): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`did not settle within ${ms}ms`)), ms); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + if (timer) clearTimeout(timer); + } +} + +function isClosedError(error: unknown): boolean { + return codeOf(error) === 'INVALID_ARGUMENT' && String((error as Error).message).includes('closed before the upload completed'); +} + +function readableFromBuffer(data: Buffer, highWaterMark: number): Readable { + let offset = 0; + return new Readable({ + highWaterMark, + read() { + if (offset >= data.length) { + this.push(null); + return; + } + const size = Math.min(highWaterMark, data.length - offset); + const slice = data.subarray(offset, offset + size); + offset += size; + this.push(slice); + }, + }); +} + +describe('consumeReadableChunks', () => { + it('keeps peak buffered memory from scaling with input size', async () => { + async function measure(size: number): Promise<{ peak: number; bytesAccepted: number }> { + const chunkSize = 1024; + const highWaterMark = 256; + const data = Buffer.alloc(size, 7); + let peak = 0; + const result = await consumeReadableChunks( + readableFromBuffer(data, highWaterMark), + 0, + size, + chunkSize, + async () => { await delayed(1); }, + (bytes) => { peak = Math.max(peak, bytes); }, + ); + return { peak, bytesAccepted: result.bytesAccepted }; + } + + const small = await measure(64 * 1024); + const large = await measure(1024 * 1024); + assert.equal(small.bytesAccepted, 64 * 1024); + assert.equal(large.bytesAccepted, 1024 * 1024); + const bound = 1024 + 256 + 64; + assert.ok(small.peak <= bound, `small peak ${small.peak} exceeded ${bound}`); + assert.ok(large.peak <= bound, `large peak ${large.peak} exceeded ${bound}`); + assert.ok( + Math.abs(large.peak - small.peak) <= 1024, + `peak scaled with input: small=${small.peak} large=${large.peak}`, + ); + }); + + it('applies backpressure and does not pull the whole payload during a slow chunk write', async () => { + const chunkSize = 32; + const total = 32 * 40; + const data = Buffer.alloc(total, 3); + let offset = 0; + let reads = 0; + let writing = false; + let producedWhileWriting = 0; + const source = new Readable({ + highWaterMark: 16, + read() { + reads += 1; + if (offset >= data.length) { + this.push(null); + return; + } + const size = Math.min(16, data.length - offset); + const slice = data.subarray(offset, offset + size); + offset += size; + if (writing) producedWhileWriting += size; + this.push(slice); + }, + }); + let writes = 0; + await consumeReadableChunks(source, 0, total, chunkSize, async () => { + writes += 1; + if (writes === 1) { + writing = true; + await delayed(25); + writing = false; + } + }); + assert.equal(writes, 40); + assert.ok(reads > 1, `expected multiple reads, got ${reads}`); + assert.ok( + producedWhileWriting <= 16 * 4, + `first write pulled ${producedWhileWriting} additional bytes; expected prefetch only, not the full ${total}-byte payload`, + ); + assert.ok(offset === total, 'stream should eventually be consumed'); + }); + + it('rejects a trailing partial after emitting complete chunks, and rejects oversize input', async () => { + const written: number[] = []; + await assert.rejects( + () => consumeReadableChunks( + Readable.from([Buffer.from('abcd'), Buffer.from('xy')]), + 0, + 8, + 4, + async (slice) => { written.push(slice.index); }, + ), + (error: unknown) => codeOf(error) === 'INVALID_ARGUMENT', + ); + assert.deepEqual(written, [0]); + + const oversized: number[] = []; + await assert.rejects( + () => consumeReadableChunks( + Readable.from([Buffer.from('abcdefgh')]), + 0, + 4, + 4, + async (slice) => { oversized.push(slice.bytes.length); }, + ), + (error: unknown) => codeOf(error) === 'INVALID_ARGUMENT', + ); + assert.deepEqual(oversized, [4]); + }); + + it('surfaces an onChunk failure after earlier slices have been emitted', async () => { + const written: number[] = []; + await assert.rejects( + () => consumeReadableChunks( + Readable.from([Buffer.from('aaaabbbbcccc')]), + 0, + 12, + 4, + async (slice) => { + written.push(slice.index); + if (slice.index === 1) throw new Error('write failed'); + }, + ), + /write failed/, + ); + assert.deepEqual(written, [0, 1]); + }); + + it('rejects a close-only destroy without hanging', async () => { + const source = new Readable({ read() { /* never ends */ } }); + const consumption = consumeReadableChunks(source, 0, 8, 4, async () => undefined); + setImmediate(() => source.destroy()); + await assert.rejects(() => settles(consumption), isClosedError); + assert.equal(source.listenerCount('readable'), 0); + assert.equal(source.listenerCount('close'), 0); + }); + + it('rejects a stream that is already destroyed before it is passed in', async () => { + const source = new Readable({ read() { /* never ends */ } }); + source.destroy(); + assert.equal(source.destroyed, true); + assert.equal(source.readableEnded, false); + await assert.rejects( + () => settles(consumeReadableChunks(source, 0, 8, 4, async () => undefined)), + isClosedError, + ); + }); + + it('handles an asynchronous destroy error after the append settles', async () => { + let pushed = false; + let finishDestroy = (): void => {}; + const failure = new Error('destroy failed'); + const source = new Readable({ + read() { + if (pushed) return; + pushed = true; + this.push(Buffer.from('aaaabbbb')); + }, + destroy(error, callback) { + finishDestroy = () => callback(error); + }, + }); + try { + await assert.rejects(settles(consumeReadableChunks(source, 0, 8, 4, async () => { + source.destroy(failure); + })), (actual) => actual === failure); + } finally { + finishDestroy(); + await new Promise((resolve) => setImmediate(resolve)); + } + assert.equal(source.listenerCount('error'), 0); + assert.equal(source.listenerCount('close'), 0); + }); + + it('removes its listeners after silent asynchronous destruction completes', async () => { + let finishDestroy = (): void => {}; + const source = new Readable({ + emitClose: false, + read() {}, + destroy(error, callback) { + finishDestroy = () => callback(error); + }, + }); + const result = consumeReadableChunks(source, 0, 4, 4, async () => undefined); + setImmediate(() => source.destroy()); + try { + await assert.rejects(settles(result), isClosedError); + assert.equal(source.closed, false); + await new Promise((resolve) => setImmediate(resolve)); + } finally { + finishDestroy(); + } + await delayed(100); + assert.equal(source.closed, true); + assert.equal(source.listenerCount('error'), 0); + assert.equal(source.listenerCount('close'), 0); + }); + + it('waits for the in-flight write and discards the remaining buffered chunks on destroy', async () => { + let writeFinished = false; + let pushed = false; + const startedWrites: number[] = []; + const source = new Readable({ + read() { + if (!pushed) { + pushed = true; + this.push(Buffer.from('aaaabbbbcccc')); + } + }, + }); + await assert.rejects( + () => settles(consumeReadableChunks(source, 0, 12, 4, async ({ index }) => { + startedWrites.push(index); + if (index === 0) { + source.destroy(); + await delayed(40); + writeFinished = true; + } + })), + isClosedError, + ); + assert.equal(writeFinished, true); + assert.deepEqual(startedWrites, [0]); + }); +}); diff --git a/website/site-budget.json b/website/site-budget.json index fee3213..624e311 100644 --- a/website/site-budget.json +++ b/website/site-budget.json @@ -4,7 +4,10 @@ "maximumTotalBytes": 68000000, "maximumSingleJavaScriptBytes": 740000, "maximumSingleHtmlBytes": 550000, - "maximumSearchIndexBytesPerLocale": 1800000, + "maximumSearchIndexBytesPerLocale": 1804000, "maximumCombinedSearchIndexBytes": 3500000, - "requiredSearchLocales": ["en", "zh"] + "requiredSearchLocales": [ + "en", + "zh" + ] }