From 09f5cc8f1c758f346967a72e6248956dc8ac340d Mon Sep 17 00:00:00 2001 From: Gregor Jarisch Date: Sun, 9 Aug 2026 13:01:03 +0200 Subject: [PATCH] fix(agents): wire the deployment-wait machinery that only the ZIP importer ever used Wave F of the Agent / Group Agent review. AgentFactory.getAgent has always had a branch for "the agent is deploying right now" -- waitForDeploymentCompletion, which awaits a future from DeploymentListener. That future was only ever registered by ONE caller in all of src/main: RestImportService, the startup ZIP importer. Every ordinary deploy fired onDeploymentEvent but never registered, so getRegisteredDeploymentEvent returned null, the wait had nothing to await, and a caller racing a deployment simply got a null agent. The machinery was dead outside one flow while reading as live. - RestAgentAdministration.deploy now registers BEFORE starting the deployment. Ordering matters: agentFactory.deployAgent is what publishes the IN_PROGRESS placeholder a waiter can observe, so registering afterwards would leave exactly the window this closes. - DeploymentListener.registerAgentDeployment self-expires. The map was pruned only by an arriving DeploymentEvent, so a registration whose event never came (a rejected deployment callable, a process that died mid-deploy) stayed for the lifetime of the JVM. Registrations now carry a REGISTRATION_TTL (5 minutes -- a leak bound, not a deployment SLA) and remove themselves on any completion. Removal is remove(key, future), not remove(key), so a stale completion cannot evict a live registration for the same agent. - RestImportService's allOf(...).join() is now tolerant. It could previously only block forever; with self-expiring registrations it can complete exceptionally, and one initial agent that never reports must not hang startup. 347 tests green; 12 new. --- docs/changelog.md | 14 ++ .../eddi/backup/impl/RestImportService.java | 18 +- .../internal/RestAgentAdministration.java | 12 ++ .../runtime/internal/DeploymentListener.java | 49 ++++- .../DeploymentListenerRegistrationTest.java | 170 ++++++++++++++++++ 5 files changed, 260 insertions(+), 3 deletions(-) create mode 100644 src/test/java/ai/labs/eddi/engine/runtime/internal/DeploymentListenerRegistrationTest.java diff --git a/docs/changelog.md b/docs/changelog.md index d2f80a1af0..6dd668479e 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -3,6 +3,20 @@ > **Purpose:** Living document tracking all changes, decisions, and reasoning during implementation. Updated as work progresses for easy reference and review. +--- + +## 🔗 fix(agents): wire the deployment-wait machinery that only the ZIP importer ever used (2026-08-09) + +**Repo:** EDDI (`fix/deployment-wait-machinery`) + +Wave F of the Agent / Group Agent review. `AgentFactory.getAgent` has always had a branch for "the agent is deploying right now" — `waitForDeploymentCompletion`, which awaits a future from `DeploymentListener`. That future was only ever registered by **one** caller in all of `src/main`: `RestImportService`, the startup ZIP importer. Every ordinary deploy fired `onDeploymentEvent` but never registered, so `getRegisteredDeploymentEvent` returned `null`, the wait had nothing to await, and a caller racing a deployment simply got a null agent. The machinery was dead outside one flow while reading as live. + +- `RestAgentAdministration.deploy` now registers **before** starting the deployment. Ordering matters: `agentFactory.deployAgent` is what publishes the IN_PROGRESS placeholder a waiter can observe, so registering afterwards would leave exactly the window this closes. +- `DeploymentListener.registerAgentDeployment` self-expires. The map was pruned only by an arriving `DeploymentEvent`, so a registration whose event never came — a rejected deployment callable, a process that died mid-deploy — stayed for the lifetime of the JVM. Registrations now carry a `REGISTRATION_TTL` (5 minutes, a leak bound rather than a deployment SLA) and remove themselves on *any* completion. Removal is `remove(key, future)`, not `remove(key)`, so a stale completion cannot evict a live registration for the same agent. +- `RestImportService`'s `allOf(...).join()` is now tolerant. It could previously only block forever; with self-expiring registrations it can complete exceptionally, and one initial agent that never reports must not hang startup — it logs and continues with the agents that did deploy. + +Suites: 347 tests green across `DeploymentListener*`, `RestAgentAdministration*`, `AgentFactory*`, `RestImportService*`; 12 new. + --- ## 🔎 fix(groups): pre-merge deep review — facilitator HITL bypass, CALL_VOTE guards, metric cardinality, template honesty (2026-08-08) diff --git a/src/main/java/ai/labs/eddi/backup/impl/RestImportService.java b/src/main/java/ai/labs/eddi/backup/impl/RestImportService.java index a9f06df96a..300f35a714 100644 --- a/src/main/java/ai/labs/eddi/backup/impl/RestImportService.java +++ b/src/main/java/ai/labs/eddi/backup/impl/RestImportService.java @@ -146,8 +146,22 @@ public List importInitialAgents() { } } - // Wait for all deployments to complete - CompletableFuture.allOf(deploymentFutures.toArray(new CompletableFuture[0])).join(); + // Wait for all deployments to complete. + // + // Tolerant of a failure, and bounded: a registration now self-expires + // (DeploymentListener.REGISTRATION_TTL), so this can complete + // exceptionally where it previously could only block forever. One initial + // agent that never reports must not hang startup — log it and carry on + // with the ones that did deploy. + CompletableFuture.allOf(deploymentFutures.toArray(new CompletableFuture[0])) + .handle((ignored, error) -> { + if (error != null) { + LOGGER.warnf("Not every initial agent reported a deployment event (%s) — continuing with those that did", + error.getMessage()); + } + return null; + }) + .join(); LOGGER.info("Imported & Deployed Initial Agents"); return restAgentAdministration.getDeploymentStatuses(production); diff --git a/src/main/java/ai/labs/eddi/engine/internal/RestAgentAdministration.java b/src/main/java/ai/labs/eddi/engine/internal/RestAgentAdministration.java index 21024f23df..5eeb86cd49 100644 --- a/src/main/java/ai/labs/eddi/engine/internal/RestAgentAdministration.java +++ b/src/main/java/ai/labs/eddi/engine/internal/RestAgentAdministration.java @@ -214,6 +214,18 @@ private void enforceAgentQuota(Deployment.Environment environment, String agentI } private Future deploy(final Deployment.Environment environment, final String agentId, final Integer version, final Boolean autoDeploy) { + // Register BEFORE the deployment starts, so a concurrent getAgent that finds + // the agent IN_PROGRESS has a future to await. + // + // This is what makes AgentFactory's wait machinery reachable at all. Only + // RestImportService ever registered, so for every ordinary deploy + // getRegisteredDeploymentEvent returned null, waitForDeploymentCompletion had + // nothing to await, and a caller racing a deployment simply got a null agent. + // Registration must precede agentFactory.deployAgent: that call is what + // publishes the IN_PROGRESS placeholder a waiter can observe, so registering + // afterwards would leave exactly the window this closes. + deploymentListener.registerAgentDeployment(agentId, version); + Callable deployAgentCallable = () -> { try { if (EnumSet.of(NOT_FOUND, ERROR).contains(checkDeploymentStatus(environment, agentId, version))) { diff --git a/src/main/java/ai/labs/eddi/engine/runtime/internal/DeploymentListener.java b/src/main/java/ai/labs/eddi/engine/runtime/internal/DeploymentListener.java index c321199530..cd6d29a042 100644 --- a/src/main/java/ai/labs/eddi/engine/runtime/internal/DeploymentListener.java +++ b/src/main/java/ai/labs/eddi/engine/runtime/internal/DeploymentListener.java @@ -5,17 +5,41 @@ package ai.labs.eddi.engine.runtime.internal; import ai.labs.eddi.engine.runtime.model.DeploymentEvent; +import ai.labs.eddi.utils.LogSanitizer; import jakarta.enterprise.context.ApplicationScoped; +import org.jboss.logging.Logger; +import java.time.Duration; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import static ai.labs.eddi.engine.model.Deployment.Status.ERROR; import static ai.labs.eddi.engine.model.Deployment.Status.READY; @ApplicationScoped public class DeploymentListener implements IDeploymentListener { + + private static final Logger LOGGER = Logger.getLogger(DeploymentListener.class); + + /** + * How long a registration may sit unresolved before it self-expires. + *

