fix(control-plane): isolate one AMS tenant's wake failure from the rest of the tick

control-plane/src/ams-wake.ts

@@ -44,6 +44,11 @@ export type AmsWakeResult = {
    *  silently coerced into a fake exit code). */
   exitCode: number | undefined;
   timedOut: boolean;
+  /** `true` when this tenant's wake threw at any point (`start()`, `pollForExitCode`'s `getState()`, or
+   *  either `registry.upsert()` call) and was caught so the rest of the tick could keep going. `exitCode` is
+   *  always `undefined` and `timedOut` is always `false` on a failed result -- neither one describes a wake
+   *  that never got the chance to finish. */
+  failed: boolean;
 };
 
 const HOSTED_ENTRY_BIN = "loopover-miner-hosted";
@@ -129,28 +134,51 @@ export async function wakeDueAmsTenants(config: AmsWakeConfig): Promise<AmsWakeR
     // Claim this tenant's next slot BEFORE starting its poll (#9143) -- see this function's own header
     // comment on why.
     const claimedNextDueAt = new Date(tickStartedAt.getTime() + schedule.intervalMs).toISOString();
-    await config.registry.upsert({
-      ...record,
-      amsSchedule: { ...schedule, nextDueAt: claimedNextDueAt },
-      updatedAt: now().toISOString(),
-    });
-
-    const stub = config.binding.getByName(instanceNameFor(record.tenant.name, record.product));
-    await stub.start({ entrypoint: [HOSTED_ENTRY_BIN, schedule.command, ...schedule.args] });
-    const { exitCode, timedOut } = await pollForExitCode(stub, pollIntervalMs, pollTimeoutMs);
-
-    const ranAt = now().toISOString();
-    await config.registry.upsert({
-      ...record,
-      amsSchedule: {
-        ...schedule,
-        lastRunAt: ranAt,
-        lastExitCode: exitCode,
-        nextDueAt: claimedNextDueAt,
-      },
-      updatedAt: ranAt,
-    });
-    results.push({ tenant: record.tenant, ranAt, exitCode, timedOut });
+
+    // Isolate this one tenant's wake: a rejection from `start()`, from `pollForExitCode`'s `getState()`, or
+    // from either `upsert()` below must cost only this tenant its cycle, not the whole tick (#10063) --
+    // mirroring http-app.ts's `provisionTenant` failure seam, which likewise records a terminal state before
+    // letting its caller move on instead of losing the surrounding request.
+    try {
+      await config.registry.upsert({
+        ...record,
+        amsSchedule: { ...schedule, nextDueAt: claimedNextDueAt },
+        updatedAt: now().toISOString(),
+      });
+
+      const stub = config.binding.getByName(instanceNameFor(record.tenant.name, record.product));
+      await stub.start({ entrypoint: [HOSTED_ENTRY_BIN, schedule.command, ...schedule.args] });
+      const { exitCode, timedOut } = await pollForExitCode(stub, pollIntervalMs, pollTimeoutMs);
+
+      const ranAt = now().toISOString();
+      await config.registry.upsert({
+        ...record,
+        amsSchedule: {
+          ...schedule,
+          lastRunAt: ranAt,
+          lastExitCode: exitCode,
+          nextDueAt: claimedNextDueAt,
+        },
+        updatedAt: ranAt,
+      });
+      results.push({ tenant: record.tenant, ranAt, exitCode, timedOut, failed: false });
+    } catch {
+      const ranAt = now().toISOString();
+      try {
+        // Still attempt the lastRunAt write so the record doesn't silently look like it never ran --
+        // whichever upsert above threw (or never ran at all), this is a fresh attempt of its own.
+        await config.registry.upsert({
+          ...record,
+          amsSchedule: { ...schedule, lastRunAt: ranAt, nextDueAt: claimedNextDueAt },
+          updatedAt: ranAt,
+        });
+      } catch {
+        // The failure write itself failed too -- nothing left to attempt this tick. The claimed
+        // `nextDueAt` (if that first upsert landed) still moved forward, so this tenant is retried next
+        // cycle rather than starving the rest of this one.
+      }
+      results.push({ tenant: record.tenant, ranAt, exitCode: undefined, timedOut: false, failed: true });
+    }
   }
 
   return results;

control-plane/test/ams-wake.test.ts

@@ -8,6 +8,7 @@ import {
   wakeDueAmsTenants,
   type AmsWakeConfig,
   type TenantRegistry,
+  type TenantRegistryRecord,
   type WakeNamespaceLike,
   type WakeStubLike,
 } from "../dist/index.js";
@@ -46,6 +47,56 @@ function fakeNamespace(stubs: Record<string, FakeWakeStub>): WakeNamespaceLike &
   };
 }
 
