feat(sdk): make consumption-paced decrypt prefetch the default (DSPX-4654)

lib/src/opentdf.ts

@@ -175,12 +175,20 @@ export type ReadOptions = {
   /**
    * Maximum number of payload segments to fetch and decrypt per batch.
    * Adjust together with `maxConcurrentSegmentBatches` for expected throughput.
+   *
+   * Defaults to a value derived from the TDF's own segment size, so that
+   * `segmentBatchSize * maxConcurrentSegmentBatches` segments stay inside a
+   * fixed byte budget (see `derivePrefetchWindow`). Raising this raises peak
+   * memory in proportion.
    */
   segmentBatchSize?: number;
 
   /**
    * Maximum number of segment batches that may be fetched concurrently.
    * Adjust together with `segmentBatchSize` for expected throughput.
+   *
+   * Also derived from the TDF's segment size by default. Setting one of these
+   * two does not rescale the other -- the unset one keeps its derived value.
    */
   maxConcurrentSegmentBatches?: number;
 

lib/tdf3/src/client/builders.ts

@@ -757,6 +757,10 @@ class DecryptParamsBuilder {
   /**
    * Set the number of payload segments to fetch and decrypt per batch.
    * Adjust together with `withMaxConcurrentSegmentBatches()` for expected throughput.
+   *
+   * Left unset, this is derived from the TDF's segment size so that the
+   * prefetch window stays inside a fixed byte budget. Raising it raises peak
+   * memory in proportion.
    */
   withSegmentBatchSize(size: number): DecryptParamsBuilder {
     this._params.segmentBatchSize = size;
@@ -766,6 +770,8 @@ class DecryptParamsBuilder {
   /**
    * Set the maximum number of payload segment batches to fetch concurrently.
    * Adjust together with `withSegmentBatchSize()` for expected throughput.
+   *
+   * Also derived from the TDF's segment size when unset.
    */
   withMaxConcurrentSegmentBatches(limit: number): DecryptParamsBuilder {
     this._params.maxConcurrentSegmentBatches = limit;

lib/tdf3/src/tdf.ts

@@ -80,6 +80,7 @@ import type { CentralDirectory } from './utils/zip-reader.js';
 import { ByteAccumulator } from './utils/byte-accumulator.js';
 import {
   DEFAULT_MANIFEST_MAX_SIZE,
+  derivePrefetchWindow,
   estimateManifestBytes,
   maxEncryptableBytes,
   maxSegmentsFor,
@@ -137,10 +138,6 @@ export function manifestEntryName(centralDirectory: CentralDirectory[]): string
 }
 
 const HEX_SEMVER_VERSION = '4.2.2';
-const LEGACY_SEGMENTS_PER_DOWNLOAD = 500;
-const LEGACY_MAX_CONCURRENT_SEGMENT_BATCHES = 3;
-const DEFAULT_BOUND_SEGMENT_BATCH_SIZE = LEGACY_SEGMENTS_PER_DOWNLOAD;
-const DEFAULT_BOUND_MAX_CONCURRENT_SEGMENT_BATCHES = LEGACY_MAX_CONCURRENT_SEGMENT_BATCHES;
 
 /**
  * Configuration for TDF3
@@ -1415,38 +1412,6 @@ async function decryptChunk(
   return await cipher.decrypt(encryptedChunk, reconstructedKey);
 }
 
-async function updateChunkQueue(
-  chunks: Chunk[],
-  centralDirectory: CentralDirectory[],
-  zipReader: ZipReader,
-  reconstructedKey: SymmetricKey,
-  cipher: SymmetricCipher,
-  segmentIntegrityAlgorithm: SegmentIntegrityAlgorithm,
-  cryptoService: CryptoService,
-  specVersion: string
-) {
-  let requests = [];
-
-  for (let i = 0; i < chunks.length; i += LEGACY_SEGMENTS_PER_DOWNLOAD) {
-    if (requests.length === LEGACY_MAX_CONCURRENT_SEGMENT_BATCHES) {
-      await Promise.all(requests);
-      requests = [];
-    }
-    requests.push(
-      fetchAndDecryptChunkSlice({
-        centralDirectory,
-        zipReader,
-        reconstructedKey,
-        cipher,
-        segmentIntegrityAlgorithm,
-        cryptoService,
-        specVersion,
-        slice: chunks.slice(i, i + LEGACY_SEGMENTS_PER_DOWNLOAD),
-      }).catch(() => undefined)
-    );
-  }
-}
-
 function rejectChunks(chunks: Chunk[], error: Error) {
   for (const chunk of chunks) {
     chunk.decryptedChunk.reject(error);
@@ -1629,28 +1594,35 @@ function normalizeSegmentBatchSetting(
 }
 
 /**
- * Enables bounded scheduling only when at least one tuning knob is set.
- * If callers set only one knob, the other falls back to the legacy value so
- * throughput stays aligned with the pre-bounded path. Adjust both knobs together
- * when tuning for predictable memory and performance characteristics.
+ * Prefetch window for a decrypt, in segments.
+ *
+ * Bounded scheduling is now the default rather than opt-in. The old default was
+ * `updateChunkQueue`, which fired every batch at once and never waited for the
+ * consumer: at 500 segments x 3 concurrent batches that is 1.5 GiB of decrypted
+ * payload in flight at the 1 MiB segment size, and 24 GiB once
+ * `chooseSegmentSize` picks 16 MiB. Neither number was chosen; they fell out of
+ * a segment count that nothing tied to memory.
+ *
+ * Explicit knobs still win, and still fall back to the derived value rather
+ * than each other, so setting one does not silently rescale the other.
  */
-function getBoundedSegmentSchedulerOptions({
+export function getBoundedSegmentSchedulerOptions({
   segmentBatchSize,
   maxConcurrentSegmentBatches,
-}: Pick<DecryptConfiguration, 'segmentBatchSize' | 'maxConcurrentSegmentBatches'>) {
-  if (segmentBatchSize === undefined && maxConcurrentSegmentBatches === undefined) {
-    return undefined;
-  }
-
+  segmentSize,
+}: Pick<DecryptConfiguration, 'segmentBatchSize' | 'maxConcurrentSegmentBatches'> & {
+  segmentSize: number;
+}) {
+  const derived = derivePrefetchWindow({ segmentSize });
   return {
     segmentBatchSize: normalizeSegmentBatchSetting(
       segmentBatchSize,
-      DEFAULT_BOUND_SEGMENT_BATCH_SIZE,
+      derived.segmentBatchSize,
       'segmentBatchSize'
     ),
     maxConcurrentSegmentBatches: normalizeSegmentBatchSetting(
       maxConcurrentSegmentBatches,
-      DEFAULT_BOUND_MAX_CONCURRENT_SEGMENT_BATCHES,
+      derived.maxConcurrentSegmentBatches,
       'maxConcurrentSegmentBatches'
     ),
   };
@@ -1839,40 +1811,27 @@ export async function decryptStreamFrom(
   // valid here; only unknown ones are rejected.
   const segmentIntegrityAlg = asSegmentIntegrityAlgorithm(segmentHashAlg || rootIntegrityAlgorithm);
 
-  const schedulerOptions = getBoundedSegmentSchedulerOptions(cfg);
-  let scheduler: SegmentBatchScheduler | undefined;
-  if (schedulerOptions) {
-    scheduler = createBoundedSegmentScheduler({
-      totalSegments: chunks.length,
-      ...schedulerOptions,
-      onError: (error, startIndex) => {
-        rejectChunks(chunks.slice(startIndex), error);
-      },
-      scheduleBatch: async (startIndex, endIndex) =>
-        fetchAndDecryptChunkSlice({
-          centralDirectory,
-          zipReader,
-          reconstructedKey: keyForDecryption,
-          cipher,
-          segmentIntegrityAlgorithm: segmentIntegrityAlg,
-          cryptoService: cfg.cryptoService,
-          specVersion,
-          slice: chunks.slice(startIndex, endIndex),
-        }),
-    });
-    scheduler.fillWindow();
-  } else {
-    void updateChunkQueue(
-      chunks,
-      centralDirectory,
-      zipReader,
-      keyForDecryption,
-      cipher,
-      segmentIntegrityAlg,
-      cfg.cryptoService,
-      specVersion
-    );
-  }
+  const scheduler = createBoundedSegmentScheduler({
+    totalSegments: chunks.length,
+    // The encrypted size is what the window actually holds -- these are ciphertext
+    // segments fetched from the archive, not yet-decrypted plaintext.
+    ...getBoundedSegmentSchedulerOptions({ ...cfg, segmentSize: encryptedSegmentSizeDefault }),
+    onError: (error, startIndex) => {
+      rejectChunks(chunks.slice(startIndex), error);
+    },
+    scheduleBatch: async (startIndex, endIndex) =>
+      fetchAndDecryptChunkSlice({
+        centralDirectory,
+        zipReader,
+        reconstructedKey: keyForDecryption,
+        cipher,
+        segmentIntegrityAlgorithm: segmentIntegrityAlg,
+        cryptoService: cfg.cryptoService,
+        specVersion,
+        slice: chunks.slice(startIndex, endIndex),
+      }),
+  });
+  scheduler.fillWindow();
 
   let progress = 0;
   let nextChunkIndex = 0;
@@ -1898,7 +1857,7 @@ export async function decryptStreamFrom(
         decryptedChunk: mailbox<DecryptResult>(),
       };
       nextChunkIndex += 1;
-      scheduler?.markConsumed();
+      scheduler.markConsumed();
     },
     ...(cfg.fileStreamServiceWorker && { fileStreamServiceWorker: cfg.fileStreamServiceWorker }),
   };

lib/tdf3/src/utils/scale-limits.ts

@@ -338,6 +338,78 @@ export function maxOutputBytes({
  */
 const ZIP_CONTAINER_OVERHEAD = 64 * 1024;
 
+/**
+ * How much decrypted-but-unconsumed payload the reader will hold ahead of the
+ * consumer.
+ *
+ * The prefetch window has to be a *byte* budget, not a segment count. A window
+ * of "8 segments x 3 batches" is 24 MiB at the 1 MiB default and 384 MiB once
+ * {@link chooseSegmentSize} picks 16 MiB for a large file -- the same
+ * configuration, sixteen times the memory, with nothing in the numbers to
+ * suggest it. Budgeting bytes makes the window mean the same thing at every
+ * segment size.
+ *
+ * 128 MiB is enough to keep several ranged reads in flight on a fast link while
+ * staying somewhere a browser tab can live.
+ */
+export const DEFAULT_PREFETCH_BYTE_BUDGET = 128 * 1024 * 1024;
+
+/**
+ * Most batches to keep in flight at once, before the byte budget reduces it.
+ *
+ * Concurrency hides request latency; past a handful it mostly adds memory. The
+ * legacy path used 3 and there is no evidence more helped.
+ */
+export const DEFAULT_MAX_CONCURRENT_SEGMENT_BATCHES = 3;
+
+/**
+ * Most segments in a single batch, i.e. a single ranged read.
+ *
+ * The byte budget alone would allow enormous batches at small segment sizes
+ * (43,690 segments per batch at 1 KiB), which stays within budget but makes the
+ * consumer wait on one absurd ranged read before the first byte. This is the
+ * legacy batch size, so tiny-segment files keep behaving as they always did.
+ */
+export const MAX_SEGMENT_BATCH_SIZE = 500;
+
+/**
+ * Splits a byte budget into a batch size and a concurrency level.
+ *
+ * The product of the two is the window, so both have to come out of the same
+ * division. Concurrency yields first: three batches of one segment is a better
+ * use of a tight budget than one batch of three, because the reads overlap.
+ *
+ * When a single segment already exceeds the budget the window is one segment
+ * and the budget is knowingly blown -- there is no smaller unit to schedule,
+ * and refusing to decrypt would be worse than using the memory.
+ */
+export function derivePrefetchWindow({
+  segmentSize,
+  byteBudget = DEFAULT_PREFETCH_BYTE_BUDGET,
+  maxConcurrentSegmentBatches = DEFAULT_MAX_CONCURRENT_SEGMENT_BATCHES,
+  maxSegmentBatchSize = MAX_SEGMENT_BATCH_SIZE,
+}: {
+  segmentSize: number;
+  byteBudget?: number;
+  maxConcurrentSegmentBatches?: number;
+  maxSegmentBatchSize?: number;
+}): { segmentBatchSize: number; maxConcurrentSegmentBatches: number } {
+  assertPositiveInteger(segmentSize, 'segmentSize');
+  assertPositiveInteger(byteBudget, 'byteBudget');
+  assertPositiveInteger(maxConcurrentSegmentBatches, 'maxConcurrentSegmentBatches');
+  assertPositiveInteger(maxSegmentBatchSize, 'maxSegmentBatchSize');
+
+  const segmentsInWindow = Math.max(1, Math.floor(byteBudget / segmentSize));
+  const batches = Math.min(maxConcurrentSegmentBatches, segmentsInWindow);
+  return {
+    segmentBatchSize: Math.min(
+      maxSegmentBatchSize,
+      Math.max(1, Math.floor(segmentsInWindow / batches))
+    ),
+    maxConcurrentSegmentBatches: batches,
+  };
+}
+
 function assertPositiveInteger(value: number, name: string): void {
   if (!Number.isInteger(value) || value < 1) {
     throw new ConfigurationError(`${name} must be a positive integer; got [${value}]`);