DX-3088: version the upload protocol

src/browser/task.ts

@@ -1,5 +1,6 @@
 import { BlobError } from '../shared/errors.ts';
-import type { CompletedBlob, UploadSnapshot, UploadTask, WireBeginResponse, WireEndResponse, WireLanded, WirePartsResponse } from '../shared/types.ts';
+import { protocolOf, UPLOAD_PROTOCOL } from '../shared/protocol.ts';
+import type { CompletedBlob, UploadSnapshot, UploadTask, WireAnswer, WireBeginResponse, WireEndResponse, WireLanded, WirePartsResponse } from '../shared/types.ts';
 import { SNIFF_BYTES } from '../shared/units.ts';
 import { abortError, clock } from './clock.ts';
 import { acquire } from './pool.ts';
@@ -78,6 +79,11 @@ class Task implements InternalTask {
   private token: string | undefined;
   private storeKey: string;
   private partSize = 0;
+  /**
+   * The protocol on the last answer to arrive. Calls overlap, so this is not a gate: a feature reads
+   * protocolOf() off the answer it holds, which a deploy may have had a different release write.
+   */
+  private protocol = 1;
   // Whether the route cut this file into real multipart parts. False is a single PUT: one url, one
   // object write, and nothing on the other side to pause into or resume from.
   private multipart = false;
@@ -552,7 +558,7 @@ class Task implements InternalTask {
         res = await fetch(this.options.route, {
           method: 'POST',
           headers: { 'content-type': 'application/json', ...authored },
-          body: JSON.stringify(body),
+          body: JSON.stringify({ ...body, protocol: UPLOAD_PROTOCOL }),
           signal,
         });
       } catch (e) {
@@ -569,7 +575,10 @@ class Task implements InternalTask {
       } catch {
         json = undefined;
       }
-      if (res.ok) return json;
+      if (res.ok) {
+        this.protocol = protocolOf((json as WireAnswer | undefined)?.protocol);
+        return json;
+      }
       const err = BlobError.fromJSON(json, res.status) ?? BlobError.fromStatus(res.status, { message: routeMessage(json, res) });
       if (classify(res.status) !== 'retry') throw err;
       last = err;

src/react/constraints.ts

@@ -2,6 +2,7 @@ import { useCallback, useEffect, useRef, useState } from 'react';
 import { clock } from '../browser/clock.ts';
 import { resolveHeaders, type HeadersProvider } from '../browser/task.ts';
 import { BlobError } from '../shared/errors.ts';
+import { protocolOf } from '../shared/protocol.ts';
 import type { ServedConstraints, WireConstraintsResponse } from '../shared/types.ts';
 import { overLimit } from '../shared/units.ts';
 
@@ -12,6 +13,11 @@ const inFlight = new Map<string, Promise<RouteFacts>>();
 
 export interface RouteFacts {
   constraints: ServedConstraints | undefined;
+  /**
+   * The route's UPLOAD_PROTOCOL: 1 when it sent none, or could not be asked. Only an upload route's
+   * means anything: a useServerUpload url is the app's own, and its GET may say whatever it likes.
+   */
+  protocol: number;
   /** Why the route could not be asked. Failed responses are not cached. */
   error?: BlobError;
 }
@@ -42,33 +48,35 @@ async function fetchFacts(route: string, headers: HeadersProvider | undefined):
   try {
     authored = await resolveHeaders(headers);
   } catch (e) {
-    return { constraints: undefined, error: BlobError.is(e) ? e : new BlobError('request_failed', { message: e instanceof Error ? e.message : String(e), status: 400, cause: e }) };
+    return { constraints: undefined, protocol: 1, error: BlobError.is(e) ? e : new BlobError('request_failed', { message: e instanceof Error ? e.message : String(e), status: 400, cause: e }) };
   }
   let res: Response;
   try {
     res = await fetch(route, { headers: authored });
   } catch (e) {
-    return { constraints: undefined, error: new BlobError('request_failed', { message: 'could not reach the route', status: 503, cause: e }) };
+    return { constraints: undefined, protocol: 1, error: new BlobError('request_failed', { message: 'could not reach the route', status: 503, cause: e }) };
   }
   // Ordinary server-upload URLs commonly expose POST only. There are no route constraints to cache.
-  if (res.status === 404 || res.status === 405) return remember(route, { constraints: undefined });
+  if (res.status === 404 || res.status === 405) return remember(route, { constraints: undefined, protocol: 1 });
   if (!res.ok) {
     let body: unknown;
     try {
       body = await res.json();
     } catch {
       body = undefined;
     }
-    return { constraints: undefined, error: BlobError.fromJSON(body, res.status) ?? BlobError.fromStatus(res.status) };
+    return { constraints: undefined, protocol: 1, error: BlobError.fromJSON(body, res.status) ?? BlobError.fromStatus(res.status) };
   }
   let constraints: ServedConstraints | undefined;
+  let protocol = 1;
   try {
     const body = (await res.json()) as WireConstraintsResponse | undefined;
     if (body?.constraints && typeof body.constraints === 'object') constraints = body.constraints;
+    protocol = protocolOf(body?.protocol);
   } catch {
     // Not a constraints document. The upload itself remains authoritative.
   }
-  return remember(route, { constraints });
+  return remember(route, { constraints, protocol });
 }
 
 function remember(route: string, facts: RouteFacts): RouteFacts {

src/server/handle-upload.ts

@@ -1,4 +1,5 @@
 import { BlobError } from '../shared/errors.ts';
+import { UPLOAD_PROTOCOL } from '../shared/protocol.ts';
 import type { UploadFile, WireBeginResponse, WireEndResponse, WireLanded, ServedConstraints, WireConstraintsResponse, WirePart, WirePartsResponse } from '../shared/types.ts';
 import { cacheControl, overLimit, parseSize, type CacheOption, type Size } from '../shared/units.ts';
 import { r2Of, type Bucket } from './bucket.ts';
@@ -109,13 +110,13 @@ export function handleUpload(options: InternalUploadOptions): InternalUploadHand
       if (!body || typeof body !== 'object' || typeof body.phase !== 'string') throw new BlobError('invalid_input', { message: 'expected a JSON body with a phase' });
       switch (body.phase) {
         case 'begin':
-          return Response.json(await begin(request, body, details));
+          return answer(await begin(request, body, details));
         case 'parts':
-          return Response.json(await parts(body, details));
+          return answer(await parts(body, details));
         case 'end':
-          return Response.json(await end(request, body, details));
+          return answer(await end(request, body, details));
         case 'cancel':
-          return Response.json(await cancel(body, details));
+          return answer(await cancel(body, details));
         default:
           throw new BlobError('invalid_input', { message: `unknown phase ${String(body.phase)}` });
       }
@@ -402,6 +403,14 @@ export function handleUpload(options: InternalUploadOptions): InternalUploadHand
   return { GET, POST };
 }
 
+/**
+ * Every successful answer carries the protocol of the server that wrote it. On a deploy, instances of
+ * two releases answer one upload side by side, so the client reads each answer by its own number.
+ */
+function answer(body: object): Response {
+  return Response.json({ ...body, protocol: UPLOAD_PROTOCOL });
+}
+
 /**
  * Every refusal leaves a route as BlobError.toJSON(), so the browser rebuilds it with its code
  * intact and callers switch on error.code instead of reading status numbers and message text.
@@ -433,7 +442,7 @@ export function constraintsEndpoint(routeConstraints: ResolvedConstraints): (req
   const served: ServedConstraints = {};
   if (routeConstraints.contentTypes) served.contentTypes = routeConstraints.contentTypes;
   if (routeConstraints.maxSize !== undefined) served.maxSize = routeConstraints.maxSize;
-  const body = JSON.stringify({ constraints: served } satisfies WireConstraintsResponse);
+  const body = JSON.stringify({ protocol: UPLOAD_PROTOCOL, constraints: served } satisfies WireConstraintsResponse);
   const etag = `"${hash(body)}"`;
   return async (request: Request): Promise<Response> => {
     const headers = { 'content-type': 'application/json', 'cache-control': `public, max-age=${LIMITS_MAX_AGE}`, etag };

src/shared/protocol.ts

@@ -0,0 +1,36 @@
+/**
+ * The version of the upload route protocol: what `uploadHandler` and `@upstash/blob/browser` (and
+ * the React hooks on top of it) say to each other. It is not the package version: a browser bundle
+ * and the server it talks to ship on their own schedules, and blob-py's handler speaks the same
+ * protocol from a different package entirely.
+ *
+ * One rule: every upload route JSON body carries it as a top-level `protocol` key, beside `phase`
+ * or `constraints`. The client puts it on every POST; the server on the GET constraints document and
+ * on every 2xx answer to a POST (begin, parts, end and cancel). Error answers carry none. It is
+ * never a header, so it needs no CORS change.
+ * Anything missing, or not a JSON number with an integer value of at least 1 (a boolean is not one),
+ * is 1, which is what every release before the field existed spoke. Both sides ignore keys they do
+ * not know; the rest of this depends on it.
+ *
+ * Every change is additive. Neither side has a minimum and nothing is ever refused over the number:
+ * a newer server answers an older client the way that client expects, and a newer client sends new
+ * keys only where a server that ignores them still does the right thing. At 1 there is nothing to
+ * gate, so the server does not read the client's number yet; protocolOf() is how it will.
+ *
+ * One upload is many requests over minutes or days, and on a deploy or a rollback instances of two
+ * releases answer it side by side. So each answer is read by its own number, never by one an earlier
+ * call returned: 'end' can reach an older server than 'begin' did. For the same reason the
+ * completion token never bumps its own `v` for an additive field, because an older instance refuses
+ * a token it does not know, and a newer instance treats a token without the field as the old
+ * behavior, because an older 'begin' minted it. The GET number is cached for up to two minutes and
+ * a bare upload() sends no GET, so a client choosing what to send at 'begin' can be wrong about the
+ * server: what it sends has to be one of those new keys an older server safely ignores.
+ *
+ * 1: phases begin, parts, end and cancel; the GET constraints document.
+ */
+export const UPLOAD_PROTOCOL = 1;
+
+/** The protocol the other side sent. Missing, or anything but a positive integer, is 1. */
+export function protocolOf(raw: unknown): number {
+  return typeof raw === 'number' && Number.isInteger(raw) && raw >= 1 ? raw : 1;
+}

src/shared/types.ts

@@ -44,6 +44,11 @@ export interface ServedConstraints {
   maxSize?: number;
 }
 
+/** On every successful answer: the UPLOAD_PROTOCOL of the server that wrote it. Missing is 1. */
+export interface WireAnswer {
+  protocol?: number;
+}
+
 export interface WirePart {
   n: number;
   url: string;
@@ -68,7 +73,7 @@ export interface WireUploadPlan {
   multipart: boolean;
 }
 
-export interface WireBeginResponse {
+export interface WireBeginResponse extends WireAnswer {
   completionToken: string;
   path: string;
   upload: WireUploadPlan;
@@ -79,24 +84,26 @@ export interface WireLanded {
   etag: string;
 }
 
-export interface WirePartsResponse {
+export interface WirePartsResponse extends WireAnswer {
   partSize: number;
   size: number;
   parts: WirePart[];
   landed: WireLanded[];
   multipart: boolean;
 }
 
-export interface WireEndResponse<TData = unknown> {
+export interface WireEndResponse<TData = unknown> extends WireAnswer {
   blob: Omit<CompletedBlob, 'uploadedAt'> & { uploadedAt: string };
   data: TData;
 }
 
-export type WireRequest =
+/** `protocol` is the client's UPLOAD_PROTOCOL, on every phase. Missing is 1. */
+export type WireRequest = { protocol?: number } & (
   | { phase: 'begin'; file: WireFile; head?: string; input?: unknown }
   | { phase: 'parts'; completionToken: string; from: number }
   | { phase: 'end'; completionToken: string; parts?: WireLanded[] }
-  | { phase: 'cancel'; completionToken: string };
+  | { phase: 'cancel'; completionToken: string }
+);
 
 /* -------------------------------------------------------------- browser -- */
 
@@ -172,6 +179,6 @@ export type UploadRoute<TInput = unknown, TData = unknown, TRoute extends string
 };
 
 /** What GET on an upload route answers: the constraints the route enforces. */
-export interface WireConstraintsResponse {
+export interface WireConstraintsResponse extends WireAnswer {
   constraints: ServedConstraints;
 }

test/live/handle-upload.test.ts

@@ -93,6 +93,7 @@ describe('GET', () => {
     expect(etag).toMatch(/^"/);
     expect((await chat.GET(new Request('https://app.test/api/upload', { headers: { 'if-none-match': etag } }))).status).toBe(304);
     expect(await res.json()).toEqual({
+      protocol: 1,
       constraints: { contentTypes: ['image/png', 'image/jpeg', 'image/webp', 'application/pdf'], maxSize: 20_000_000 },
     });
     const img = uploadHandler({ bucket: pub, constraints: { contentTypes: ['image/*'] }, onBeforeUpload: () => ({ path: 'x' }) });

test/unit/browser.test.ts

@@ -11,7 +11,7 @@ const memory = new Map<string, string>();
 const sleeps: number[] = [];
 let timers: { ms: number; cb: () => void }[] = [];
 let wakers: (() => void)[] = [];
-let calls: { phase: string; body: any }[] = [];
+let calls: { phase: string; body: any; protocol: unknown }[] = [];
 let onPhase: (body: any) => unknown | Response;
 const restore: (() => void)[] = [];
 
@@ -86,7 +86,9 @@ beforeAll(() => {
     const url = typeof input === 'string' ? input : input.url;
     if (!url.startsWith('/api/')) return real(input, init);
     const body = JSON.parse(init!.body as string);
-    calls.push({ phase: body.phase, body });
+    // Kept apart so the body assertions stay about the phase; afterEach checks it on every call.
+    const { protocol, ...rest } = body;
+    calls.push({ phase: body.phase, body: rest, protocol });
     const out = onPhase(body);
     if (out instanceof Response) return out;
     return Response.json(out);
@@ -112,6 +114,8 @@ beforeEach(() => {
 });
 afterEach(() => {
   expect(poolState().active).toBe(0);
+  // Every phase of every test: the client's protocol rides each POST.
+  expect(calls.map((c) => c.protocol)).toEqual(calls.map(() => 1));
 });
 
 async function settle(rounds = 6) {
@@ -517,6 +521,24 @@ describe('a single PUT', () => {
     }
   });
 
+  test('answers from servers of different protocols, as on a deploy, still make one upload', async () => {
+    // begin and end can be answered by two releases: every mix of newer, none and garbage.
+    for (const [atBegin, atEnd] of [[2, undefined], [undefined, 2], [2, 'x'], ['x', 1]]) {
+      const from = calls.length;
+      const route = defaultRoute(3);
+      onPhase = (body) => {
+        const out = route(body) as Record<string, unknown>;
+        const advertised = body.phase === 'begin' ? atBegin : atEnd;
+        return advertised === undefined ? out : { ...out, protocol: advertised };
+      };
+      const task = upload(png(), { route: '/api/upload' });
+      await settle();
+      ManualXhr.pending[0]!.respond(200, { etag: '"x"' });
+      await expect(task.done).resolves.toMatchObject({ path: 'p' });
+      expect(calls.slice(from).map((c) => c.phase)).toEqual(['begin', 'end']);
+    }
+  });
+
   test('cancel while queued rejects done with AbortError and never PUTs', async () => {
     let release!: () => void;
     onPhase = () => new Response(new ReadableStream({ start: (c) => (release = () => c.close()) }));

test/unit/handler.test.ts

@@ -133,7 +133,7 @@ describe('dispatch', () => {
     const uploads = handler();
     const res = await uploads.GET(new Request(url('large')));
     expect(res.status).toBe(200);
-    expect(await res.json()).toEqual({ constraints: { maxSize: 2_000_000_000 } });
+    expect(await res.json()).toEqual({ protocol: 1, constraints: { maxSize: 2_000_000_000 } });
 
     const begin = await post(uploads, 'attachment', { phase: 'begin', file: { name: 'a.png', type: 'image/png', size: 10 } });
     expect(begin.status).toBe(200);
@@ -171,7 +171,7 @@ describe('dispatch', () => {
     const uploads = uploadHandler({ bucket: bucket(), constraints: { maxSize: '1mb' }, onBeforeUpload: () => ({ path: 'only/1.png' }) });
     const begin = await began(uploads, undefined, { name: 'a.png', type: 'image/png', size: 10 });
     expect(begin.path).toBe('only/1.png');
-    expect(await (await uploads.GET(new Request(url()))).json()).toEqual({ constraints: { maxSize: 1_000_000 } });
+    expect(await (await uploads.GET(new Request(url()))).json()).toEqual({ protocol: 1, constraints: { maxSize: 1_000_000 } });
     // A name on the query is a client bound to some other handler: it does not silently get this
     // route, for GET or for POST.
     for (const name of ['anything', '__proto__', 'toString']) {
@@ -254,11 +254,11 @@ describe('dispatch', () => {
 describe('defaults', () => {
   test('a route replaces constraints per key, and null clears one', async () => {
     const uploads = handler();
-    expect(await (await uploads.GET(new Request(url('attachment')))).json()).toEqual({ constraints: { contentTypes: ['image/png'], maxSize: 20_000_000 } });
+    expect(await (await uploads.GET(new Request(url('attachment')))).json()).toEqual({ protocol: 1, constraints: { contentTypes: ['image/png'], maxSize: 20_000_000 } });
     // maxSize replaced, contentTypes cleared by null.
-    expect(await (await uploads.GET(new Request(url('large')))).json()).toEqual({ constraints: { maxSize: 2_000_000_000 } });
+    expect(await (await uploads.GET(new Request(url('large')))).json()).toEqual({ protocol: 1, constraints: { maxSize: 2_000_000_000 } });
     // maxSize replaced, the handler's type list inherited.
-    expect(await (await uploads.GET(new Request(url('avatar')))).json()).toEqual({ constraints: { contentTypes: ['image/png'], maxSize: 2_000_000 } });
+    expect(await (await uploads.GET(new Request(url('avatar')))).json()).toEqual({ protocol: 1, constraints: { contentTypes: ['image/png'], maxSize: 2_000_000 } });
   });
 
   test('the inherited constraints still refuse', async () => {
@@ -362,8 +362,8 @@ describe('defaults', () => {
       routes: { omega: { constraints: { maxSize: '1kb' } }, alpha: {} },
     });
     for (const uploads of [first, second]) {
-      expect(await (await uploads.GET(new Request(url('alpha')))).json()).toEqual({ constraints: { maxSize: 20_000_000 } });
-      expect(await (await uploads.GET(new Request(url('omega')))).json()).toEqual({ constraints: { maxSize: 1_000 } });
+      expect(await (await uploads.GET(new Request(url('alpha')))).json()).toEqual({ protocol: 1, constraints: { maxSize: 20_000_000 } });
+      expect(await (await uploads.GET(new Request(url('omega')))).json()).toEqual({ protocol: 1, constraints: { maxSize: 1_000 } });
     }
   });
 });
@@ -477,7 +477,7 @@ test('a 32MiB upload limit accepts the exact file size and rejects one byte over
     constraints: { maxSize: '32MiB' },
     onBeforeUpload: () => ({ path: 'large/file.bin' }),
   });
-  expect(await (await uploads.GET(new Request(url()))).json()).toEqual({ constraints: { maxSize: 33_554_432 } });
+  expect(await (await uploads.GET(new Request(url()))).json()).toEqual({ protocol: 1, constraints: { maxSize: 33_554_432 } });
   const begin = await post(uploads, undefined, { phase: 'begin', file: { name: 'file.bin', type: 'application/octet-stream', size: 33_554_432 } });
   expect(begin.status).toBe(200);
   expect((await begin.json()).upload.multipart).toBe(true);