skills/docs/foundation/flow/streams-chunk-queue/src/queue/streaming-chunk-queue.md

streaming-chunk-queue

Purpose

StreamingChunkQueue is the public class that reconstitutes out-of-order chunked file transfers into a sequential ReadableStream<Uint8Array>. It composes the package's pluggable validator, decryptor, persister, rate limiter, hooks, event emitter, and OTel telemetry into a single coherent reception primitive — the piece every chunked-file consumer (drive-mesh, drive-webrtc, drive-transport-https) plugs into when chunks arrive via a jittery or out-of-order transport.

Features

• Out-of-order ingestion: chunks may arrive in any order; duplicates are silently ignored.
• Sequential output: the readable stream emits chunks strictly in ascending chunkId order.
• Auto-start: the first addChunk automatically materialises the stream (configurable via autoStart: false).
• Pluggable validator + decryptor strategies with adapters for @suaveplan/packer ChunkVerifier.
• Optional TokenBucketRateLimiterAdapter integration for bandwidth throttling without dropping chunks.
• Optional StorageChunkPersister for resumable transfers.
• Eight lifecycle hooks, ten event types, and wildcard listener support.
• Binary chunk-request protocol exposed through requestChunks and requestMissing.
• Missing-chunk timeout that errors the readable stream after chunkTimeout ms (default 30 000).
• Frozen status snapshot via getStatus.
• OTel counters and a histogram (stream_duration_ms).

Usage: Basic

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

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

await queue.addChunk(0n, bytes0);
await queue.addChunk(4n, bytes4);
await queue.addChunk(1n, bytes1);
await queue.addChunk(2n, bytes2);
await queue.addChunk(3n, bytes3);

const reader = queue.getStream().getReader();
for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    process(value);
}

Usage: Advanced

import { ChunkVerifier } from "@suaveplan/packer";
import { TokenBucket } from "@suaveplan/rate-limiter";
import { MemoryStorage } from "@suaveplan/storage";
import {
    ChunkVerifierValidator,
    StorageChunkPersister,
    StreamingChunkQueue,
    TokenBucketRateLimiterAdapter,
} from "@suaveplan/streams-chunk-queue";

const verifier = new ChunkVerifier(serverPubKey);
const bucket = new TokenBucket({ capacity: 1_048_576, refillRate: 1024 });
const storage = new MemoryStorage();

const queue = new StreamingChunkQueue({
    fileId: 12345n,
    totalChunks: 1024n,
    validator: new ChunkVerifierValidator(verifier),
    rateLimiter: new TokenBucketRateLimiterAdapter(bucket),
    persister: new StorageChunkPersister({ storage, fileId: 12345n }),
    chunkTimeout: 60_000,
    hooks: {
        onStreamProgress: (pct) => statusBar.set(pct),
        onError: (err) => toaster.error(err.message),
    },
});

queue.addEventListener("*", (event) =>
    telemetry.record("chunk_queue.event", { type: event.type }),
);

const stream = queue.getStream();
await stream.pipeTo(diskWriter);

API Reference

new StreamingChunkQueue(options)

options is StreamingChunkQueueOptions. The constructor throws StreamsChunkQueueError(INVALID_TOTAL_CHUNKS) when totalChunks <= 0n.

addChunk(chunkId, data) → Promise<void>

Persists, validates, decrypts, and enqueues the chunk. Throws INVALID_CHUNK_ID on out-of-range ids, or VALIDATION_FAILED / DECRYPTION_FAILED if the pipeline rejects.

getStream() → ReadableStream<Uint8Array>

Idempotent: subsequent calls return the same stream.

requestChunks(ids?) / requestMissing() → Uint8Array

Encode the binary chunk-request frame.

getMissingChunkIds() → bigint[]

Ascending list of un-received chunk ids.

getStatus() → StreamingQueueStatus

Frozen snapshot.

loadChunk(chunkId) / loadAllChunks() → Promise<void>

Hydrate from the configured persister. Throw STORAGE_NOT_CONFIGURED / STORAGE_CHUNK_MISSING on failure.

isAllChunksReceived() / isStreamingComplete() → boolean

addEventListener(type, listener) / removeEventListener(type, listener)

type is any StreamEvent['type'] or '*'.

clear() → void

Reset internal state for reuse.

Implementation Notes

The validator and decryptor stages are extracted to pipeline.ts so the queue file stays close to the per-file line budget. Storage guards live in storage-ops.ts for the same reason. Rate-limiting denial is non-destructive — the queue sleeps for the bucket's retryAfter (or the adapter's fallback delay) and retries the same chunk; no bytes are ever dropped. Missing-chunk timeout polls the in-memory map at a configurable interval (default 100 ms) so deterministic tests can inject a custom clock + sleep.