test(supervisor): test the pod-count logic as pure functions, drop the fake client

Author: nicktrn

apps/supervisor/src/backpressure/k8sPodCountSignalSource.test.ts

@@ -1,48 +1,41 @@
 import { describe, it, expect } from "vitest";
 import { K8sPodCountSignalSource } from "./k8sPodCountSignalSource.js";
-import { createPodCountFetcher } from "../clients/kubernetes.js";
-import type { K8sApi } from "../clients/kubernetes.js";
+import { podCountFromList, withTimeout } from "../clients/kubernetes.js";
 
-function apiReturning(list: unknown): K8sApi {
-  return { core: { listNamespacedPod: async () => list } } as unknown as K8sApi;
-}
-
-describe("createPodCountFetcher", () => {
-  it("returns items.length when the list is not truncated", async () => {
-    const fetch = createPodCountFetcher(
-      apiReturning({ items: [{}], metadata: {} }),
-      "v4-runs",
-      1000
-    );
-    expect(await fetch()).toBe(1);
+describe("podCountFromList", () => {
+  it("returns items.length when the list is not truncated", () => {
+    expect(podCountFromList({ items: [{}], metadata: {} })).toBe(1);
   });
 
-  it("returns zero for an empty namespace", async () => {
-    const fetch = createPodCountFetcher(apiReturning({ items: [], metadata: {} }), "v4-runs", 1000);
-    expect(await fetch()).toBe(0);
+  it("returns zero for an empty namespace", () => {
+    expect(podCountFromList({ items: [], metadata: {} })).toBe(0);
   });
 
-  it("adds remainingItemCount when the list is truncated", async () => {
+  it("adds remainingItemCount when the list is truncated", () => {
     const list = { items: [{}], metadata: { _continue: "tok", remainingItemCount: 24492 } };
-    const fetch = createPodCountFetcher(apiReturning(list), "v4-runs", 1000);
-    expect(await fetch()).toBe(24493);
+    expect(podCountFromList(list)).toBe(24493);
   });
 
-  it("throws when truncated but remainingItemCount is absent", async () => {
+  it("throws when truncated but remainingItemCount is absent", () => {
     const list = { items: [{}], metadata: { _continue: "tok" } };
-    const fetch = createPodCountFetcher(apiReturning(list), "v4-runs", 1000);
-    await expect(fetch()).rejects.toThrow(/remainingItemCount/);
+    expect(() => podCountFromList(list)).toThrow(/remainingItemCount/);
   });
 
-  it("throws when truncated but remainingItemCount is negative", async () => {
+  it("throws when truncated but remainingItemCount is negative", () => {
     const list = { items: [{}], metadata: { _continue: "tok", remainingItemCount: -1 } };
-    const fetch = createPodCountFetcher(apiReturning(list), "v4-runs", 1000);
-    await expect(fetch()).rejects.toThrow(/remainingItemCount/);
+    expect(() => podCountFromList(list)).toThrow(/remainingItemCount/);
+  });
+});
+
+describe("withTimeout", () => {
+  it("rejects once the deadline passes", async () => {
+    await expect(withTimeout(new Promise(() => {}), 10, "pod count list")).rejects.toThrow(
+      /timed out/
+    );
   });
 
-  it("rejects when the list hangs past the timeout", async () => {
-    const api = { core: { listNamespacedPod: () => new Promise(() => {}) } } as unknown as K8sApi;
-    await expect(createPodCountFetcher(api, "v4-runs", 10)()).rejects.toThrow(/timed out/);
+  it("passes a value through when it settles first", async () => {
+    await expect(withTimeout(Promise.resolve(7), 1000, "pod count list")).resolves.toBe(7);
   });
 });
 

apps/supervisor/src/clients/kubernetes.ts

@@ -70,38 +70,61 @@ export function createPodCountFetcher(
   timeoutMs: number
 ): () => Promise<number> {
   const serverTimeoutSeconds = Math.max(1, Math.floor(timeoutMs / 1000));
+  let pending: Promise<unknown> | undefined;
 
   return async () => {
-    const list = await withTimeout(
-      api.core.listNamespacedPod({
-        namespace,
-        limit: 1,
-        timeoutSeconds: serverTimeoutSeconds,
-      }),
-      timeoutMs,
-      "pod count list"
-    );
-
-    if (!list.metadata?._continue) {
-      return list.items.length; // not truncated, so this is all of them
+    if (pending) {
+      throw new Error("pod count list still in flight from a previous tick");
     }
 
-    const remaining = list.metadata.remainingItemCount;
-    if (typeof remaining !== "number" || !Number.isFinite(remaining) || remaining < 0) {
-      throw new Error("pod list truncated but remainingItemCount absent or invalid");
-    }
+    const request = api.core.listNamespacedPod({
+      namespace,
+      limit: 1,
+      timeoutSeconds: serverTimeoutSeconds,
+    });
+
+    pending = request
+      .catch(() => {})
+      .finally(() => {
+        pending = undefined;
+      });
 
-    return list.items.length + remaining;
+    return podCountFromList(await withTimeout(request, timeoutMs, "pod count list"));
   };
 }
 
+/**
+ * podCountFromList turns a `limit=1` pod list into a population.
+ *
+ * `remainingItemCount` is only set when the list is truncated, so `_continue` is the
+ * truncation signal: absent means the returned page is the whole collection and its
+ * length is already the answer. Truncated without a usable estimate is unknowable, so
+ * it throws rather than guessing a low number the brake would act on.
+ */
+export function podCountFromList(list: {
+  items: unknown[];
+  metadata?: { _continue?: string; remainingItemCount?: number };
+}): number {
+  if (!list.metadata?._continue) {
+    return list.items.length;
+  }
+
+  const remaining = list.metadata.remainingItemCount;
+  if (typeof remaining !== "number" || !Number.isFinite(remaining) || remaining < 0) {
+    throw new Error("pod list truncated but remainingItemCount absent or invalid");
+  }
+
+  return list.items.length + remaining;
+}
+
 /**
  * withTimeout rejects if `promise` outlives `timeoutMs`, so a hung request cannot
- * freeze the caller. A backstop only: the k8s client threads no AbortSignal through
- * to fetch, so callers must also bound the request server-side (`timeoutSeconds`),
- * or an abandoned request would stay open. The timer is cleared either way.
+ * freeze the caller. It cannot cancel: the k8s client threads no AbortSignal through to
+ * fetch, so an abandoned request keeps running. Callers must therefore also bound the
+ * request server-side (`timeoutSeconds`) and refuse to start a second one while the
+ * first is pending, or a blackholed connection accumulates one socket per attempt.
  */
-function withTimeout<T>(promise: Promise<T>, timeoutMs: number, what: string): Promise<T> {
+export function withTimeout<T>(promise: Promise<T>, timeoutMs: number, what: string): Promise<T> {
   let timer: NodeJS.Timeout;
   const deadline = new Promise<never>((_resolve, reject) => {
     timer = setTimeout(