From 33756b03d245367e1293fbc8374f264bb90f27c3 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 00:49:00 -0700 Subject: [PATCH 1/2] Add tests for reclaiming workflow-run pack push state --- .../src/workflow-run-pack-client.test.ts | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 apps/sidecar/src/workflow-run-pack-client.test.ts diff --git a/apps/sidecar/src/workflow-run-pack-client.test.ts b/apps/sidecar/src/workflow-run-pack-client.test.ts new file mode 100644 index 000000000..2c332fe3a --- /dev/null +++ b/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((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((_, 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(); +}); From a2c768ef3b3e7b77f558d4d662df57fa7713755e Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 00:49:08 -0700 Subject: [PATCH 2/2] Reclaim workflow-run pack push slots on undeploy unregisterDeployment now drops the coalescing slot and blocked address for that deployment so a torn-down run cannot leak push state for the life of the sidecar process. --- apps/sidecar/src/index.ts | 3 ++- apps/sidecar/src/workflow-run-pack-client.ts | 23 ++++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/apps/sidecar/src/index.ts b/apps/sidecar/src/index.ts index b2a136e7d..bc443c887 100644 --- a/apps/sidecar/src/index.ts +++ b/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` diff --git a/apps/sidecar/src/workflow-run-pack-client.ts b/apps/sidecar/src/workflow-run-pack-client.ts index 10f255542..e983bb25d 100644 --- a/apps/sidecar/src/workflow-run-pack-client.ts +++ b/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);