SuavePlan/suaveplan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
suaveplan/openspec/specs/streams-chunk-queue/spec.md
admin fbc70f96e6 Initial Commit
2026-05-27 23:29:53 +01:00

13 KiB
Raw Permalink Blame History

Capability Spec Delta: streams-chunk-queue (streams-chunk-queue)

Package: @suaveplan/streams-chunk-queue Status: PROPOSED

Every requirement below carries a **Verified by:** line at authoring time. No exceptions.

ADDED Requirements

Requirement: Universal-tier purity

The @suaveplan/streams-chunk-queue package SHALL be universal-tier: zero imports of browser-only globals (window, document, navigator, localStorage, IndexedDB) and zero imports of Node-only modules (node:fs, node:path, node:crypto, process). Runtime dependencies SHALL be limited to @suaveplan/error, @suaveplan/types, @suaveplan/storage, @suaveplan/rate-limiter, @suaveplan/packer, tslib, zod, and @opentelemetry/api.

Scenario: Source tree references no platform globals

  • WHEN grep -rn "window\|document\|node:fs\|node:path\|node:crypto" packages/foundation/flow/streams-chunk-queue/src/ is run
  • THEN it produces zero matches

Verified by: src/queue/streaming-chunk-queue.test.ts::"is universal-tier and avoids browser+node globals"

Requirement: Bigint chunk identifiers (packer v2 u64)

fileId, totalChunks, chunkId, nextChunkId, and every chunk-id field on emitted events SHALL be bigint.

Scenario: Public API accepts and returns bigint

  • WHEN a caller constructs new StreamingChunkQueue({ fileId: 12345n, totalChunks: 5n }) and calls addChunk(0n, bytes)
  • THEN getStatus().fileId === 12345n, getStatus().nextChunkId is bigint, and getMissingChunkIds() returns bigint[]

Verified by: src/queue/streaming-chunk-queue.test.ts::"exposes bigint identifiers on status and missing-chunk APIs"

Requirement: Structured error registry

The package SHALL export STREAMS_CHUNK_QUEUE_CODES registered via defineCodes from @suaveplan/error, and a StreamsChunkQueueError subclass of SuaveplanError. The registry SHALL include the codes INVALID_TOTAL_CHUNKS, INVALID_CHUNK_ID, STORAGE_NOT_CONFIGURED, STORAGE_CHUNK_MISSING, VALIDATION_FAILED, DECRYPTION_FAILED, STREAM_TIMEOUT, and RATE_LIMIT_EXHAUSTED.

Scenario: Constructor rejects non-positive totalChunks

  • WHEN a caller constructs with totalChunks <= 0n
  • THEN the constructor throws StreamsChunkQueueError whose code === STREAMS_CHUNK_QUEUE_CODES.INVALID_TOTAL_CHUNKS

Scenario: addChunk rejects out-of-range chunkId

  • WHEN a caller calls addChunk(totalChunks, data)
  • THEN the call rejects with StreamsChunkQueueError whose code === STREAMS_CHUNK_QUEUE_CODES.INVALID_CHUNK_ID

Verified by: src/errors/streams-chunk-queue-error.test.ts::"registers every documented STREAMS_CHUNK_QUEUE code" AND src/queue/streaming-chunk-queue.test.ts::"rejects out-of-range chunk ids"

Requirement: Out-of-order chunk ingestion

addChunk(chunkId, data) SHALL accept chunks in any order, deduplicate identical chunk ids, build ChunkMetadata per arrival, emit chunk:received, invoke onChunkReceived, and update progress.

Scenario: Out-of-order assembly

  • WHEN chunks arrive [2, 0, 1, 3, 4]
  • THEN the reader of getStream() receives them in order [0, 1, 2, 3, 4]

Verified by: src/queue/streaming-chunk-queue.test.ts::"deduplicates and assembles out-of-order chunks"

Requirement: ReadableStream consumer surface

getStream() SHALL return a WHATWG ReadableStream<Uint8Array> that outputs chunks in monotonically increasing chunkId order, closes when the last chunk has been streamed, and emits stream:complete plus invokes onStreamComplete. Calling getStream() twice SHALL return the same stream instance.

Scenario: Stream closes on completion

  • WHEN every chunk has been added and read
  • THEN the stream reader observes done: true and isStreamingComplete() returns true

Verified by: src/queue/streaming-chunk-queue.test.ts::"closes the readable stream after the final chunk"