+// A stub whose start() (or getState(), depending on which array holds a rejecting entry) throws instead of
+// resolving -- used to simulate #10063's unreachable-container failure modes. `starts`/`getStateCalls` are
+// tracked the same way fakeWakeStub's are, so assertions on call counts/entrypoints still work.
+function throwingStartStub(): FakeWakeStub {
+  const starts: Array<{ entrypoint?: string[] }> = [];
+  return {
+    starts,
+    getStateCalls: 0,
+    async start(options) {
+      starts.push(options ?? {});
+      throw new Error("container failed to start");
+    },
+    async getState() {
+      throw new Error("should not be called: start() already rejected");
+    },
+  };
+}
+
+function throwingGetStateStub(): FakeWakeStub {
+  const starts: Array<{ entrypoint?: string[] }> = [];
+  let getStateCalls = 0;
+  return {
+    starts,
+    get getStateCalls() {
+      return getStateCalls;
+    },
+    async start(options) {
+      starts.push(options ?? {});
+    },
+    async getState() {
+      getStateCalls += 1;
+      throw new Error("container health check failed");
+    },
+  };
+}
+
+// Wraps a registry so `upsert()` rejects for any record matching `shouldFail` -- PERSISTENTLY (every
+// matching call, not just the first), so a test can exercise both the outer per-tenant catch and its own
+// inner catch (the "the failure write itself also failed" branch) in one go, rather than having the retry
+// inside the catch quietly succeed.
+function withFailingUpsert(base: TenantRegistry, shouldFail: (record: TenantRegistryRecord) => boolean): TenantRegistry {
+  return {
+    ...base,
+    async upsert(record) {
+      if (shouldFail(record)) throw new Error("registry upsert rejected");
+      await base.upsert(record);
+    },
+  };
+}
+
 function baseConfig(overrides: Partial<AmsWakeConfig> & { binding: WakeNamespaceLike; registry: TenantRegistry }): AmsWakeConfig {
   return { pollIntervalMs: 1, pollTimeoutMs: 50, ...overrides };
 }
@@ -364,3 +415,181 @@ test("wakeDueAmsTenants: defaults now/pollIntervalMs/pollTimeoutMs when not give
 
   assert.equal(results.length, 1);
 });