+ * A registration is removed when its {@link DeploymentEvent} arrives, and the + * deploy path fires one on both the success and the failure branch — but a + * process that dies in between, or a runtime that rejects the deployment + * callable outright, leaves an entry no event will ever claim. Nothing swept + * this map, so those accumulated for the lifetime of the JVM. + *

+ * Generous by design: this bounds a leak, it is not a deployment SLA. A + * deployment that takes longer than this has other problems, and a waiter that + * wants a tighter bound arms its own (see + * {@code AgentFactory#waitForDeploymentCompletion}, which waits 60s). + */ + static final Duration REGISTRATION_TTL = Duration.ofMinutes(5); + private final Map> deploymentFutures = new ConcurrentHashMap<>(); @Override @@ -23,8 +47,31 @@ public CompletableFuture getRegisteredDeploymentEvent(String agentId, Inte return deploymentFutures.get(createKey(agentId, version)); } + /** + * Registers interest in a deployment, so a concurrent + * {@code AgentFactory#getAgent} that finds the agent IN_PROGRESS has something + * to await instead of falling straight through. + *

+ * The returned future self-expires after {@link #REGISTRATION_TTL} and removes + * itself from the map on any completion — normal, exceptional or + * timed-out — so the map cannot grow without bound when an event never arrives. + */ public CompletableFuture registerAgentDeployment(String agentId, Integer version) { - return deploymentFutures.computeIfAbsent(createKey(agentId, version), k -> new CompletableFuture<>()); + return deploymentFutures.computeIfAbsent(createKey(agentId, version), key -> { + var future = new CompletableFuture(); + // orTimeout returns THIS future, so the timeout arms the entry itself + // rather than a derived one nobody holds. + future.orTimeout(REGISTRATION_TTL.toSeconds(), TimeUnit.SECONDS); + future.whenComplete((result, error) -> { + // remove(key, future), not remove(key): a later registration for the + // same agent+version must not be evicted by this one's completion. + if (deploymentFutures.remove(key, future) && error instanceof TimeoutException) { + LOGGER.warnf("No deployment event arrived for %s within %s — dropping its registration", + LogSanitizer.sanitize(key), REGISTRATION_TTL); + } + }); + return future; + }); } public void onDeploymentEvent(DeploymentEvent event) { diff --git a/src/test/java/ai/labs/eddi/engine/runtime/internal/DeploymentListenerRegistrationTest.java b/src/test/java/ai/labs/eddi/engine/runtime/internal/DeploymentListenerRegistrationTest.java new file mode 100644 index 0000000000..a9791925ec --- /dev/null +++ b/src/test/java/ai/labs/eddi/engine/runtime/internal/DeploymentListenerRegistrationTest.java @@ -0,0 +1,170 @@ +/* + * Copyright EDDI contributors + * SPDX-License-Identifier: Apache-2.0 + */ +package ai.labs.eddi.engine.runtime.internal; + +import ai.labs.eddi.engine.model.Deployment; +import ai.labs.eddi.engine.runtime.model.DeploymentEvent; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Registration lifecycle in {@link DeploymentListener}. + *

+ * Two properties matter here. A registration must be discoverable + * while the deployment is in flight — that is what makes + * {@code AgentFactory#waitForDeploymentCompletion} able to wait at all — and it + * must not outlive the deployment it describes: the map was only ever pruned by + * an arriving {@link DeploymentEvent}, so a registration whose event never came + * (a rejected callable, a process that died mid-deploy) stayed for the lifetime + * of the JVM. + * + * @author ginccc + */ +@DisplayName("DeploymentListener — registration lifecycle") +class DeploymentListenerRegistrationTest { + + private DeploymentListener listener; + + @BeforeEach + void setUp() { + listener = new DeploymentListener(); + } + + private static DeploymentEvent event(String agentId, Integer version, Deployment.Status status) { + return new DeploymentEvent(agentId, version, Deployment.Environment.production, status); + } + + @Test + @DisplayName("a registered deployment is discoverable while it is in flight") + void registeredDeploymentIsDiscoverable() { + CompletableFuture registered = listener.registerAgentDeployment("agent-1", 1); + + assertNotNull(registered); + assertSame(registered, listener.getRegisteredDeploymentEvent("agent-1", 1), + "a waiter must find the same future the deployer registered"); + } + + @Test + @DisplayName("an unregistered deployment is not discoverable") + void unregisteredDeploymentIsNotDiscoverable() { + assertNull(listener.getRegisteredDeploymentEvent("never-registered", 1)); + } + + @Test + @DisplayName("registering twice for the same agent+version hands back the same future") + void repeatedRegistrationIsIdempotent() { + var first = listener.registerAgentDeployment("agent-1", 1); + var second = listener.registerAgentDeployment("agent-1", 1); + + assertSame(first, second); + } + + @Test + @DisplayName("different versions register independently") + void versionsAreIndependent() { + var v1 = listener.registerAgentDeployment("agent-1", 1); + var v2 = listener.registerAgentDeployment("agent-1", 2); + + assertNotSame(v1, v2); + } + + @Test + @DisplayName("a READY event completes the registration and removes it") + void readyEventCompletesAndRemoves() { + var registered = listener.registerAgentDeployment("agent-1", 1); + + listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.READY)); + + assertTrue(registered.isDone()); + assertNull(listener.getRegisteredDeploymentEvent("agent-1", 1), "a settled registration must not linger"); + } + + @Test + @DisplayName("an ERROR event completes it exceptionally and removes it") + void errorEventCompletesExceptionallyAndRemoves() { + var registered = listener.registerAgentDeployment("agent-1", 1); + + listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.ERROR)); + + assertTrue(registered.isCompletedExceptionally()); + assertNull(listener.getRegisteredDeploymentEvent("agent-1", 1)); + } + + @Test + @DisplayName("a registration whose event never arrives expires instead of leaking") + void unclaimedRegistrationExpires() { + var registered = listener.registerAgentDeployment("agent-1", 1); + + // Stand in for the TTL firing: any exceptional completion must evict the + // entry. Asserting the real 5-minute timeout would mean a 5-minute test. + registered.completeExceptionally(new TimeoutException("simulated TTL")); + + assertNull(listener.getRegisteredDeploymentEvent("agent-1", 1), + "the map was only ever pruned by an arriving event, so an unclaimed registration stayed for the " + + "lifetime of the JVM"); + } + + @Test + @DisplayName("the registration carries a timeout, so it cannot wait forever") + void registrationCarriesATimeout() { + assertTrue(DeploymentListener.REGISTRATION_TTL.toSeconds() > 0, + "an unbounded registration is exactly the leak this bounds"); + } + + @Test + @DisplayName("an expired registration does not evict a later one for the same agent") + void expiryDoesNotEvictALaterRegistration() { + var first = listener.registerAgentDeployment("agent-1", 1); + listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.READY)); + + var second = listener.registerAgentDeployment("agent-1", 1); + assertNotSame(first, second, "precondition: the first was evicted, so this is a fresh registration"); + + // The first future's completion callback must not remove the second entry. + first.complete(null); + + assertSame(second, listener.getRegisteredDeploymentEvent("agent-1", 1), + "remove(key, future) — a stale completion must not evict a live registration"); + } + + @Test + @DisplayName("an event for an unregistered deployment is a no-op, not an error") + void eventForUnregisteredDeploymentIsANoOp() { + listener.onDeploymentEvent(event("ghost", 7, Deployment.Status.READY)); + listener.onDeploymentEvent(event("ghost", 7, Deployment.Status.ERROR)); + + assertNull(listener.getRegisteredDeploymentEvent("ghost", 7)); + } + + @Test + @DisplayName("a completed registration is awaitable by the waiter that holds it") + void completedRegistrationIsAwaitable() throws Exception { + var registered = listener.registerAgentDeployment("agent-1", 1); + listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.READY)); + + registered.get(5, TimeUnit.SECONDS); + } + + @Test + @DisplayName("a failed registration surfaces the failure to its waiter") + void failedRegistrationSurfacesToWaiter() { + var registered = listener.registerAgentDeployment("agent-1", 1); + listener.onDeploymentEvent(event("agent-1", 1, Deployment.Status.ERROR)); + + assertThrows(Exception.class, () -> registered.get(5, TimeUnit.SECONDS)); + } +}