| … | … | @@ -202,12 +202,19 @@ const LEAF_TYPE = Buffer.from([0]) |
| 202 | 202 | */ |
| 203 | 203 | export function createStreamingHasher (opts = {}) { |
| 204 | 204 | const state = Buffer.alloc(sodium.crypto_generichash_STATEBYTES) |
| 205 | | sodium.crypto_generichash_init(state, null, 32) |
| 206 | 205 | let bytes = 0 |
| 207 | | const knownSize = (opts && typeof opts.size === 'number') ? opts.size : null |
| 208 | | if (knownSize !== null) { |
| 209 | | sodium.crypto_generichash_update(state, LEAF_TYPE) |
| 210 | | sodium.crypto_generichash_update(state, c.encode(c.uint64, knownSize)) |
| 206 | if (opts && opts.restore) { |
| 207 | // The saved state already includes the init + leaf/size prefix + every byte |
| 208 | // consumed so far, so do not re-init. |
| 209 | opts.restore.state.copy(state) |
| 210 | bytes = opts.restore.bytes |
| 211 | } else { |
| 212 | sodium.crypto_generichash_init(state, null, 32) |
| 213 | const knownSize = (opts && typeof opts.size === 'number') ? opts.size : null |
| 214 | if (knownSize !== null) { |
| 215 | sodium.crypto_generichash_update(state, LEAF_TYPE) |
| 216 | sodium.crypto_generichash_update(state, c.encode(c.uint64, knownSize)) |
| 217 | } |
| 211 | 218 | } |
| 212 | 219 | return { |
| 213 | 220 | update (buf) { |
| … | … | @@ -219,6 +226,7 @@ export function createStreamingHasher (opts = {}) { |
| 219 | 226 | sodium.crypto_generichash_final(state, out) |
| 220 | 227 | return out.toString('hex') |
| 221 | 228 | }, |
| 229 | snapshot () { return { state: Buffer.from(state), bytes } }, |
| 222 | 230 | get bytes () { return bytes } |
| 223 | 231 | } |
| 224 | 232 | } |
| … | … | @@ -250,6 +258,15 @@ function findCutPoint (data, params, masks) { |
| 250 | 258 | return maxLen |
| 251 | 259 | } |
| 252 | 260 | |
| 261 | // Opt-in memcpy accounting for perf validation. Zero cost when off (one |
| 262 | // boolean check at the increment sites). Module-level singletons — enabled |
| 263 | // only by tests + bench/prepare.js, never in library code; NOT |
| 264 | // concurrency-safe, so do not enable it in production. |
| 265 | let STATS_ON = false |
| 266 | export const chunkStats = { concatBytes: 0, copyBytes: 0, blocks: 0, chunks: 0 } |
| 267 | export function resetChunkStats () { chunkStats.concatBytes = 0; chunkStats.copyBytes = 0; chunkStats.blocks = 0; chunkStats.chunks = 0 } |
| 268 | export function setChunkStats (on) { STATS_ON = !!on; if (on) resetChunkStats() } |
| 269 | |
| 253 | 270 | /** |
| 254 | 271 | * Stream-friendly FastCDC chunker. Consumes an AsyncIterable<Buffer> |
| 255 | 272 | * (e.g. fs.createReadStream) and yields chunks one at a time. Memory |
| … | … | @@ -259,41 +276,45 @@ function findCutPoint (data, params, masks) { |
| 259 | 276 | * |
| 260 | 277 | * Each yielded chunk has the same shape as `chunk()` entries: |
| 261 | 278 | * { hash, offset, length, data } |
| 262 | | * but `data` is a fresh `Buffer.from(slice)` detached from the input |
| 263 | | * stream's pool, so callers can store/forward it without pinning the |
| 264 | | * reader's buffers. |
| 279 | * With the default `copy: true`, `data` is a fresh `Buffer.from(slice)` |
| 280 | * detached from the input stream's pool, so callers can store/forward it. |
| 281 | * With `copy: false`, `data` is a view backed by the reader's memory — |
| 282 | * valid only until the next iteration; the caller MUST consume it |
| 283 | * synchronously and not retain it across an await. |
| 265 | 284 | * |
| 266 | 285 | * Cut-point determinism: same content + same tier → identical chunk |
| 267 | | * boundaries as `chunk()`. Peers using either function on the same |
| 268 | | * file produce matching chunk hashes. |
| 286 | * boundaries as `chunk()`, regardless of input block size. Peers using |
| 287 | * either function on the same file produce matching chunk hashes. |
| 269 | 288 | * |
| 270 | 289 | * @param {AsyncIterable<Buffer>} readable |
| 271 | | * @param {{ tier?: number }} [opts] tier index (0–3). Callers with a |
| 272 | | * known file size should pass `selectTier(size)`. Defaults to tier 0. |
| 290 | * @param {{ tier?: number, copy?: boolean }} [opts] tier index (0–3); copy |
| 291 | * defaults true. Callers with a known file size should pass `selectTier(size)`. |
| 273 | 292 | */ |
| 274 | 293 | export async function * chunkStream (readable, opts = {}) { |
| 275 | 294 | const tier = (opts && typeof opts.tier === 'number') ? opts.tier : 0 |
| 295 | const copy = opts.copy !== false // false → yield views (see JSDoc above) |
| 276 | 296 | const params = TIERS[tier] || TIERS[0] |
| 277 | 297 | const masks = TIER_MASKS[tier] || TIER_MASKS[0] |
| 278 | 298 | |
| 279 | 299 | let pending = Buffer.alloc(0) |
| 280 | 300 | let absoluteOffset = 0 |
| 281 | 301 | |
| 302 | // Single emit point for the stats accounting + copy:false decision. |
| 303 | const emit = (slice) => { |
| 304 | if (STATS_ON) { chunkStats.chunks++; if (copy) chunkStats.copyBytes += slice.length } |
| 305 | return { hash: hashChunk(slice), offset: absoluteOffset, length: slice.length, data: copy ? Buffer.from(slice) : slice } |
| 306 | } |
| 307 | |
| 282 | 308 | for await (const inChunk of readable) { |
| 283 | 309 | const buf = Buffer.isBuffer(inChunk) ? inChunk : Buffer.from(inChunk) |
| 310 | if (STATS_ON) { chunkStats.blocks++; if (pending.length !== 0) chunkStats.concatBytes += pending.length + buf.length } |
| 284 | 311 | pending = pending.length === 0 ? buf : Buffer.concat([pending, buf]) |
| 285 | 312 | |
| 286 | 313 | // While we have at least maxSize ahead, we can definitively find |
| 287 | 314 | // the next cut point inside `pending` (the cut is bounded by maxSize). |
| 288 | 315 | while (pending.length >= params.maxSize) { |
| 289 | 316 | const cutLen = findCutPoint(pending, params, masks) |
| 290 | | const slice = pending.subarray(0, cutLen) |
| 291 | | yield { |
| 292 | | hash: hashChunk(slice), |
| 293 | | offset: absoluteOffset, |
| 294 | | length: cutLen, |
| 295 | | data: Buffer.from(slice) |
| 296 | | } |
| 317 | yield emit(pending.subarray(0, cutLen)) |
| 297 | 318 | absoluteOffset += cutLen |
| 298 | 319 | pending = pending.subarray(cutLen) |
| 299 | 320 | } |
| … | … | @@ -303,23 +324,10 @@ export async function * chunkStream (readable, opts = {}) { |
| 303 | 324 | // the buffer-mode `chunk()`. |
| 304 | 325 | while (pending.length > params.minSize) { |
| 305 | 326 | const cutLen = findCutPoint(pending, params, masks) |
| 306 | | const slice = pending.subarray(0, cutLen) |
| 307 | | yield { |
| 308 | | hash: hashChunk(slice), |
| 309 | | offset: absoluteOffset, |
| 310 | | length: cutLen, |
| 311 | | data: Buffer.from(slice) |
| 312 | | } |
| 327 | yield emit(pending.subarray(0, cutLen)) |
| 313 | 328 | absoluteOffset += cutLen |
| 314 | 329 | pending = pending.subarray(cutLen) |
| 315 | 330 | } |
| 316 | 331 | |
| 317 | | if (pending.length > 0) { |
| 318 | | yield { |
| 319 | | hash: hashChunk(pending), |
| 320 | | offset: absoluteOffset, |
| 321 | | length: pending.length, |
| 322 | | data: Buffer.from(pending) |
| 323 | | } |
| 324 | | } |
| 332 | if (pending.length > 0) yield emit(pending) |
| 325 | 333 | } |