+
+// #10063: one tenant's wake failing must not abort the tick for every OTHER due tenant, and the failed
+// tenant must still show up in the result array (flagged, not silently dropped) rather than being
+// indistinguishable from a tenant that never ran.
+test("wakeDueAmsTenants: the FIRST due tenant's start() rejecting still lets the second one wake", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({
+    tenant: { name: "acme" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  await registry.upsert({
+    tenant: { name: "beta" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "attempt", args: ["item-1"], intervalMs: 30_000, nextDueAt: PAST },
+  });
+  const betaStub = fakeWakeStub([{ status: "stopped_with_code", exitCode: 0 }]);
+  const namespace = fakeNamespace({ "ams:acme": throwingStartStub(), "ams:beta": betaStub });
+
+  const results = await wakeDueAmsTenants(baseConfig({ binding: namespace, registry, now: () => NOW }));
+
+  assert.equal(results.length, 2);
+  assert.equal(results[0]!.tenant.name, "acme");
+  assert.equal(results[0]!.failed, true);
+  assert.equal(results[0]!.exitCode, undefined);
+  assert.equal(results[0]!.timedOut, false);
+  assert.equal(results[1]!.tenant.name, "beta");
+  assert.equal(results[1]!.failed, false);
+  assert.equal(results[1]!.exitCode, 0);
+  assert.deepEqual(betaStub.starts, [{ entrypoint: ["loopover-miner-hosted", "attempt", "item-1"] }]);
+
+  // The failed tenant's lastRunAt write was still attempted, so it doesn't look like it never ran.
+  const acme = await registry.get("acme", "ams");
+  assert.equal(acme?.amsSchedule?.lastRunAt, results[0]!.ranAt);
+});
+
+test("wakeDueAmsTenants: a getState() rejection mid-poll is flagged failed, not timedOut, and the next tenant still runs", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({
+    tenant: { name: "acme" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  await registry.upsert({
+    tenant: { name: "beta" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  const namespace = fakeNamespace({ "ams:acme": throwingGetStateStub(), "ams:beta": fakeWakeStub([{ status: "stopped_with_code", exitCode: 0 }]) });
+
+  const results = await wakeDueAmsTenants(baseConfig({ binding: namespace, registry, now: () => NOW }));
+
+  assert.equal(results.length, 2);
+  assert.equal(results[0]!.failed, true);
+  assert.equal(results[0]!.timedOut, false);
+  assert.equal(results[0]!.exitCode, undefined);
+  assert.equal(results[1]!.tenant.name, "beta");
+  assert.equal(results[1]!.exitCode, 0);
+});
+
+test("wakeDueAmsTenants: the post-run registry.upsert rejecting is caught, and the next tenant still runs", async () => {
+  const baseRegistry = createFakeTenantRegistry();
+  await baseRegistry.upsert({
+    tenant: { name: "acme" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  await baseRegistry.upsert({
+    tenant: { name: "beta" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  // Fails every upsert carrying a lastRunAt for "acme" -- both the happy-path post-run write AND the
+  // catch's own retry of that same write, so this one test also exercises the catch's inner catch (the
+  // retry-also-throws branch), not just the outer one.
+  const registry = withFailingUpsert(baseRegistry, (record) => record.tenant.name === "acme" && record.amsSchedule?.lastRunAt !== undefined);
+  const namespace = fakeNamespace({
+    "ams:acme": fakeWakeStub([{ status: "stopped_with_code", exitCode: 0 }]),
+    "ams:beta": fakeWakeStub([{ status: "stopped_with_code", exitCode: 0 }]),
+  });
+
+  const results = await wakeDueAmsTenants(baseConfig({ binding: namespace, registry, now: () => NOW }));
+
+  assert.equal(results.length, 2);
+  assert.equal(results[0]!.tenant.name, "acme");
+  assert.equal(results[0]!.failed, true);
+  assert.equal(results[1]!.tenant.name, "beta");
+  assert.equal(results[1]!.failed, false);
+  assert.equal(results[1]!.exitCode, 0);
+  // "acme"'s claim (the FIRST upsert, which doesn't carry lastRunAt) landed fine -- only the write this
+  // registry was rigged to fail on never took, so the record still shows no lastRunAt at all.
+  const acme = await baseRegistry.get("acme", "ams");
+  assert.equal(acme?.amsSchedule?.lastRunAt, undefined);
+});
+
+test("wakeDueAmsTenants: a failed tenant still counts toward maxTenantsPerTick", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({
+    tenant: { name: "acme" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  await registry.upsert({
+    tenant: { name: "beta" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  const namespace = fakeNamespace({ "ams:acme": throwingStartStub(), "ams:beta": fakeWakeStub([{ status: "stopped_with_code", exitCode: 0 }]) });
+
+  const results = await wakeDueAmsTenants(baseConfig({ binding: namespace, registry, now: () => NOW, maxTenantsPerTick: 1 }));
+
+  assert.equal(results.length, 1);
+  assert.equal(results[0]!.tenant.name, "acme");
+  assert.equal(results[0]!.failed, true);
+  // "beta" was left completely untouched by the cap, exactly as a successful "acme" would have left it.
+  assert.deepEqual(namespace.requestedNames, ["ams:acme"]);
+  const beta = await registry.get("beta", "ams");
+  assert.equal(beta?.amsSchedule?.nextDueAt, PAST);
+});
+
+// #10063 regression: a permanently-broken tenant sorted FIRST by name must not starve every other due
+// tenant sorted after it, tick after tick -- before this fix, its thrown rejection aborted the whole
+// `wakeDueAmsTenants` call, so a healthy tenant later in the (name-sorted) list never even got reached.
+test("wakeDueAmsTenants: a permanently-failing tenant sorted first by name never blocks a healthy tenant sorted after it (#10063)", async () => {
+  const registry = createFakeTenantRegistry();
+  await registry.upsert({
+    tenant: { name: "aaa-broken" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  await registry.upsert({
+    tenant: { name: "zzz-healthy" },
+    product: "ams",
+    state: "active",
+    createdAt: "t0",
+    updatedAt: "t0",
+    amsSchedule: { command: "discover", args: [], intervalMs: 60_000, nextDueAt: PAST },
+  });
+  const healthyStub = fakeWakeStub([{ status: "stopped_with_code", exitCode: 0 }]);
+  const namespace = fakeNamespace({ "ams:aaa-broken": throwingStartStub(), "ams:zzz-healthy": healthyStub });
+
+  const results = await wakeDueAmsTenants(baseConfig({ binding: namespace, registry, now: () => NOW }));
+
+  assert.equal(results.length, 2);
+  assert.equal(results[0]!.tenant.name, "aaa-broken");
+  assert.equal(results[0]!.failed, true);
+  assert.equal(results[1]!.tenant.name, "zzz-healthy");
+  assert.equal(results[1]!.failed, false);
+  assert.equal(results[1]!.exitCode, 0);
+  assert.deepEqual(healthyStub.starts, [{ entrypoint: ["loopover-miner-hosted", "discover"] }]);
+});