feat(control-plane): pin per-tenant image versions for fleet rollout and rollback

control-plane/src/container-driver.ts

@@ -58,12 +58,25 @@ function bindingFor(config: ContainerDriverConfig, product: Product): ContainerN
   return binding;
 }
 
+/** The env var a tenant's container reads its pinned image version from at (re)start (#4898). A Cloudflare
+ *  Container binding is fixed to one image at the wrangler.jsonc level (see `ContainerDriverConfig.bindings`),
+ *  so per-tenant versioning cannot swap the image reference binding-side — instead the tenant's own
+ *  `pinnedVersion` rides into the container, whose entrypoint resolves the versioned artifact itself. */
+export const PINNED_VERSION_ENV_VAR = "LOOPOVER_PINNED_VERSION";
+
 /** Idempotent: an already-provisioned tenant's container is left running as-is, never restarted -- a repeat
- *  create must not interrupt a container mid-work. */
+ *  create must not interrupt a container mid-work. A tenant with a `pinnedVersion` (#4898) starts with that
+ *  version in {@link PINNED_VERSION_ENV_VAR}; an unpinned tenant gets the exact pre-#4898 `start()` call, so
+ *  every existing tenant's behavior is byte-identical until a rollout pins it. */
 export async function createTenantContainer(config: ContainerDriverConfig, request: TenantProvisioningRequest): Promise<void> {
   const stub = bindingFor(config, request.product).getByName(instanceNameFor(request));
   if (await stub.isProvisioned()) return;
-  await stub.start();
+  const pinnedVersion = request.tenant.pinnedVersion;
+  if (pinnedVersion) {
+    await stub.start({ envVars: { [PINNED_VERSION_ENV_VAR]: pinnedVersion } });
+  } else {
+    await stub.start();
+  }
   await stub.markProvisioned();
 }
 

control-plane/src/http-app.ts

@@ -34,6 +34,25 @@ function safeRecord(record: Pick<TenantRegistryRecord, "tenant" | "product" | "s
   return { tenant: record.tenant, product: record.product, state: record.state };
 }
 
+/** Validated body of `POST /v1/tenants/rollout` (#4898): an explicit tenant-name list (no percentage/canary
+ *  selector — no such primitive exists elsewhere in this codebase to build on) plus the version to pin.
+ *  `pinnedVersion: null` is an explicit unpin (revert to the release channel's default). */
+type RolloutRequest = { names: string[]; pinnedVersion: string | null };
+
+function parseRolloutRequest(body: unknown): RolloutRequest | string {
+  if (body === null || typeof body !== "object" || Array.isArray(body)) return "body must be a JSON object";
+  const { names, pinnedVersion } = body as Record<string, unknown>;
+  if (!Array.isArray(names) || names.length === 0) return "names must be a non-empty array of tenant names";
+  if (!names.every((name): name is string => typeof name === "string" && name.trim() !== "")) {
+    return "names must be a non-empty array of tenant names";
+  }
+  if (new Set(names).size !== names.length) return "names must not repeat a tenant";
+  if (pinnedVersion !== null && (typeof pinnedVersion !== "string" || !pinnedVersion.trim())) {
+    return "pinnedVersion must be a non-blank string, or null to unpin";
+  }
+  return { names, pinnedVersion: pinnedVersion === null ? null : pinnedVersion.trim() };
+}
+
 export function createTenantHttpApp(deps: TenantHttpAppDeps): Hono {
   const app = new Hono();
 
@@ -79,6 +98,42 @@ export function createTenantHttpApp(deps: TenantHttpAppDeps): Hono {
     return c.json({ tenants: records.map((record) => ({ ...safeRecord(record), createdAt: record.createdAt, updatedAt: record.updatedAt })) });
   });
 
+  // #4898: rollout/rollback = updating one or more tenants' pinnedVersion via an explicit list. Validates the
+  // WHOLE list before touching any record (all-or-nothing) so a typo'd name can never leave a fleet half
+  // rolled out; each updated tenant's container picks its new version up at its next (re)start
+  // (container-driver.ts's PINNED_VERSION_ENV_VAR). Every unlisted tenant is untouched by construction —
+  // the per-tenant-independence guarantee this endpoint exists to keep.
+  app.post("/v1/tenants/rollout", async (c) => {
+    const body: unknown = await c.req.json().catch(() => null);
+    if (body === null) return c.json({ error: "invalid_json" }, 400);
+    const parsed = parseRolloutRequest(body);
+    if (typeof parsed === "string") return c.json({ error: "invalid_request", message: parsed }, 400);
+
+    const existing = new Map<string, TenantRegistryRecord>();
+    for (const name of parsed.names) {
+      const record = await deps.registry.get(name);
+      if (!record) return c.json({ error: "tenant_not_found", message: `unknown tenant "${name}"` }, 404);
+      // A torn-down tenant has no container to ever read the pin — surfacing the mistake beats silently
+      // stamping a version onto a terminated record (same conflict posture as the create route's 409).
+      if (record.state === "torn down") return c.json({ error: "tenant_torn_down", message: `tenant "${name}" is torn down` }, 409);
+      existing.set(name, record);
+    }
+
+    const now = new Date().toISOString();
+    const updated: TenantRegistryRecord[] = [];
+    for (const name of parsed.names) {
+      const record = existing.get(name)!;
+      const next: TenantRegistryRecord = {
+        ...record,
+        tenant: { ...record.tenant, pinnedVersion: parsed.pinnedVersion },
+        updatedAt: now,
+      };
+      await deps.registry.upsert(next);
+      updated.push(next);
+    }
+    return c.json({ tenants: updated.map((record) => ({ ...safeRecord(record), createdAt: record.createdAt, updatedAt: record.updatedAt })) });
+  });
+
   app.delete("/v1/tenants/:name", async (c) => {
     const name = c.req.param("name");
     const existing = await deps.registry.get(name);

control-plane/src/index.ts

@@ -55,6 +55,7 @@ export {
   createContainerDriver,
   createTenantContainer,
   destroyTenantContainer,
+  PINNED_VERSION_ENV_VAR,
   tenantContainerExists,
   type ContainerDriver,
   type ContainerDriverConfig,

control-plane/src/tenant-provisioning-driver.ts

@@ -18,6 +18,12 @@ export type Product = string;
  *  admin commands address a tenant by. */
 export type Tenant = {
   name: string;
+  /** #4898 (fleet rollout, decision ratified 2026-07-21): the image version THIS tenant's container resolves
+   *  at (re)start, instead of a shared `:latest` tag. Product-agnostic — the same field for ORB and AMS
+   *  tenants. Absent/null = unpinned (the tenant follows its release channel's default, exactly the pre-#4898
+   *  behavior). A rollout updates this field on an explicit list of tenants; rollback reverts it — see
+   *  http-app.ts's `POST /v1/tenants/rollout`. */
+  pinnedVersion?: string | null;
 };
 
 /** The full tenant lifecycle vocabulary the #7180 provisioning API reports, passed through verbatim by

control-plane/test/container-driver.test.ts

@@ -8,6 +8,7 @@ import {
   createContainerDriver,
   createTenantContainer,
   destroyTenantContainer,
+  PINNED_VERSION_ENV_VAR,
   tenantContainerExists,
   type ContainerDriverConfig,
   type ContainerNamespaceLike,
@@ -130,3 +131,60 @@ test("createContainerDriver bundles all three functions closed over one config",
   await driver.destroyContainer(REQUEST);
   assert.equal(await driver.containerExists(REQUEST), false);
 });
+
+// #4898: a tenant's pinnedVersion rides into its container at (re)start as PINNED_VERSION_ENV_VAR — the only
+// per-tenant versioning seam available when the image reference itself is fixed at the wrangler.jsonc binding
+// level. The stub here captures start()'s options, which the package's shared fake deliberately doesn't.
+type StartOptions = Parameters<ContainerStubLike["start"]>[0];
+
+function optionCapturingStub(): ContainerStubLike & { startOptions: StartOptions[] } {
+  let provisioned = false;
+  const startOptions: StartOptions[] = [];
+  return {
+    startOptions,
+    async start(options?: StartOptions) {
+      startOptions.push(options);
+    },
+    async stop() {},
+    async isProvisioned() {
+      return provisioned;
+    },
+    async markProvisioned() {
+      provisioned = true;
+    },
+    async markDeprovisioned() {
+      provisioned = false;
+    },
+  };
+}
+
+function configFor(stub: ContainerStubLike): ContainerDriverConfig {
+  return { bindings: { orb: { getByName: () => stub } } };
+}
+
+test("a pinned tenant's container starts with PINNED_VERSION_ENV_VAR carrying its own version (#4898)", async () => {
+  const stub = optionCapturingStub();
+
+  await createTenantContainer(configFor(stub), { tenant: { name: "acme", pinnedVersion: "v1.4.2" }, product: "orb" });
+
+  assert.deepEqual(stub.startOptions, [{ envVars: { [PINNED_VERSION_ENV_VAR]: "v1.4.2" } }]);
+});
+
+test("an unpinned tenant's container start is byte-identical to the pre-#4898 call (no options at all)", async () => {
+  for (const tenant of [{ name: "acme" }, { name: "acme", pinnedVersion: null }]) {
+    const stub = optionCapturingStub();
+
+    await createTenantContainer(configFor(stub), { tenant, product: "orb" });
+
+    assert.deepEqual(stub.startOptions, [undefined]);
+  }
+});
+
+test("a repeat create of an already-provisioned pinned tenant never restarts it (#4898 keeps the idempotence contract)", async () => {
+  const stub = optionCapturingStub();
+  await stub.markProvisioned();
+
+  await createTenantContainer(configFor(stub), { tenant: { name: "acme", pinnedVersion: "v2.0.0" }, product: "orb" });
+
+  assert.deepEqual(stub.startOptions, []);
+});

control-plane/test/http-app.test.ts

@@ -255,3 +255,142 @@ test("a driver failure surfaces as a logged 500 via onError, not an unhandled re
   assert.match(errors[0]!, /control_plane_http_error/);
   assert.match(errors[0]!, /cloudflare containers api unavailable/);
 });
+
+// #4898: POST /v1/tenants/rollout — pin/unpin an explicit list of tenants' pinnedVersion, all-or-nothing.
+// The registry-seeding style mirrors the GET /v1/tenants tests above (records seeded directly, no driver run).
+
+function rollout(app: ReturnType<typeof createTenantHttpApp>, body: unknown) {
+  return app.request(
+    "/v1/tenants/rollout",
+    authed({ method: "POST", headers: { "content-type": "application/json" }, body: typeof body === "string" ? body : JSON.stringify(body) }),
+  );
+}
+
+test("POST /v1/tenants/rollout pins exactly the listed tenants and leaves every other tenant untouched (#4898 acceptance)", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({ tenant: { name: "acme" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  await registry.upsert({ tenant: { name: "beta" }, product: "ams", state: "active", createdAt: "t0", updatedAt: "t0" });
+  await registry.upsert({ tenant: { name: "gamma" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  const app = createTenantHttpApp(baseDeps({ registry }));
+
+  const res = await rollout(app, { names: ["acme", "gamma"], pinnedVersion: "v1.4.2" });
+
+  assert.equal(res.status, 200);
+  const payload = (await res.json()) as { tenants: Array<{ tenant: { name: string; pinnedVersion?: string | null } }> };
+  assert.deepEqual(payload.tenants.map((t) => t.tenant), [
+    { name: "acme", pinnedVersion: "v1.4.2" },
+    { name: "gamma", pinnedVersion: "v1.4.2" },
+  ]);
+  // The unlisted tenant is completely unaffected — no pin, no updatedAt churn.
+  const beta = await registry.get("beta");
+  assert.deepEqual(beta?.tenant, { name: "beta" });
+  assert.equal(beta?.updatedAt, "t0");
+  // The pinned tenants' records persisted the pin and kept their createdAt.
+  const acme = await registry.get("acme");
+  assert.deepEqual(acme?.tenant, { name: "acme", pinnedVersion: "v1.4.2" });
+  assert.equal(acme?.createdAt, "t0");
+  assert.notEqual(acme?.updatedAt, "t0");
+});
+
+test("POST /v1/tenants/rollout rolls back independently: re-pinning one tenant leaves another tenant's pin alone", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({ tenant: { name: "acme", pinnedVersion: "v2.0.0" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  await registry.upsert({ tenant: { name: "beta", pinnedVersion: "v2.0.0" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  const app = createTenantHttpApp(baseDeps({ registry }));
+
+  const back = await rollout(app, { names: ["acme"], pinnedVersion: "v1.9.0" });
+  assert.equal(back.status, 200);
+  assert.deepEqual((await registry.get("acme"))?.tenant, { name: "acme", pinnedVersion: "v1.9.0" });
+  assert.deepEqual((await registry.get("beta"))?.tenant, { name: "beta", pinnedVersion: "v2.0.0" });
+
+  // Explicit unpin (null) reverts the tenant to its release channel's default.
+  const unpin = await rollout(app, { names: ["acme"], pinnedVersion: null });
+  assert.equal(unpin.status, 200);
+  assert.deepEqual((await registry.get("acme"))?.tenant, { name: "acme", pinnedVersion: null });
+  assert.deepEqual((await registry.get("beta"))?.tenant, { name: "beta", pinnedVersion: "v2.0.0" });
+});
+
+test("POST /v1/tenants/rollout trims the pinned version before storing it", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({ tenant: { name: "acme" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  const app = createTenantHttpApp(baseDeps({ registry }));
+
+  const res = await rollout(app, { names: ["acme"], pinnedVersion: "  v1.4.2  " });
+
+  assert.equal(res.status, 200);
+  assert.deepEqual((await registry.get("acme"))?.tenant, { name: "acme", pinnedVersion: "v1.4.2" });
+});
+
+test("POST /v1/tenants/rollout 400s malformed bodies without touching any record", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({ tenant: { name: "acme" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  const app = createTenantHttpApp(baseDeps({ registry }));
+
+  const notJson = await rollout(app, "not json at all");
+  assert.equal(notJson.status, 400);
+  assert.deepEqual(await notJson.json(), { error: "invalid_json" });
+
+  for (const [body, message] of [
+    [[], "body must be a JSON object"],
+    [{ names: [], pinnedVersion: "v1" }, "names must be a non-empty array of tenant names"],
+    [{ names: "acme", pinnedVersion: "v1" }, "names must be a non-empty array of tenant names"],
+    [{ names: ["acme", "  "], pinnedVersion: "v1" }, "names must be a non-empty array of tenant names"],
+    [{ names: ["acme", 7], pinnedVersion: "v1" }, "names must be a non-empty array of tenant names"],
+    [{ names: ["acme", "acme"], pinnedVersion: "v1" }, "names must not repeat a tenant"],
+    [{ names: ["acme"], pinnedVersion: "   " }, "pinnedVersion must be a non-blank string, or null to unpin"],
+    [{ names: ["acme"], pinnedVersion: 7 }, "pinnedVersion must be a non-blank string, or null to unpin"],
+    [{ names: ["acme"] }, "pinnedVersion must be a non-blank string, or null to unpin"],
+  ] as const) {
+    const res = await rollout(app, body);
+    assert.equal(res.status, 400, JSON.stringify(body));
+    assert.deepEqual(await res.json(), { error: "invalid_request", message });
+  }
+  assert.deepEqual((await registry.get("acme"))?.tenant, { name: "acme" });
+});
+
+test("POST /v1/tenants/rollout is all-or-nothing: one unknown name 404s and applies nothing", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({ tenant: { name: "acme" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  const app = createTenantHttpApp(baseDeps({ registry }));
+
+  const res = await rollout(app, { names: ["acme", "ghost"], pinnedVersion: "v1.4.2" });
+
+  assert.equal(res.status, 404);
+  assert.deepEqual(await res.json(), { error: "tenant_not_found", message: 'unknown tenant "ghost"' });
+  assert.deepEqual((await registry.get("acme"))?.tenant, { name: "acme" });
+});
+
+test("POST /v1/tenants/rollout 409s a torn-down tenant and applies nothing", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({ tenant: { name: "acme" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  await registry.upsert({ tenant: { name: "gone" }, product: "orb", state: "torn down", createdAt: "t0", updatedAt: "t0" });
+  const app = createTenantHttpApp(baseDeps({ registry }));
+
+  const res = await rollout(app, { names: ["acme", "gone"], pinnedVersion: "v1.4.2" });
+
+  assert.equal(res.status, 409);
+  assert.deepEqual(await res.json(), { error: "tenant_torn_down", message: 'tenant "gone" is torn down' });
+  assert.deepEqual((await registry.get("acme"))?.tenant, { name: "acme" });
+});
+
+test("POST /v1/tenants/rollout sits behind the same Bearer wall as every other /v1/tenants route", async () => {
+  const app = createTenantHttpApp(baseDeps());
+
+  const res = await app.request("/v1/tenants/rollout", { method: "POST", body: JSON.stringify({ names: ["acme"], pinnedVersion: "v1" }) });
+
+  assert.equal(res.status, 401);
+  assert.deepEqual(await res.json(), { error: "unauthorized" });
+});
+
+test("GET /v1/tenants surfaces each tenant's pinnedVersion once one is set (#4898 admin visibility)", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({ tenant: { name: "acme", pinnedVersion: "v1.4.2" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" });
+  const app = createTenantHttpApp(baseDeps({ registry }));
+
+  const res = await app.request("/v1/tenants", authed());
+
+  assert.equal(res.status, 200);
+  assert.deepEqual(await res.json(), {
+    tenants: [{ tenant: { name: "acme", pinnedVersion: "v1.4.2" }, product: "orb", state: "active", createdAt: "t0", updatedAt: "t0" }],
+  });
+});

control-plane/test/tenant-registry.test.ts

@@ -110,3 +110,15 @@ test("createKvTenantRegistry: list tolerates a key disappearing between the list
 
   assert.deepEqual(await registry.list(), []);
 });
+
+test("a tenant's pinnedVersion (#4898) survives the KV JSON round-trip, and its absence stays absent", async () => {
+  const kv = fakeKv();
+  const registry = createKvTenantRegistry(kv);
+
+  await registry.upsert({ ...recordFor("acme"), tenant: { name: "acme", pinnedVersion: "v1.4.2" } });
+  await registry.upsert(recordFor("beta"));
+
+  assert.deepEqual((await registry.get("acme"))?.tenant, { name: "acme", pinnedVersion: "v1.4.2" });
+  // A pre-#4898 record (no pinnedVersion key at all) reads back exactly as stored — unpinned.
+  assert.deepEqual((await registry.get("beta"))?.tenant, { name: "beta" });
+});