Requirement: Auto-start

When autoStart is unset or true (default), the queue SHALL begin streaming as soon as the first chunk arrives. When autoStart === false, the stream materialises only on getStream().

Scenario: Auto-start materialises the stream on first addChunk

  • WHEN autoStart is left default and addChunk(0n, bytes) is called
  • THEN getStatus().isStreaming flips to true (or isComplete is true if the file has a single chunk)

Verified by: src/queue/streaming-chunk-queue.test.ts::"auto-starts streaming on the first chunk when autoStart defaults to true"

Requirement: Pluggable validator

The package SHALL define ChunkValidator and SHALL ship ChunkVerifierValidator wrapping @suaveplan/packer ChunkVerifier. On validator failure the queue SHALL emit validation:failed, fire onError, and reject addChunk with StreamsChunkQueueError(VALIDATION_FAILED).

Scenario: ChunkVerifier failure propagates a typed error

  • WHEN an injected verifier returns { ok: false }
  • THEN the adapter rejects with StreamsChunkQueueError(VALIDATION_FAILED) with context.chunkId set

Verified by: src/validator/chunk-verifier-validator.test.ts::"surfaces verifier failure as StreamsChunkQueueError" AND src/queue/streaming-chunk-queue.test.ts::"emits validation:failed and rejects when validator throws"

Requirement: Pluggable decryptor

The package SHALL define ChunkDecryptor. On failure the queue SHALL emit decryption:failed and reject with StreamsChunkQueueError(DECRYPTION_FAILED).

Scenario: Decryptor failure propagates a typed error

  • WHEN a decryptor throws
  • THEN addChunk rejects with StreamsChunkQueueError(DECRYPTION_FAILED) and decryption:failed is emitted

Verified by: src/queue/streaming-chunk-queue.test.ts::"emits decryption:failed and rejects when decryptor throws"

Requirement: Rate-limit-aware streaming

When a ChunkRateLimiter is configured, getStream() SHALL call consume(bytes) before enqueueing each chunk. On denial the queue SHALL emit rate:limited, fire onRateLimit, sleep for the returned delay, then retry. Rate-limit admission SHALL NOT drop chunks.

Scenario: Rate-limit denial defers but does not drop

  • WHEN the limiter denies the first request and admits the second
  • THEN the consumer still receives every chunk and at least one rate:limited event is observed

Verified by: src/rate-limiter/adapter.test.ts::"defers and retries when the bucket denies a request" AND src/queue/streaming-chunk-queue.test.ts::"defers streaming under rate-limit pressure without dropping chunks"

Requirement: Persistent storage adapter

The queue SHALL accept an optional ChunkPersister and SHALL ship StorageChunkPersister over AbstractStorage. Every addChunk SHALL persist the chunk under file:<fileId>:chunk:<chunkId>. loadChunk and loadAllChunks SHALL hydrate the queue, throwing STORAGE_NOT_CONFIGURED or STORAGE_CHUNK_MISSING on failure.

Scenario: Round-trip through storage hydrates a new queue

  • WHEN a queue persists three chunks and a fresh queue calls loadAllChunks()
  • THEN getStatus().receivedChunks === 3 on the new queue

Verified by: src/storage-adapter/persister.test.ts::"round-trips chunks through AbstractStorage" AND src/queue/streaming-chunk-queue.test.ts::"hydrates from storage via loadAllChunks"

Requirement: Missing-chunk timeout

If a sequential chunk is still missing after chunkTimeout ms (default 30 000) the readable stream SHALL error with StreamsChunkQueueError(STREAM_TIMEOUT), emit stream:error, and invoke onError.

Scenario: Stream errors when next chunk never arrives

  • WHEN chunkTimeout is small and chunk 1n never arrives after chunk 0n
  • THEN the reader rejects with a StreamsChunkQueueError(STREAM_TIMEOUT)

Verified by: src/queue/streaming-chunk-queue.test.ts::"errors the readable stream after chunkTimeout"

Requirement: Hook system (8 lifecycle hooks)

The queue SHALL accept eight hooks: onChunkReceived, onChunkValidated, onChunkDecrypted, onChunkStreamed, onStreamProgress, onStreamComplete, onError, onRateLimit. Hooks returning a Promise SHALL be awaited; a thrown hook SHALL NOT crash the queue — the error is re-routed via onError.

