From 43850aaed3aed330aae85d075e1b1ffe577f31ae Mon Sep 17 00:00:00 2001 From: rocky Date: Thu, 12 Nov 2026 17:57:38 +0800 Subject: [PATCH 1/4] fix(gridfs): close remaining upload races and stream append safely Re-check files after beginResumableUpload inserts a session and never delete chunks of a file that became visible in that window. Roll a finalizing session back to pending after GRIDFS_FILE_EXISTS or GRIDFS_INTEGRITY_MISMATCH so public abort works. Align renameFile release-owner handling with upload so a cleanup failure cannot mask the real result. Re-read visibility after a recovery claim and fail closed when verifyVisibleFile cannot confirm a required visible file. Consume Readable appends incrementally (design A): claim first, honor backpressure, buffer only the residual plus the in-flight chunk, and reject trailing partial or oversize input during consumption. Add an inspect-only, paginated orphan reporter with no apply or delete path. Clarify buffering, large-file resumable guidance, and the caller-persisted verification contract in the English and Chinese docs. Record Search/Vector Search Atlas deployments as UNVERIFIED. --- docs/en/gridfs-recovery.md | 9 +- docs/en/gridfs.md | 10 +- docs/en/support-matrix.md | 2 +- docs/zh/gridfs-recovery.md | 10 ++ docs/zh/gridfs.md | 10 +- docs/zh/support-matrix.md | 2 +- .../maintenance/gridfs-inspect-orphans.cjs | 158 ++++++++++++++++++ scripts/maintenance/gridfs-recover.cjs | 17 +- .../validation/run-memory-server-matrix.cjs | 2 + src/capabilities/gridfs/operations.ts | 16 +- src/capabilities/gridfs/resumable.ts | 110 +++++++++--- src/capabilities/gridfs/stream-chunks.ts | 139 +++++++++++++++ .../gridfs/gridfs-inspect-orphans.test.ts | 120 +++++++++++++ .../gridfs/gridfs-remaining-issues.test.ts | 138 +++++++++++++++ .../gridfs/gridfs-stale-recovery.test.ts | 66 ++++++++ test/run-tests.cjs | 3 + test/unit/gridfs/gridfs.test.ts | 119 +++++++++++++ test/unit/gridfs/stream-chunks.test.ts | 148 ++++++++++++++++ 18 files changed, 1036 insertions(+), 43 deletions(-) create mode 100644 scripts/maintenance/gridfs-inspect-orphans.cjs create mode 100644 src/capabilities/gridfs/stream-chunks.ts create mode 100644 test/integration/gridfs/gridfs-inspect-orphans.test.ts create mode 100644 test/integration/gridfs/gridfs-remaining-issues.test.ts create mode 100644 test/unit/gridfs/stream-chunks.test.ts diff --git a/docs/en/gridfs-recovery.md b/docs/en/gridfs-recovery.md index 5409cf5..9031b07 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 unknown-id leftover chunks without deleting anything, use the inspect-only reporter `scripts/maintenance/gridfs-inspect-orphans.cjs`. It walks `.chunks` in index order with range pagination, re-checks `files` and `uploads` for each `files_id` at classification time, 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. + +```sh +export MONSQLIZE_RECOVERY_MONGODB_URI='' +node scripts/maintenance/gridfs-inspect-orphans.cjs --db=my_database --bucket=fs --page-size=200 +``` ## Deployment and rollback gates (required) diff --git a/docs/en/gridfs.md b/docs/en/gridfs.md index 85d6d5f..65e395e 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 that already exists in `files` or `uploads` throws `GRIDFS_FILE_EXISTS`. Those prechecks are not a lock: the session insert is the atomic id claim. After the insert, `files` is checked again. If a file became visible in that window, the new pending session is dropped (best effort), chunks of the visible file are never deleted, and the call throws `GRIDFS_FILE_EXISTS`. 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. Retry from the reported `nextChunk`. Peak memory is the residual (shorter than the next expected chunk) plus the current incoming stream buffer, not the full payload. Use a modest `highWaterMark` for large files. `appendChunks` is addressed by `fromChunk`: @@ -94,7 +94,9 @@ Chunk bytes must follow the session size: `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. -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 and the file length match 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`. 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. 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..1e0ada4 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` 所有权。 + +排查未知 ID 的残留 chunks 时，使用只读巡检脚本 `scripts/maintenance/gridfs-inspect-orphans.cjs`。它按已有 `{ files_id, n }` 索引分页扫描 `.chunks`，在分类时重新核对 `files` 和 `uploads`，报告 `visible_file`、`owned` 或 `candidate_orphan`。没有 apply 模式，不会删除，也不会按文档年龄删除。它不是公开的 `GridFSBucket` API。 + +```sh +export MONSQLIZE_RECOVERY_MONGODB_URI='<从密钥管理系统读取的 MongoDB URI>' +node scripts/maintenance/gridfs-inspect-orphans.cjs --db=database --bucket=fs --page-size=200 +``` + + ## 上线和回滚步骤 1. 冻结所有 GridFS 写入，禁止新请求和自动重启，核查所有旧版 Writer 已排空；保留相关运行实例、进程、队列、版本和时间证据。 diff --git a/docs/zh/gridfs.md b/docs/zh/gridfs.md index e08df52..50d11b2 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`。预检查不是锁，会话插入才是原子占位。插入之后会再次检查 `files`。若在这个窗口里文件已经可见，会尽量删除刚插入的 pending 会话，**绝不会**删除可见文件的 chunks，并抛出 `GRIDFS_FILE_EXISTS`。 分片长度必须符合会话： - 除最后一片外，每片长度都是 `chunkSizeBytes`。 - 最后一片长度是 `length % chunkSizeBytes`；能够整除时仍是 `chunkSizeBytes`。 -- 无法按这些长度切开的 payload 抛出 `INVALID_ARGUMENT`。整段 payload 会在插入任何分片之前完成检查。 +- 无法按这些长度切开的 payload 抛出 `INVALID_ARGUMENT`。对 `Buffer` / `Uint8Array`，整段 payload 会在插入任何分片之前完成规划与检查。对 `Readable`，完整分片会随读随写；末尾不足或超出的残余会在消费过程中被拒绝，此前已经写入的完整前缀分片会保留，调用方应从返回的 `nextChunk` 重试。峰值内存是“下一片所需长度以内的残余 + 当前流入的 stream 缓冲”，不会随整个 payload 线性增长。大文件应使用较小的 `highWaterMark`。 `appendChunks` 以 `fromChunk` 定位： @@ -94,7 +94,9 @@ await bucket.abortResumableUpload(session.uploadId); `finalizeResumableUpload` 要求分片从 0 连续、总长度等于会话长度，且 sha256 等于会话摘要。然后插入 files 文档、写入 `metadata.sha256`，并删除会话。缺少的分片序号抛出 `GRIDFS_UPLOAD_INCOMPLETE`，`details[0].missingChunks` 列出缺口。摘要不符或分片长度不对抛出 `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`。会话已不存在时（上一次 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。 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..8863a0f --- /dev/null +++ b/scripts/maintenance/gridfs-inspect-orphans.cjs @@ -0,0 +1,158 @@ +'use strict'; + +/** + * Operator-only, inspect-only reporter for GridFS chunk files_id groups. + * Walks .chunks via the existing { files_id, n } index in range pages. + * 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 + * + * 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] + */ +const { MongoClient, ObjectId } = require('mongodb'); + +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 file = await files.findOne({ _id: group.id }, { projection: { _id: 1 } }); + const owner = await uploads.findOne( + { _id: group.id }, + { projection: { _id: 1, kind: 1, status: 1 } }, + ); + let classification; + if (file) classification = 'visible_file'; + else if (owner) classification = 'owned'; + else classification = 'candidate_orphan'; + const finding = { + filesId: group.id instanceof ObjectId ? group.id.toHexString() : String(group.id), + 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; +} + +async function inspectGridFSOrphans(input) { + const { db, bucket, pageSize = 200 } = 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'); + } + if (input.apply === true) fail('This inspector has no apply mode and never deletes'); + + const chunks = db.collection(bucket + '.chunks'); + const findings = []; + let lastId = null; + let lastN = -1; + let current = null; + let pages = 0; + + const flush = async (group) => { + findings.push(await classifyFilesId(db, bucket, group)); + }; + + for (;;) { + const filter = lastId == null + ? {} + : { + $or: [ + { files_id: { $gt: lastId } }, + { files_id: lastId, n: { $gt: lastN } }, + ], + }; + const page = await chunks.find(filter) + .sort({ files_id: 1, n: 1 }) + .project({ files_id: 1, n: 1 }) + .limit(pageSize) + .toArray(); + pages += 1; + if (page.length === 0) break; + for (const doc of page) { + const filesId = doc.files_id; + if (!current || !sameId(current.id, filesId)) { + if (current) await flush(current); + current = { id: filesId, 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; + } + lastId = filesId; + lastN = doc.n; + } + if (page.length < pageSize) break; + } + if (current) await flush(current); + + const countOf = (name) => findings.filter((row) => row.classification === name).length; + return { + bucket, + pageSize, + pages, + filesIdsScanned: findings.length, + visibleFiles: countOf('visible_file'), + owned: countOf('owned'), + candidateOrphans: countOf('candidate_orphan'), + findings, + }; +} + +function sameId(left, right) { + if (left === right) return true; + if (left && right && typeof left.equals === 'function') return left.equals(right); + return String(left) === String(right); +} + +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 client = new MongoClient(uri); + await client.connect(); + try { + const report = await inspectGridFSOrphans({ db: client.db(dbName), bucket, pageSize }); + process.stdout.write(JSON.stringify(report, null, 2) + '\n'); + } finally { + await client.close(); + } +} + +module.exports = { inspectGridFSOrphans }; +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..aad5fda 100644 --- a/scripts/maintenance/gridfs-recover.cjs +++ b/scripts/maintenance/gridfs-recover.cjs @@ -243,6 +243,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 }, { projection: { _id: 1 } }); + 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,7 +258,7 @@ 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) }, @@ -263,16 +270,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) { + 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/operations.ts b/src/capabilities/gridfs/operations.ts index f425cd8..efc375a 100644 --- a/src/capabilities/gridfs/operations.ts +++ b/src/capabilities/gridfs/operations.ts @@ -291,9 +291,19 @@ export async function renameFile(ctx: GridFSContext, id: ObjectId, filename: str } 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; } } diff --git a/src/capabilities/gridfs/resumable.ts b/src/capabilities/gridfs/resumable.ts index bb3245a..dc0f885 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,7 @@ import { toNodeBuffer, withContentType, } from './validate'; -import { readReadable } from './bytes'; +import { consumeReadableChunks } from './stream-chunks'; export async function beginResumableUpload(ctx: GridFSContext, filename: string, opts: BeginResumableUploadOptions): Promise { assertFilename(filename, 'beginResumableUpload'); @@ -66,9 +66,18 @@ export async function beginResumableUpload(ctx: GridFSContext, filename: string, } 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. + // The insert is the atomic id claim. A concurrent upload can still make a + // files document visible after the prechecks and even after this insert. + // Re-check files and fail closed: never delete chunks of a visible file. try { + if (await files.findOne({ _id: uploadId }, { projection: { _id: 1 } })) { + try { + await uploads.deleteOne({ _id: uploadId, kind: 'resumable', status: 'pending' }); + } catch { /* Best-effort session cleanup; chunks of the visible file stay. */ } + throw fileExists('beginResumableUpload', ctx, { id: uploadId, filename, sha256: opts.sha256 }); + } + // 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. await ctx.db.collection(chunksCollectionName(ctx.bucketName)).deleteMany({ files_id: uploadId }); } catch (error) { throw mapDriverError(error, 'beginResumableUpload', ctx, { uploadId, filename }); @@ -84,8 +93,15 @@ export async function appendChunks( ): Promise { assertObjectId(uploadId, 'appendChunks', 'uploadId'); assertNonNegativeInteger(fromChunk, 'appendChunks', 'fromChunk'); - const payload = await payloadBytes(data); - if (payload.length === 0) throw badChunkSize(); + const isStream = isReadableStream(data); + if (!isStream && !Buffer.isBuffer(data) && !(data instanceof Uint8Array)) { + throw invalidArgument('appendChunks', 'data', 'data must be a Buffer, Uint8Array, or Readable'); + } + let payload: Buffer | undefined; + if (!isStream) { + payload = toNodeBuffer(data as Buffer | Uint8Array); + if (payload.length === 0) throw badChunkSize(); + } const uploads = ctx.db.collection(uploadsCollectionName(ctx.bucketName)); const token = randomUUID(); // A durable single-append claim excludes finalize, abort, and concurrent @@ -111,14 +127,34 @@ export async function appendChunks( 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 (isStream) { + const streamed = await consumeReadableChunks( + data as Readable, + fromChunk, + session.length, + session.chunkSizeBytes, + writeSlice, + ); + bytesAccepted = streamed.bytesAccepted; + } else { + const slices = planChunkSlices(payload as Buffer, 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 = (payload as Buffer).length; } const advanced = await contiguousNext(ctx, uploadId, nextChunk); const completed = await uploads.updateOne( @@ -130,7 +166,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 @@ -169,8 +205,24 @@ export async function finalizeResumableUpload(ctx: GridFSContext, uploadId: Obje // fail its pending -> aborting CAS instead of deleting validated chunks. session = await requireSession(ctx, uploadId, 'finalizeResumableUpload'); const files = ctx.db.collection(filesCollectionName(ctx.bucketName)); - const committed = await committedFile(ctx, uploadId, session); - if (committed) return committed; + + const restorePendingAfterClosedFailure = async (error: unknown): Promise => { + if (!isClosedFinalizeFailure(error)) return; + 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. */ } + }; + + try { + const already = await committedFile(ctx, uploadId, session); + if (already) return already; + } catch (error) { + await restorePendingAfterClosedFailure(error); + throw error; + } try { const expected = chunkCount(session.length, session.chunkSizeBytes); @@ -237,12 +289,21 @@ export async function finalizeResumableUpload(ctx: GridFSContext, uploadId: Obje }); } 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) { + await restorePendingAfterClosedFailure(conflict); + throw conflict; + } + } + const mapped = mapDriverError(error, 'finalizeResumableUpload', ctx, { uploadId, filename: session.filename }); + if (isClosedFinalizeFailure(mapped)) { + await restorePendingAfterClosedFailure(mapped); } - // 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' }); @@ -346,10 +407,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() { diff --git a/src/capabilities/gridfs/stream-chunks.ts b/src/capabilities/gridfs/stream-chunks.ts new file mode 100644 index 0000000..f3e1ba8 --- /dev/null +++ b/src/capabilities/gridfs/stream-chunks.ts @@ -0,0 +1,139 @@ +import type { Readable } from 'node:stream'; +import { invalidArgument } from './errors'; +import { chunkCount, chunkLengthAt } from './validate'; + +export interface StreamedChunk { + index: number; + bytes: Buffer; +} + +function badChunkSize(): never { + throw invalidArgument('appendChunks', 'data', 'chunk payload must match the session chunk size'); +} + +/** + * 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. + */ +export async function consumeReadableChunks( + source: Readable, + fromChunk: number, + length: number, + chunkSizeBytes: number, + onChunk: (slice: StreamedChunk) => Promise, + onBufferedBytes?: (bytes: number) => void, +): Promise<{ bytesAccepted: number }> { + const total = chunkCount(length, chunkSizeBytes); + let residual = Buffer.alloc(0); + let index = fromChunk; + let bytesAccepted = 0; + let ended = false; + 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 (index < total && offset < incoming.length) { + 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 }); + } + if (index >= total && offset < incoming.length) badChunkSize(); + report(); + }; + + await new Promise((resolve, reject) => { + const finish = (error?: unknown): void => { + if (settled) return; + settled = true; + if (error) reject(error); + else resolve(); + }; + + const pump = async (): Promise => { + if (settled || busy) return; + busy = true; + try { + let incoming: Buffer | string | null; + while ((incoming = source.read() as Buffer | string | null) !== null) { + await takeFrom(Buffer.isBuffer(incoming) ? incoming : Buffer.from(incoming)); + } + 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; + } + pump().then(() => { + if (settled) return; + if (pending) { + pending = false; + run(); + return; + } + // Only the pump that observed a clean end may resolve. A stale + // success handler must not hide a later residual/size failure. + if ((ended || source.readableEnded) && residual.length === 0) finish(); + }, (error: unknown) => finish(error)); + }; + + source.on('readable', run); + source.on('end', () => { + ended = true; + run(); + }); + source.on('error', (error) => finish(error)); + run(); + if (!ended && source.readableEnded) { + ended = true; + run(); + } + }).finally(() => { + if (!source.destroyed && !source.readableEnded) source.destroy(); + }); + + if (bytesAccepted === 0) badChunkSize(); + return { bytesAccepted }; +} 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..d7ee58f --- /dev/null +++ b/test/integration/gridfs/gridfs-inspect-orphans.test.ts @@ -0,0 +1,120 @@ +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 { 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('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; + (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 (filter?._id && filter._id.equals(ids[0])) { + 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; + } + }); + }); +}); 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..0f7fe2a --- /dev/null +++ b/test/integration/gridfs/gridfs-remaining-issues.test.ts @@ -0,0 +1,138 @@ +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'); +} + +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('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('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); + }); + }); +}); diff --git a/test/integration/gridfs/gridfs-stale-recovery.test.ts b/test/integration/gridfs/gridfs-stale-recovery.test.ts index 0f95c7f..a0426ba 100644 --- a/test/integration/gridfs/gridfs-stale-recovery.test.ts +++ b/test/integration/gridfs/gridfs-stale-recovery.test.ts @@ -547,4 +547,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..d79a94c 100644 --- a/test/unit/gridfs/gridfs.test.ts +++ b/test/unit/gridfs/gridfs.test.ts @@ -204,6 +204,7 @@ function bucketFor(options: { driver?: boolean; instanceId?: string; poolName?: string; + logger?: { warn: (...args: unknown[]) => void }; } = {}) { const db = new MemoryDb(); const calls: string[] = []; @@ -267,6 +268,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 +641,121 @@ 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); + }); }); diff --git a/test/unit/gridfs/stream-chunks.test.ts b/test/unit/gridfs/stream-chunks.test.ts new file mode 100644 index 0000000..e093f3f --- /dev/null +++ b/test/unit/gridfs/stream-chunks.test.ts @@ -0,0 +1,148 @@ +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)); +} + +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]); + }); +}); From 7f8bf89cb9d5af0b8d50d0d4d42e79661ad5f9cf Mon Sep 17 00:00:00 2001 From: rocky Date: Thu, 12 Nov 2026 17:57:44 +0800 Subject: [PATCH 2/4] chore(budget): raise runtime bundle ceilings for GridFS remaining fixes The remaining-issue change set exceeded the 890000 / 887000 ceilings (892639 CJS / 889922 ESM in the local 2026-11-12 build). The growth is from the incremental Readable append consumer, the beginResumableUpload post-insert files re-check, finalize pending rollback, and rename release-owner alignment; dependencies are unchanged. Raise to 896000 / 893000 to keep about 3 KiB of reviewed headroom. --- config/package-budget.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/config/package-budget.json b/config/package-budget.json index 19f781d..f0900b8 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": "The previous 890000 / 887000 ceilings were exceeded by the GridFS remaining-issue fix (local 2026-11-12 build: 892639 CJS / 889922 ESM). The added bytes are the incremental Readable append consumer, beginResumableUpload post-insert files re-check, finalize pending rollback, and rename release-owner alignment — no dependency change. Raise to 896000 / 893000, keeping ~3 KiB reviewed headroom. Enforce these ceilings in release preflight.", "artifacts": { - "dist/cjs/index.cjs": 890000, - "dist/esm/index.mjs": 887000, + "dist/cjs/index.cjs": 896000, + "dist/esm/index.mjs": 893000, "dist/cjs/cli/data-task.cjs": 50000 } } From 8519c80ec63ccf40f7209705182d49e25200fc91 Mon Sep 17 00:00:00 2001 From: rocky Date: Thu, 12 Nov 2026 22:26:28 +0800 Subject: [PATCH 3/4] fix(gridfs): settle premature close, bound inspect, and correct resume docs R1-01: consumeReadableChunks now handles already-destroyed streams and premature close (destroy without error). The read promise rejects in finite time, settles once, waits for an in-flight onChunk write before settling, and removes its listeners on every path so activeAppend can be released. R1-03: gridfs-inspect-orphans streams NDJSON findings, caps each run with maxFindings (default 1000), and exposes an exclusive resumeAfter.filesId cursor. Summary counters accumulate incrementally. The tool remains inspect-only. R1-04: English and Chinese GridFS docs no longer say a failed append returns nextChunk. Clients resume from the last successfully acknowledged position, or from GRIDFS_RESUME_MISMATCH.expectedNextChunk when the stored prefix has moved. --- docs/en/gridfs-recovery.md | 4 +- docs/en/gridfs.md | 2 +- docs/zh/gridfs-recovery.md | 4 +- docs/zh/gridfs.md | 2 +- .../maintenance/gridfs-inspect-orphans.cjs | 118 ++++++++++++++---- src/capabilities/gridfs/stream-chunks.ts | 73 +++++++++-- .../gridfs/gridfs-inspect-orphans.test.ts | 49 ++++++++ .../gridfs/gridfs-remaining-issues.test.ts | 90 +++++++++++++ test/unit/gridfs/gridfs.test.ts | 24 ++++ test/unit/gridfs/stream-chunks.test.ts | 60 +++++++++ 10 files changed, 386 insertions(+), 40 deletions(-) diff --git a/docs/en/gridfs-recovery.md b/docs/en/gridfs-recovery.md index 9031b07..212ff4c 100644 --- a/docs/en/gridfs-recovery.md +++ b/docs/en/gridfs-recovery.md @@ -8,11 +8,11 @@ The ownership registry is the per-bucket `.uploads` collection. A reserv 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 unknown-id leftover chunks without deleting anything, use the inspect-only reporter `scripts/maintenance/gridfs-inspect-orphans.cjs`. It walks `.chunks` in index order with range pagination, re-checks `files` and `uploads` for each `files_id` at classification time, 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. +To survey unknown-id leftover chunks without deleting anything, use the inspect-only reporter `scripts/maintenance/gridfs-inspect-orphans.cjs`. It walks `.chunks` in index order with range pagination, re-checks `files` and `uploads` for each `files_id` at classification time, 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 +node scripts/maintenance/gridfs-inspect-orphans.cjs --db=my_database --bucket=fs --page-size=200 --max-findings=1000 ``` ## Deployment and rollback gates (required) diff --git a/docs/en/gridfs.md b/docs/en/gridfs.md index 65e395e..5a890d3 100644 --- a/docs/en/gridfs.md +++ b/docs/en/gridfs.md @@ -81,7 +81,7 @@ 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`. 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. Retry from the reported `nextChunk`. Peak memory is the residual (shorter than the next expected chunk) plus the current incoming stream buffer, not the full payload. Use a modest `highWaterMark` for large files. +- 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`. Peak memory is the residual (shorter than the next expected chunk) plus the current incoming stream buffer, not the full payload. Use a modest `highWaterMark` for large files. `appendChunks` is addressed by `fromChunk`: diff --git a/docs/zh/gridfs-recovery.md b/docs/zh/gridfs-recovery.md index 1e0ada4..42139d1 100644 --- a/docs/zh/gridfs-recovery.md +++ b/docs/zh/gridfs-recovery.md @@ -10,11 +10,11 @@ dry-run 可以使用领取所有权之前的可见性快照；**apply 必须在独占 claim 之后重新读取 `files`**。若可见性已经变化，或在需要可见文件时 `verifyVisibleFile` 无法确认，脚本失败并保留 `recovering` 所有权。 -排查未知 ID 的残留 chunks 时，使用只读巡检脚本 `scripts/maintenance/gridfs-inspect-orphans.cjs`。它按已有 `{ files_id, n }` 索引分页扫描 `.chunks`，在分类时重新核对 `files` 和 `uploads`，报告 `visible_file`、`owned` 或 `candidate_orphan`。没有 apply 模式，不会删除，也不会按文档年龄删除。它不是公开的 `GridFSBucket` API。 +排查未知 ID 的残留 chunks 时，使用只读巡检脚本 `scripts/maintenance/gridfs-inspect-orphans.cjs`。它按已有 `{ files_id, n }` 索引分页扫描 `.chunks`，在分类时重新核对 `files` 和 `uploads`，报告 `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 +node scripts/maintenance/gridfs-inspect-orphans.cjs --db=database --bucket=fs --page-size=200 --max-findings=1000 ``` diff --git a/docs/zh/gridfs.md b/docs/zh/gridfs.md index 50d11b2..fb19ca2 100644 --- a/docs/zh/gridfs.md +++ b/docs/zh/gridfs.md @@ -81,7 +81,7 @@ await bucket.abortResumableUpload(session.uploadId); - 除最后一片外，每片长度都是 `chunkSizeBytes`。 - 最后一片长度是 `length % chunkSizeBytes`；能够整除时仍是 `chunkSizeBytes`。 -- 无法按这些长度切开的 payload 抛出 `INVALID_ARGUMENT`。对 `Buffer` / `Uint8Array`，整段 payload 会在插入任何分片之前完成规划与检查。对 `Readable`，完整分片会随读随写；末尾不足或超出的残余会在消费过程中被拒绝，此前已经写入的完整前缀分片会保留，调用方应从返回的 `nextChunk` 重试。峰值内存是“下一片所需长度以内的残余 + 当前流入的 stream 缓冲”，不会随整个 payload 线性增长。大文件应使用较小的 `highWaterMark`。 +- 无法按这些长度切开的 payload 抛出 `INVALID_ARGUMENT`。对 `Buffer` / `Uint8Array`，整段 payload 会在插入任何分片之前完成规划与检查。对 `Readable`，完整分片会随读随写；末尾不足或超出的残余会在消费过程中被拒绝，此前已经写入的完整前缀分片会保留。失败的 append 会抛错，不会返回 `nextChunk`。应从上一次已成功确认的位置继续；若已写入前缀已前进，以 `GRIDFS_RESUME_MISMATCH.expectedNextChunk` 为准。峰值内存是“下一片所需长度以内的残余 + 当前流入的 stream 缓冲”，不会随整个 payload 线性增长。大文件应使用较小的 `highWaterMark`。 `appendChunks` 以 `fromChunk` 定位： diff --git a/scripts/maintenance/gridfs-inspect-orphans.cjs b/scripts/maintenance/gridfs-inspect-orphans.cjs index 8863a0f..95c3f87 100644 --- a/scripts/maintenance/gridfs-inspect-orphans.cjs +++ b/scripts/maintenance/gridfs-inspect-orphans.cjs @@ -8,15 +8,22 @@ * - owned: no files document, but an uploads owner/session exists * - candidate_orphan: no files document and no uploads owner at re-check time * - * This tool never deletes, never applies, and never uses document age as a - * deletion criterion. It is not part of the public GridFSBucket API. + * 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] + * --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'; @@ -49,34 +56,72 @@ async function classifyFilesId(db, bucket, group) { return finding; } +function parseAfter(after) { + if (after == null) return null; + const raw = after.filesId ?? after.files_id; + if (raw == null || raw === '') fail('after.filesId is required to resume'); + const filesId = raw instanceof ObjectId ? raw : new ObjectId(String(raw)); + return filesId; +} + async function inspectGridFSOrphans(input) { - const { db, bucket, pageSize = 200 } = 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 = []; - let lastId = null; + const findings = collectFindings ? [] : undefined; + let lastId = afterId; let lastN = -1; let current = null; let pages = 0; + let filesIdsScanned = 0; + let visibleFiles = 0; + let owned = 0; + let candidateOrphans = 0; + let truncated = false; + let resumeAfter = null; + let lastClassifiedId = afterId; const flush = async (group) => { - findings.push(await classifyFilesId(db, bucket, 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); + lastClassifiedId = group.id; + if (filesIdsScanned >= maxFindings) { + const more = await chunks.find({ files_id: { $gt: group.id } }).project({ _id: 1 }).limit(1).next(); + if (more) { + truncated = true; + resumeAfter = { + filesId: group.id instanceof ObjectId ? group.id.toHexString() : String(group.id), + }; + } + } }; for (;;) { const filter = lastId == null ? {} - : { - $or: [ - { files_id: { $gt: lastId } }, - { files_id: lastId, n: { $gt: lastN } }, - ], - }; + : lastN < 0 + ? { files_id: { $gt: lastId } } + : { + $or: [ + { files_id: { $gt: lastId } }, + { files_id: lastId, n: { $gt: lastN } }, + ], + }; const page = await chunks.find(filter) .sort({ files_id: 1, n: 1 }) .project({ files_id: 1, n: 1 }) @@ -87,7 +132,13 @@ async function inspectGridFSOrphans(input) { for (const doc of page) { const filesId = doc.files_id; if (!current || !sameId(current.id, filesId)) { - if (current) await flush(current); + if (current) { + await flush(current); + if (truncated) { + current = null; + break; + } + } current = { id: filesId, minN: doc.n, maxN: doc.n, count: 1 }; } else { current.count += 1; @@ -97,20 +148,23 @@ async function inspectGridFSOrphans(input) { lastId = filesId; lastN = doc.n; } + if (truncated) break; if (page.length < pageSize) break; } - if (current) await flush(current); + if (current && !truncated) await flush(current); - const countOf = (name) => findings.filter((row) => row.classification === name).length; return { bucket, pageSize, + maxFindings, pages, - filesIdsScanned: findings.length, - visibleFiles: countOf('visible_file'), - owned: countOf('owned'), - candidateOrphans: countOf('candidate_orphan'), - findings, + filesIdsScanned, + visibleFiles, + owned, + candidateOrphans, + truncated, + resumeAfter, + ...(findings ? { findings } : {}), }; } @@ -139,17 +193,33 @@ async function main() { 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 client = new MongoClient(uri); await client.connect(); try { - const report = await inspectGridFSOrphans({ db: client.db(dbName), bucket, pageSize }); - process.stdout.write(JSON.stringify(report, null, 2) + '\n'); + const report = await inspectGridFSOrphans({ + db: client.db(dbName), + bucket, + pageSize, + maxFindings, + after: afterFilesId ? { filesId: afterFilesId } : undefined, + 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 }; +module.exports = { inspectGridFSOrphans, DEFAULT_MAX_FINDINGS }; if (require.main === module) { main().catch((error) => { process.stderr.write(String(error.code || 'INSPECT_ERROR') + ': ' + error.message + '\n'); diff --git a/src/capabilities/gridfs/stream-chunks.ts b/src/capabilities/gridfs/stream-chunks.ts index f3e1ba8..553cbe8 100644 --- a/src/capabilities/gridfs/stream-chunks.ts +++ b/src/capabilities/gridfs/stream-chunks.ts @@ -11,11 +11,23 @@ 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; +} + /** * 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 so the caller can release its append claim safely. */ export async function consumeReadableChunks( source: Readable, @@ -30,6 +42,8 @@ export async function consumeReadableChunks( let index = fromChunk; let bytesAccepted = 0; let ended = false; + let aborted = false; + let queuedError: unknown; let busy = false; let settled = false; let pending = false; @@ -76,21 +90,52 @@ export async function consumeReadableChunks( }; await new Promise((resolve, reject) => { + const detach = (): void => { + 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) return; + abort(streamClosedError()); + }; + const pump = async (): Promise => { if (settled || busy) return; busy = true; try { let incoming: Buffer | string | null; - while ((incoming = source.read() as Buffer | string | null) !== null) { + while (!aborted && (incoming = source.read() as Buffer | string | null) !== null) { await takeFrom(Buffer.isBuffer(incoming) ? incoming : Buffer.from(incoming)); } + if (queuedError) throw queuedError; + if (isPrematureClose(source, ended)) throw streamClosedError(); if (ended || source.readableEnded) { ended = true; if (residual.length > 0) badChunkSize(); @@ -106,25 +151,33 @@ export async function consumeReadableChunks( pending = true; return; } + if (queuedError) { + finish(queuedError); + return; + } + if (isPrematureClose(source, ended)) { + finish(streamClosedError()); + return; + } pump().then(() => { if (settled) return; - if (pending) { + if (pending || queuedError || isPrematureClose(source, ended)) { pending = false; run(); return; } - // Only the pump that observed a clean end may resolve. A stale - // success handler must not hide a later residual/size failure. if ((ended || source.readableEnded) && residual.length === 0) finish(); }, (error: unknown) => finish(error)); }; - source.on('readable', run); - source.on('end', () => { - ended = true; - run(); - }); - source.on('error', (error) => finish(error)); + source.on('readable', onReadable); + source.on('end', onEnd); + source.on('error', onError); + source.on('close', onClose); + if (isPrematureClose(source, ended)) { + finish(streamClosedError()); + return; + } run(); if (!ended && source.readableEnded) { ended = true; diff --git a/test/integration/gridfs/gridfs-inspect-orphans.test.ts b/test/integration/gridfs/gridfs-inspect-orphans.test.ts index d7ee58f..6b38317 100644 --- a/test/integration/gridfs/gridfs-inspect-orphans.test.ts +++ b/test/integration/gridfs/gridfs-inspect-orphans.test.ts @@ -117,4 +117,53 @@ describe('GridFS inspect-only orphan reporter', () => { } }); }); + + 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); + const seen = new Set([ + ...first.findings.map((row: { filesId: string }) => row.filesId), + ...rest.findings.map((row: { filesId: string }) => row.filesId), + ...streamed, + ]); + // 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 index 0f7fe2a..ce9b065 100644 --- a/test/integration/gridfs/gridfs-remaining-issues.test.ts +++ b/test/integration/gridfs/gridfs-remaining-issues.test.ts @@ -11,6 +11,10 @@ function sha(data: Buffer): string { return createHash('sha256').update(data).digest('hex'); } +function delayed(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + describe('GridFS remaining-issue regressions (begin race, finalize rollback, streaming append)', () => { const bootstrap = createMemoryServerBootstrap(); let uri = ''; @@ -135,4 +139,90 @@ describe('GridFS remaining-issue regressions (begin race, finalize rollback, str 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('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/unit/gridfs/gridfs.test.ts b/test/unit/gridfs/gridfs.test.ts index d79a94c..c466b43 100644 --- a/test/unit/gridfs/gridfs.test.ts +++ b/test/unit/gridfs/gridfs.test.ts @@ -758,4 +758,28 @@ describe('gridfs bucket', () => { 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 index e093f3f..46f4f8e 100644 --- a/test/unit/gridfs/stream-chunks.test.ts +++ b/test/unit/gridfs/stream-chunks.test.ts @@ -11,6 +11,22 @@ 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({ @@ -145,4 +161,48 @@ describe('consumeReadableChunks', () => { ); 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('waits for an in-flight onChunk write before rejecting a mid-write destroy', async () => { + let writeFinished = false; + let pushed = false; + const source = new Readable({ + read() { + if (!pushed) { + pushed = true; + this.push(Buffer.from('aaaa')); + } + }, + }); + const started = Date.now(); + await assert.rejects( + () => settles(consumeReadableChunks(source, 0, 8, 4, async () => { + source.destroy(); + await delayed(40); + writeFinished = true; + })), + isClosedError, + ); + assert.equal(writeFinished, true); + assert.ok(Date.now() - started >= 40, 'rejected before the in-flight write finished'); + }); }); From af1603053fdb8cf11e4c5a3ddb854d0096576a57 Mon Sep 17 00:00:00 2001 From: rocky Date: Thu, 12 Nov 2026 22:26:31 +0800 Subject: [PATCH 4/4] chore(budget): raise website search-index ceiling for GridFS doc growth After the R1 documentation corrections, the English search index measures 1800511 bytes, 511 over the 1800000 ceiling. CI run 43196147300 already failed at 1800085 on the previous PR head; this round added the failed append resume contract (R1-04) and inspect memory-bound run notes (R1-03) on the GridFS and recovery pages. The Chinese index is 1672743 and the combined total 3473254 remains under 3500000. The growth is from GridFS remaining-issue documentation, not from new website dependencies. The maintainer explicitly authorized raising this ceiling instead of trimming documentation. Raise maximumSearchIndexBytesPerLocale from 1800000 to 1804000 to keep about 3.4 KiB of reviewed headroom. --- website/site-budget.json | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) 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" + ] }