suaveplan/packages/foundation/flow/streams-chunk-queue

@suaveplan/streams-chunk-queue - Universal out-of-order chunk reconstitution: pluggable validation, decryption, rate limiting, storage, hooks, events, binary protocol, ReadableStream output

Universal out-of-order chunk reconstitution. Accepts chunks for a file in any order and exposes a sequential ReadableStream<Uint8Array> consumer surface, composed with pluggable validation, decryption, rate limiting, persistent storage, lifecycle hooks, a wildcard event bus, an efficient binary chunk-request protocol, missing-chunk timeouts, and OpenTelemetry instrumentation.

Documentation tree

• binary-protocol — Produces the chunk-request wire frame consumed by @suaveplan/drive-mesh, drive-transport-https, and any peer that n…
  • encoder — Produces the chunk-request wire frame consumed by @suaveplan/drive-mesh, drive-transport-https, and any peer that n…
• decryptor — Declares the ChunkDecryptor strategy interface and provides the NoopChunkDecryptor default. Decryption is the third…
  • decryptor — Declares the ChunkDecryptor strategy interface and provides the NoopChunkDecryptor default. Decryption is the third…
• errors — StreamsChunkQueueError is the single typed error class raised by every public surface of `@suaveplan/streams-chunk-qu…
  • streams-chunk-queue-error — StreamsChunkQueueError is the single typed error class raised by every public surface of `@suaveplan/streams-chunk-qu…
• events — Implements the lightweight event bus that powers the queue's addEventListener / removeEventListener surface. The em…
  • events — Implements the lightweight event bus that powers the queue's addEventListener / removeEventListener surface. The em…
• hooks — Provides the helper functions that invoke the queue's eight lifecycle callbacks safely. The queue itself wires hook fir…
  • hooks — Provides the helper functions that invoke the queue's eight lifecycle callbacks safely. The queue itself wires hook fir…
• pipeline — Houses the validation + decryption pipeline stages that the StreamingChunkQueue runs on every received chunk. Extract…
• storage-ops — Holds the storage-related guards used by StreamingChunkQueue.loadChunk and loadAllChunks. Both methods need the sam…
• streaming-chunk-queue — StreamingChunkQueue is the public class that reconstitutes out-of-order chunked file transfers into a sequential `Rea…
• rate-limiter — Normalises the @suaveplan/rate-limiter TokenBucket API onto the ChunkRateLimiter interface used internall…
  • adapter — Normalises the @suaveplan/rate-limiter TokenBucket API onto the ChunkRateLimiter interface used internall…
• storage-adapter — Bridges @suaveplan/storage AbstractStorage onto the queue's ChunkPersister contract. Persistence enables resuming…
  • persister — Bridges @suaveplan/storage AbstractStorage onto the queue's ChunkPersister contract. Persistence enables resuming…
• telemetry — Exposes the OpenTelemetry instruments that the StreamingChunkQueue records during its lifecycle. Centralising instrum…
  • telemetry — Exposes the OpenTelemetry instruments that the StreamingChunkQueue records during its lifecycle. Centralising instrum…
• types — Declares the shared types consumed across every feature dir of @suaveplan/streams-chunk-queue. Keeping them in one ty…
  • types — Declares the shared types consumed across every feature dir of @suaveplan/streams-chunk-queue. Keeping them in one ty…
• validator — Declares the ChunkValidator strategy interface and provides the NoopChunkValidator default. Validation is the secon…
  • chunk-verifier-validator — Bridges the structural ChunkVerifier shape from @suaveplan/packer into the canonical ChunkValidator strategy. The…
  • validator — Declares the ChunkValidator strategy interface and provides the NoopChunkValidator default. Validation is the secon…

Install

bun add @suaveplan/streams-chunk-queue

Quick start

import { StreamingChunkQueue } from "@suaveplan/streams-chunk-queue";

const queue = new StreamingChunkQueue({
    fileId: 12345n,
    totalChunks: 5n,
});

await queue.addChunk(2n, chunk2);
await queue.addChunk(0n, chunk0);
await queue.addChunk(1n, chunk1);
await queue.addChunk(3n, chunk3);
await queue.addChunk(4n, chunk4);

await queue.getStream().pipeTo(destination);

Modules

• src/queue/streaming-chunk-queue.md — the public StreamingChunkQueue class.
• src/queue/pipeline.md — validation + decryption stages.
• src/queue/storage-ops.md — storage-side guards.
• src/validator/validator.md — ChunkValidator strategy + NoopChunkValidator.
• src/validator/chunk-verifier-validator.md — adapter for @suaveplan/packer ChunkVerifier.
• src/decryptor/decryptor.md — ChunkDecryptor strategy + NoopChunkDecryptor.
• src/rate-limiter/adapter.md — TokenBucket adapter onto the queue's ChunkRateLimiter.
• src/storage-adapter/persister.md — StorageChunkPersister over AbstractStorage.
• src/binary-protocol/encoder.md — chunk-request wire format.
• src/hooks/hooks.md — safe lifecycle-hook invocation.
• src/events/events.md — wildcard-capable event emitter.
• src/telemetry/telemetry.md — OTel counters + histogram.
• src/errors/streams-chunk-queue-error.md — error codes + class.
• src/types/types.md — shared types.

Features

• Out-of-order ingestion; duplicates silently ignored.
• Sequential ReadableStream<Uint8Array> output.
• Auto-start on first chunk.
• Pluggable validator + decryptor strategies with packer ChunkVerifier adapter.
• Non-destructive rate-limit backoff via TokenBucket adapter.
• Resumable transfers via StorageChunkPersister over AbstractStorage.
• Eight lifecycle hooks, wildcard event bus, 10 event types.
• Binary chunk-request protocol (12-byte header + bitmap).
• Configurable missing-chunk timeout (default 30 s).
• Progress events (0–100) + frozen StreamingQueueStatus snapshot.
• Bigint chunk identifiers (packer v2 u64).
• StreamsChunkQueueError with 8 stable codes.
• OTel: 5 counters + 1 histogram.

License

MIT