Scenario: Every hook fires on the matching lifecycle event

  • WHEN a queue with all eight hooks ingests one chunk
  • THEN at minimum onChunkReceived, onChunkValidated, onChunkDecrypted, onChunkStreamed, onStreamProgress, onStreamComplete all fire

Verified by: src/hooks/hooks.test.ts::"safely invokes async and sync hooks and re-routes errors" AND src/queue/streaming-chunk-queue.test.ts::"fires every lifecycle hook"

Requirement: Event system with wildcard

addEventListener(type, listener) and removeEventListener(type, listener) SHALL accept any concrete StreamEvent['type'] or '*'. Wildcard listeners SHALL receive every emitted event.

Scenario: Wildcard listener observes every concrete event

  • WHEN a wildcard listener and a chunk:received listener are attached and a chunk is added
  • THEN both listeners observe the chunk:received event

Verified by: src/events/events.test.ts::"delivers events to wildcard and concrete listeners"

Requirement: Binary chunk-request protocol

requestChunks(chunkIds?) and requestMissing() SHALL return a Uint8Array whose layout is 8 bytes BE fileId u64, 4 bytes BE totalChunks u32, then ⌈totalChunks/8⌉ bytes bitmap where bit i of byte floor(i/8) is 1 iff chunk i is requested.

Scenario: Bitmap encodes the requested chunk ids correctly

  • WHEN encodeChunkRequest({fileId: 12345n, totalChunks: 10n, chunkIds: [0n, 2n, 5n]}) is called
  • THEN byte 12 equals 0b00100101 and byte 13 equals 0x00

Verified by: src/binary-protocol/encoder.test.ts::"encodes file id, total chunks, and bitmap exactly"

Requirement: Missing chunks + status snapshot

getMissingChunkIds() SHALL return ascending bigint[] of un-received chunk ids. getStatus() SHALL return a frozen snapshot containing fileId, totalChunks, receivedChunks, validatedChunks, decryptedChunks, streamedChunks, progress, isStreaming, isComplete, missingChunks, and nextChunkId. isAllChunksReceived() and isStreamingComplete() SHALL be derivable from the same state.

Scenario: Status reflects partial progress

  • WHEN 3 of 5 chunks have been ingested
  • THEN getStatus() reports receivedChunks === 3, progress === 60, and missingChunks.length === 2

Verified by: src/queue/streaming-chunk-queue.test.ts::"reports a precise status snapshot during partial ingestion"

Requirement: clear() resets queue state

clear() SHALL drop every stored chunk and reset internal state so the queue can be reused.

Scenario: Cleared queue accepts a fresh ingest

  • WHEN a queue completes a transfer and then clear() is called
  • THEN getStatus().receivedChunks === 0 and addChunk(0n, …) proceeds without error

Verified by: src/queue/streaming-chunk-queue.test.ts::"clears all state and is reusable"

Requirement: OTel instrumentation

The package SHALL acquire the meter @suaveplan/streams-chunk-queue and SHALL record five counters (streams_chunk_queue.chunks_received_total, chunks_streamed_total, validation_failures_total, decryption_failures_total, rate_limited_total) and one histogram (stream_duration_ms). Instruments SHALL be created lazily and the meter SHALL be reused.

Scenario: Telemetry module re-uses one meter

  • WHEN getTelemetry() is invoked twice
  • THEN both calls return the same object reference

Verified by: src/telemetry/telemetry.test.ts::"creates instruments lazily and reuses the meter"

Requirement: Test-runner imports go through @suaveplan/testing/runner

Every test file under packages/foundation/flow/streams-chunk-queue/ SHALL import test runner primitives from @suaveplan/testing/runner, never directly from vitest or bun:test.

Scenario: No direct runner imports

  • WHEN grep -rnE "from ['\"](vitest|bun:test)['\"]" packages/foundation/flow/streams-chunk-queue/src/ is run
  • THEN it produces zero output

Verified by: src/queue/streaming-chunk-queue.test.ts::"executes under @suaveplan/testing/runner"

Requirement: 100 / 100 / 100 / 100 coverage

The package SHALL ship with vitest.config.ts whose coverage.thresholds are statements: 100, branches: 100, functions: 100, lines: 100.

Scenario: Coverage thresholds enforce 100 across all axes

  • WHEN the coverage gate runs
  • THEN every axis is 100 and the gate exits zero

Verified by: vitest.config.ts::"coverage.thresholds equal 100/100/100/100"

Reference in New Issue View Git Blame Copy Permalink