CL-7218 workflow run pack push state slotsBlockedAddresses is never

apps/sidecar/src/index.ts

@@ -300,8 +300,9 @@ const orchestrator = createSidecarOrchestrator({
         deploymentAddressRegistry.record(deploymentId, agentAddress);
         bootRestorePushHold.onDeploymentRegistered(agentAddress);
       },
-      unregisterDeployment: ({ deploymentId }) => {
+      unregisterDeployment: ({ deploymentId, agentAddress }) => {
         deploymentAddressRegistry.unregister(deploymentId);
+        wrappedRepoStore.reclaimPushState({ deploymentId, agentAddress });
       },
       // Lazy-bound the same way as `workflowRunPackClient.hubLink` above:
       // `createDeployRouter` runs synchronously during `createSidecarOrchestrator`

apps/sidecar/src/workflow-run-pack-client.test.ts

@@ -0,0 +1,86 @@
+import { expect, test } from "bun:test";
+
+import type { Principal, RepoId, RepoStore } from "@intx/hub-sessions";
+
+import {
+  createDeploymentAddressRegistry,
+  createWorkflowRunPackPushingRepoStore,
+  type WorkflowRunPackClient,
+} from "./workflow-run-pack-client";
+
+const REF = "refs/heads/main";
+const DEPLOYMENT_ID = "dep-reclaim";
+const AGENT_ADDRESS = "run_reclaim@bench.localhost";
+const REPO_ID: RepoId = { kind: "workflow-run", id: DEPLOYMENT_ID };
+const PRINCIPAL: Principal = { kind: "hub" };
+
+function stubRepoStore(): RepoStore {
+  const result = { commitSha: "sha", newlyTerminalRuns: [] };
+  const unused = async () => result;
+  return {
+    initRepo: async () => undefined,
+    writeTree: unused,
+    writeTreePreservingPrefix: unused,
+    writeTreeDelta: unused,
+    receivePack: async () => ({ newlyTerminalRuns: [] }),
+    createPack: async () => ({ pack: new Uint8Array(), commitSha: "sha" }),
+    commitPackedTip: () => undefined,
+    resolveRef: async () => "sha",
+    listRefs: async () => [],
+    resolveHead: async () => "sha",
+    getRepoDir: () => "/tmp",
+    subscribe: () => () => undefined,
+    openCommittedReads: unused,
+    openCommittedReadsAtCommit: unused,
+  } as unknown as RepoStore;
+}
+
+const deltaArgs = {
+  computeDelta: async () => ({ puts: {}, deletes: [] as const }),
+  changedPathPrefixes: undefined,
+  message: "test",
+};
+
+test("unregister reclaim drops the push slot so flush does not wait on an in-flight push", async () => {
+  let pushCalls = 0;
+  let releaseFirst!: () => void;
+  const firstPush = new Promise<void>((resolve) => {
+    releaseFirst = resolve;
+  });
+  const packClient: WorkflowRunPackClient = {
+    async push() {
+      pushCalls += 1;
+      if (pushCalls === 1) await firstPush;
+    },
+    markRestored() {},
+  };
+  const registry = createDeploymentAddressRegistry();
+  registry.record(DEPLOYMENT_ID, AGENT_ADDRESS);
+  const store = createWorkflowRunPackPushingRepoStore({
+    underlying: stubRepoStore(),
+    packClient,
+    registry,
+  });
+
+  await store.writeTreeDelta(PRINCIPAL, REPO_ID, REF, deltaArgs);
+  expect(pushCalls).toBe(1);
+
+  store.markAddressUnroutable(AGENT_ADDRESS);
+  store.reclaimPushState({
+    deploymentId: DEPLOYMENT_ID,
+    agentAddress: AGENT_ADDRESS,
+  });
+
+  await Promise.race([
+    store.flushWorkflowRunPushes(REPO_ID, REF),
+    new Promise<void>((_, reject) => {
+      setTimeout(() => reject(new Error("flush hung on a residual slot")), 50);
+    }),
+  ]);
+
+  registry.record(DEPLOYMENT_ID, AGENT_ADDRESS);
+  await store.writeTreeDelta(PRINCIPAL, REPO_ID, REF, deltaArgs);
+  expect(pushCalls).toBe(2);
+
+  releaseFirst();
+});

apps/sidecar/src/workflow-run-pack-client.ts

@@ -647,6 +647,16 @@ export type WorkflowRunPackPushingRepoStore = RepoStore & {
    * failing -- is what lets the reconnect re-ship wait for the challenge.
    */
   markAddressUnroutable: (agentAddress: string) => void;
+  /**
+   * Drop coalescing slots and routability blocks for a torn-down
+   * deployment. Called from the boot edge's `unregisterDeployment`
+   * hook so an undeploy does not leave a slot (and its blocked address)
+   * in this process-wide map forever.
+   */
+  reclaimPushState: (entry: {
+    deploymentId: string;
+    agentAddress: string;
+  }) => void;
 };
 
 export function createWorkflowRunPackPushingRepoStore(
@@ -767,6 +777,18 @@ export function createWorkflowRunPackPushingRepoStore(
     blockedAddresses.add(agentAddress);
   }
 
+  function reclaimPushState(entry: {
+    deploymentId: string;
+    agentAddress: string;
+  }): void {
+    blockedAddresses.delete(entry.agentAddress);
+    for (const [key, slot] of slots) {
+      if (slot.repoId.id === entry.deploymentId) {
+        slots.delete(key);
+      }
+    }
+  }
+
   function notifyAddressRoutable(agentAddress: string): void {
     // The reconnect challenge re-routed this address on the hub. Clear the
     // block and re-drive so a push the disconnect cancelled -- or one held
@@ -833,6 +855,7 @@ export function createWorkflowRunPackPushingRepoStore(
     flushWorkflowRunPushes,
     notifyAddressRoutable,
     markAddressUnroutable,
+    reclaimPushState,
     async writeTreePreservingPrefix(principal, repoId, ref, args) {
       if (repoId.kind === "workflow-run") {
         const latched = takeLatchedError(repoId, ref);