A store allocates about 10 MiB up front, however small the upload #48

Activity

1. mihar-22 commented on Nov 7, 2026

   

   mihar-22

   on Nov 7, 2026

   Author

   More actions

   We've made this change in our vendored copy, and it's been deployed since 7 Nov. First, a correction to the issue: our large values aren't task logs. Each one is Vite+'s record of a task's inputs. With input: [{ auto: true }], that's every file the task read, up to about 25,000 paths for a test task that loads much of node_modules, which is what takes some of them near 4 MiB.

   What changed:

   • collect() grows its buffer by doubling, from 64 KiB up to its limit, and still refuses past the limit. One-byte chunks still cost little bookkeeping.
   • uploadBlob() starts its part buffer empty and grows it to PART_SIZE only when a blob gets that far. A blob of one part or less is one put of what was read, and a larger one still goes up in 5 MiB parts.
   • Admission counts bytes, not requests: a 96 MiB budget per isolate. Each request leases 128 KiB, plus each buffer's capacity as it grows, holding the new capacity before it frees the old. It's refused with 503 memory_limit past the budget, and gives everything back when it ends.
   • The decoder's copies are counted. An indefinite-length string is joined into a copy of its own while the other fields' views keep the envelope alive. So decodeMetadata() holds the envelope's size before decoding, and frees it when every field is a view.
   • Our limits: values of 8 MiB, metadata 9 MiB and a whole store 76 MiB. A store at every limit holds at most about 27 MiB, so 3 fit at once. A usual store of ours holds about 1.4 MiB, so about 70 fit.

   The result: a full run of our test tasks made 277 stores, all 200, with no memory_limit in the Worker's logs. Before, at the default cap of 2 stores, a run of three test tasks at once saw an upload refused with 503.

   The patch, against our vendored copy

   Our copy had already raised Admission's caps to 6 stores and 8 reads, so that's what admission.ts's removed lines show. uploadBlob() is exported, and takes a structural type of the R2 calls it makes, so a test can record them.

   diff --git a/src/admission.ts b/src/admission.ts
   index c71b57f7d..aee8f3200 100644
   --- a/src/admission.ts
   +++ b/src/admission.ts
   @@ -1,20 +1,53 @@
    import { HttpError } from './errors.ts';
   +import { MiB } from './limits.ts';
   +
   +// ref0: an isolate admits requests by the bytes their buffers hold, not by a count of stores and
   +// reads. Upstream's 2 stores and 4 reads stood for about 10 MiB a store and 4 MiB a read, the
   +// buffers each allocated whole before reading a byte. Buffers now grow as bytes arrive, so a store
   +// of our usual sizes holds about 1.4 MiB, and a count would refuse stores the memory could take.
   +// 96 MiB of an isolate's 128 MiB leaves 32 MiB for the runtime, the code and garbage not yet
   +// collected. A store at every limit holds at most about 27 MiB: a 9 MiB metadata buffer, the 9 MiB
   +// decoding it may copy, and a part buffer growing from 4 to 5 MiB. So 3 such stores fit at once,
   +// or about 70 of our usual size.
   +export const BUDGET = 96 * MiB;
   +
   +// ref0: what each request holds besides its buffers: its input's 64 KiB window and framing.
   +export const REQUEST_BYTES = 128 * 1024;
   +
   +// ref0: one request's share of the budget. A buffer holds its new capacity before it frees its
   +// old one, so a copy counts both.
   +export class Lease {
   +  private bytes = 0;
   +  constructor(private admission: Admission) {}
   +  hold(bytes: number): void {
   +    this.admission.take(bytes);
   +    this.bytes += bytes;
   +  }
   +  free(bytes: number): void {
   +    const freed = Math.min(bytes, this.bytes);
   +    this.admission.give(freed);
   +    this.bytes -= freed;
   +  }
   +  release(): void {
   +    this.free(this.bytes);
   +  }
   +}
    
    // Isolate-wide resource counters contain no request data or I/O promises. Rate
    // limiting controls traffic; these counters additionally bound simultaneous buffers.
    export class Admission {
   -  private stores = 0;
   -  private reads = 0;
   -  acquire(store: boolean): () => void {
   -    // ref0: 6 stores and 8 reads, not upstream's 2 and 4, which refused uploads from a full local
   -    // test run. A store holds about 10 MiB, its envelope's 5 MiB buffer and one 5 MiB part, and a
   -    // read 4 MiB, so all at once hold about 92 MiB of an isolate's 128 MiB.
   -    if (store ? this.stores >= 6 : this.reads >= 8) throw new HttpError(503, 'concurrency_limit');
   -    if (store) this.stores++;
   -    else this.reads++;
   -    return () => {
   -      if (store) this.stores--;
   -      else this.reads--;
   -    };
   +  private held = 0;
   +  constructor(private budget = BUDGET) {}
   +  acquire(): Lease {
   +    const lease = new Lease(this);
   +    lease.hold(REQUEST_BYTES);
   +    return lease;
   +  }
   +  take(bytes: number): void {
   +    if (this.held + bytes > this.budget) throw new HttpError(503, 'memory_limit');
   +    this.held += bytes;
   +  }
   +  give(bytes: number): void {
   +    this.held -= bytes;
      }
    }
   diff --git a/src/fetch.ts b/src/fetch.ts
   index 73b8283e2..3f70b9a5d 100644
   --- a/src/fetch.ts
   +++ b/src/fetch.ts
   @@ -1,3 +1,4 @@
   +import type { Lease } from './admission.ts';
    import { decodeEnvelope } from './cbor.ts';
    import { selectEntry } from './database.ts';
    import { badRequest, HttpError, unavailable } from './errors.ts';
   @@ -17,11 +18,12 @@ export async function fetchMetadata(
      limits: Limits,
      deadline: Deadline,
      stats: Observations,
   +  lease: Lease,
    ): Promise<FetchResult> {
      if (parameters(request.headers.get('Content-Type') ?? '').type !== 'application/cbor')
        badRequest();
      contentLength(request, limits.fetch);
   -  const body = await readBody(request.body, limits.fetch, deadline);
   +  const body = await readBody(request.body, limits.fetch, deadline, lease);
      stats.request_bytes = body.length;
      const data = decodeEnvelope(body, false, limits);
      const selected = await deadline.run(
   @@ -36,7 +38,7 @@ export async function fetchMetadata(
        .run(env.ARTIFACTS.get(selected.value_object))
        .catch(() => unavailable());
      if (!object || object.size !== selected.value_size || object.size > limits.value) unavailable();
   -  const value = await readBody(object.body, limits.value, deadline).catch(() => unavailable());
   +  const value = await readBody(object.body, limits.value, deadline, lease).catch(() => unavailable());
      if (value.length !== selected.value_size) unavailable();
      return { kind: 'exact', value, blob_id: selected.blob_id };
    }
   diff --git a/src/limits.ts b/src/limits.ts
   index b3c1f2650..38e665e21 100644
   --- a/src/limits.ts
   +++ b/src/limits.ts
   @@ -3,13 +3,17 @@ export const PART_SIZE = 5 * MiB;
    export const LEASE_SECONDS = 15 * 60;
    export const GRACE_SECONDS = 10 * 60;
    
   +// ref0: values of 8 MiB, not 4, with metadata and the whole store 4 MiB larger to match. A value is
   +// Vite+'s record of a task's inputs, and our server test tasks' reach 4.0 MB, so a few more files
   +// would see each store refused as too large, and Vite+ says nothing. `admission.ts`'s budget
   +// bounds what the larger limits hold.
    export const defaults = {
      key: 16 * 1024,
   -  value: 4 * MiB,
   -  metadata: 5 * MiB,
   +  value: 8 * MiB,
   +  metadata: 9 * MiB,
      fetch: 40 * 1024,
      blob: 64 * MiB,
   -  store: 72 * MiB,
   +  store: 76 * MiB,
      headers: 8 * 1024,
      deadlineMs: 120_000,
    };
   diff --git a/src/store.ts b/src/store.ts
   index d6ecc29f3..c7c507f0c 100644
   --- a/src/store.ts
   +++ b/src/store.ts
   @@ -10,21 +10,52 @@ import {
    import { badRequest, tooLarge, unavailable } from './errors.ts';
    import { PART_SIZE, type Limits } from './limits.ts';
    import { boundaryFrom, multipart } from './multipart.ts';
   -import { collect, contentLength, Deadline, Input } from './streams.ts';
   +import { collect, contentLength, Deadline, grow, Input } from './streams.ts';
   +import type { Lease } from './admission.ts';
    import type { WriteIdentity } from './auth.ts';
    import type { Observations } from './observations.ts';
    
   -async function uploadBlob(
   -  env: Env,
   +// ref0: decodes a store's metadata with what decoding copies held on `lease`. A chunked string is
   +// joined into a copy of its own, and every copy's bytes come from `body`, so `body`'s size is held
   +// before decoding, and freed when every field is a view of `body`.
   +export function decodeMetadata(
   +  body: Uint8Array<ArrayBuffer>,
   +  limits: Limits,
   +  lease: Lease,
   +): StoreMetadata {
   +  lease.hold(body.length);
   +  const metadata = decodeEnvelope(body, true, limits);
   +  const fields = [metadata.key, metadata.secondary_key, metadata.value];
   +  if (fields.every((field) => field.buffer === body.buffer)) lease.free(body.length);
   +  return metadata;
   +}
   +
   +// ref0: what `uploadBlob` uses of R2, so `../../../src/worker.test.ts` can record its calls.
   +interface Artifacts {
   +  put(key: string, value: Uint8Array): Promise<object | null>;
   +  createMultipartUpload(key: string): Promise<{
   +    uploadId: string;
   +    uploadPart(part: number, value: Uint8Array): Promise<R2UploadedPart>;
   +    complete(parts: R2UploadedPart[]): Promise<object>;
   +    abort(): Promise<void>;
   +  }>;
   +}
   +
   +// ref0: exported for `../../../src/worker.test.ts`.
   +export async function uploadBlob(
   +  env: { ARTIFACTS: Artifacts; INDEX: D1Database },
      generation: Generation,
      source: AsyncIterable<Uint8Array>,
      limit: number,
      deadline: Deadline,
      stats: Observations,
   +  lease: Lease,
    ): Promise<void> {
   -  let buffer = new Uint8Array(PART_SIZE);
   +  // ref0: the part buffer starts empty and grows to `PART_SIZE` only for a blob that gets that far,
   +  // where upstream allocated `PART_SIZE` before reading a byte. A smaller blob is one `put`.
   +  let buffer = new Uint8Array(0);
      let used = 0;
   -  let upload: R2MultipartUpload | undefined;
   +  let upload: Awaited<ReturnType<Artifacts['createMultipartUpload']>> | undefined;
      const parts: R2UploadedPart[] = [];
      for await (const chunk of source) {
        generation.blob_size += chunk.length;
   @@ -52,7 +83,9 @@ async function uploadBlob(
            buffer = new Uint8Array(PART_SIZE);
            used = 0;
          }
   -      const size = Math.min(PART_SIZE - used, chunk.length - offset);
   +      if (used === buffer.length)
   +        buffer = grow(buffer, used, used + chunk.length - offset, PART_SIZE, lease);
   +      const size = Math.min(buffer.length - used, chunk.length - offset);
          buffer.set(chunk.subarray(offset, offset + size), used);
          used += size;
          offset += size;
   @@ -82,6 +115,7 @@ export async function store(
      limits: Limits,
      deadline: Deadline,
      stats: Observations,
   +  lease: Lease,
    ): Promise<Response> {
      const boundary = boundaryFrom(request.headers.get('Content-Type'));
      const length = contentLength(request, limits.store);
   @@ -94,7 +128,7 @@ export async function store(
        let metadata: StoreMetadata | undefined;
        for await (const part of multipart(input, boundary, limits.headers)) {
          if (part.name === 'metadata') {
   -        metadata = decodeEnvelope(await collect(part.body, limits.metadata), true, limits);
   +        metadata = decodeMetadata(await collect(part.body, limits.metadata, lease), limits, lease);
            generation.value_size = metadata.value.length;
            stats.r2_operations++;
            const result = await deadline.run(
   @@ -102,7 +136,7 @@ export async function store(
            );
            if (!result) unavailable();
          } else {
   -        await uploadBlob(env, generation, part.body, limits.blob, deadline, stats);
   +        await uploadBlob(env, generation, part.body, limits.blob, deadline, stats, lease);
          }
        }
        if (!metadata || (length !== null && input.bytes !== length)) badRequest();
   diff --git a/src/streams.ts b/src/streams.ts
   index 5a47b0c1f..2078b2a61 100644
   --- a/src/streams.ts
   +++ b/src/streams.ts
   @@ -1,4 +1,5 @@
    import { Buffer } from 'node:buffer';
   +import type { Lease } from './admission.ts';
    import { badRequest, HttpError, tooLarge } from './errors.ts';
    
    export class Deadline {
   @@ -36,15 +37,41 @@ export class Deadline {
      }
    }
    
   +// ref0: a buffer's smallest capacity once it holds a byte.
   +const FIRST_CAPACITY = 64 * 1024;
   +
   +// ref0: `buffer`, grown to hold `needed` bytes, by doubling and never past `limit`, with its
   +// first `used` bytes copied. `lease` holds the new capacity before it frees the old.
   +export function grow(
   +  buffer: Uint8Array<ArrayBuffer>,
   +  used: number,
   +  needed: number,
   +  limit: number,
   +  lease?: Lease,
   +): Uint8Array<ArrayBuffer> {
   +  const capacity = Math.min(limit, Math.max(needed, 2 * buffer.length, FIRST_CAPACITY));
   +  lease?.hold(capacity);
   +  const next = new Uint8Array(capacity);
   +  next.set(buffer.subarray(0, used));
   +  lease?.free(buffer.length);
   +  return next;
   +}
   +
    export async function collect(
      source: AsyncIterable<Uint8Array>,
      limit: number,
   +  // ref0: a collect with no lease is the multipart epilogue's, at most `headers` bytes, which the
   +  // request's own share covers, or GitHub's keys', at most 64 KiB, read once in 30 seconds.
   +  lease?: Lease,
    ): Promise<Uint8Array<ArrayBuffer>> {
      let size = 0;
   -  // Fixed-capacity accumulation also bounds bookkeeping for one-byte chunks.
   -  const buffer = new Uint8Array(limit);
   +  // ref0: the buffer grows as bytes arrive, where upstream allocated `limit` up front. Doubling
   +  // keeps the copies to about twice the bytes, and still bounds bookkeeping for one-byte chunks.
   +  let buffer = new Uint8Array(0);
      for await (const chunk of source) {
        if (chunk.length > limit - size) tooLarge();
   +    if (chunk.length > buffer.length - size)
   +      buffer = grow(buffer, size, size + chunk.length, limit, lease);
        buffer.set(chunk, size);
        size += chunk.length;
      }
   @@ -116,10 +143,11 @@ export async function readBody(
      body: ReadableStream<Uint8Array> | null,
      limit: number,
      deadline: Deadline,
   +  lease?: Lease,
    ): Promise<Uint8Array<ArrayBuffer>> {
      const input = new Input(body, limit, deadline);
      try {
   -    return await collect(input.rest(), limit);
   +    return await collect(input.rest(), limit, lease);
      } finally {
        input.close();
      }

   Glad to send it as a PR, without our limits and with tests in your suite's shape, if you'd like it.