diff --git a/core/src/main/java/art/arcane/iris/engine/EngineHotloader.java b/core/src/main/java/art/arcane/iris/engine/EngineHotloader.java index 3a6ee0db8..f83a08557 100644 --- a/core/src/main/java/art/arcane/iris/engine/EngineHotloader.java +++ b/core/src/main/java/art/arcane/iris/engine/EngineHotloader.java @@ -175,7 +175,7 @@ final class EngineHotloader { broadcastStudioHotload(false, ""); } catch (Throwable e) { if (!published) { - if (replacementData != null) { + if (replacementData != null && !engine.shutdownSequence.retainsData(replacementData)) { replacementData.unregisterEngine(engine); Throwable replacementDataFailure = runCleanup(null, replacementData::close); if (replacementDataFailure != null) { diff --git a/core/src/main/java/art/arcane/iris/engine/EngineRuntimeBuilder.java b/core/src/main/java/art/arcane/iris/engine/EngineRuntimeBuilder.java index 099affb6a..2dd1cb789 100644 --- a/core/src/main/java/art/arcane/iris/engine/EngineRuntimeBuilder.java +++ b/core/src/main/java/art/arcane/iris/engine/EngineRuntimeBuilder.java @@ -209,7 +209,12 @@ final class EngineRuntimeBuilder { next.generation().mantle(), null); if (retirementFailure != null) { - retirementFailure = engine.shutdownSequence.closeRuntime(next, retirementFailure); + Throwable cleanupFailure = engine.shutdownSequence.closeRuntime( + next, previous == null ? null : previous.generation().mantle(), null); + if (cleanupFailure != null) { + engine.shutdownSequence.retainUnpublishedRuntime(next); + retirementFailure = EngineShutdownSequence.appendFailure(retirementFailure, cleanupFailure); + } engine.lifecycleState = LifecycleState.FAILED; throw new IllegalStateException("Failed to retire the previous Iris engine runtime.", retirementFailure); } @@ -229,6 +234,7 @@ final class EngineRuntimeBuilder { engine.getClosing().set(true); engine.backgroundTasks.closeBackgroundTaskAdmission(); engine.lifecycleState = LifecycleState.FAILED; + Throwable cleanupFailure = null; try { engine.getGenerationSessions().sealAndAwait( "failed world manager start", @@ -236,11 +242,17 @@ final class EngineRuntimeBuilder { true ); } catch (Throwable drainFailure) { - e.addSuppressed(drainFailure); + cleanupFailure = drainFailure; } - engine.shutdownSequence.closeRuntime(next, e); - if (engine.runtime == next) { - engine.runtime = null; + if (cleanupFailure == null) { + cleanupFailure = engine.shutdownSequence.closeRuntime(next, null); + } + if (cleanupFailure == null) { + if (engine.runtime == next) { + engine.runtime = null; + } + } else { + e.addSuppressed(cleanupFailure); } throw new IllegalStateException("Failed to start the Iris world manager.", e); } diff --git a/core/src/main/java/art/arcane/iris/engine/EngineShutdownSequence.java b/core/src/main/java/art/arcane/iris/engine/EngineShutdownSequence.java index 5892d57d8..482ddb263 100644 --- a/core/src/main/java/art/arcane/iris/engine/EngineShutdownSequence.java +++ b/core/src/main/java/art/arcane/iris/engine/EngineShutdownSequence.java @@ -18,6 +18,7 @@ package art.arcane.iris.engine; +import art.arcane.iris.core.loader.IrisData; import art.arcane.iris.engine.EngineBackgroundTasks.BackgroundTaskDrain; import art.arcane.iris.engine.EngineRuntimeBuilder.RuntimeAssembly; import art.arcane.iris.engine.IrisEngine.LifecycleState; @@ -29,6 +30,11 @@ import art.arcane.iris.engine.mantle.EngineMantle; import art.arcane.iris.spi.IrisLogging; import art.arcane.iris.spi.IrisServices; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + /** * Owns the ordered teardown of a single {@link IrisEngine}, both for a normal close and for a * construction that failed part way through. Every resource release is gated on the previous one @@ -38,6 +44,10 @@ final class EngineShutdownSequence { private static final long CLOSE_RETRY_DRAIN_TIMEOUT_MILLIS = 5000L; private final IrisEngine engine; + private final Set incompleteAssemblies = ConcurrentHashMap.newKeySet(); + private final Set unpublishedRuntimes = ConcurrentHashMap.newKeySet(); + private final Set incompleteTargets = Collections.synchronizedSet( + Collections.newSetFromMap(new IdentityHashMap<>())); private boolean runtimeReleased; private boolean targetReleased; private boolean engineDataReleased; @@ -172,17 +182,21 @@ final class EngineShutdownSequence { EngineTickRegistry.unregisterTicking(engine); engine.lifecycleState = LifecycleState.FAILED; Throwable cleanupFailure = null; + boolean sessionsDrained = true; try { engine.getGenerationSessions().sealAndAwait("failed initialization", 0L, true); } catch (Throwable e) { + sessionsDrained = false; cleanupFailure = appendFailure(cleanupFailure, e); } engine.backgroundTasks.cancelBackgroundTasks("failed initialization"); BackgroundTaskDrain backgroundDrain = engine.backgroundTasks.drainBackgroundTasks("failed initialization"); cleanupFailure = appendFailure(cleanupFailure, backgroundDrain.failure()); - if (!backgroundDrain.allowsResourceRelease()) { - cleanupFailure = appendFailure(cleanupFailure, - new IllegalStateException("Iris background tasks remain active after failed initialization.")); + if (!sessionsDrained || !backgroundDrain.allowsResourceRelease()) { + if (!backgroundDrain.allowsResourceRelease()) { + cleanupFailure = appendFailure(cleanupFailure, + new IllegalStateException("Iris background tasks remain active after failed initialization.")); + } if (cleanupFailure != original) { original.addSuppressed(cleanupFailure); } @@ -192,27 +206,19 @@ final class EngineShutdownSequence { () -> NativeStructureOwnershipStore.close(engine)); cleanupFailure = appendFailure(cleanupFailure, ownershipFailure); if (ownershipFailure == null) { - Throwable runtimeFailure = closeDetachedGenerationRuntimes(null); - if (runtimeFailure == null) { - runtimeFailure = closeRuntime(engine.runtime, null); + cleanupFailure = releaseRuntime(cleanupFailure); + if (runtimeReleased) { + cleanupFailure = releaseTarget(cleanupFailure); } - cleanupFailure = appendFailure(cleanupFailure, runtimeFailure); - if (runtimeFailure != null) { - if (cleanupFailure != original) { - original.addSuppressed(cleanupFailure); - } - return; + if (targetReleased) { + cleanupFailure = releaseEngineDataForShutdown(cleanupFailure); + } + if (engineDataReleased) { + cleanupFailure = releasePreservation(cleanupFailure); + } + if (runtimeReleased && targetReleased && engineDataReleased && preservationReleased) { + engine.closed = true; } - engine.runtime = null; - cleanupFailure = runCleanup(cleanupFailure, engine.publishedTarget::close); - cleanupFailure = runCleanup(cleanupFailure, engine.engineDataStore::releaseEngineData); - engine.closed = true; - cleanupFailure = runCleanup(cleanupFailure, () -> { - PreservationRegistry registry = IrisServices.getOrNull(PreservationRegistry.class); - if (registry != null) { - registry.dereference(); - } - }); } if (cleanupFailure != null && cleanupFailure != original) { original.addSuppressed(cleanupFailure); @@ -223,41 +229,55 @@ final class EngineShutdownSequence { if (assembly == null) { return failure; } - failure = runCleanup(failure, () -> { - if (assembly.worldManager != null) { - assembly.worldManager.close(); - } - }); - failure = runCleanup(failure, () -> { - if (assembly.effects != null) { - assembly.effects.close(); - } - }); - failure = runCleanup(failure, () -> { - if (assembly.mode != null) { - assembly.mode.close(); - } - }); - Throwable complexFailure = runCleanup(null, () -> { - if (assembly.complex != null) { - assembly.complex.close(); - } - }); - failure = appendFailure(failure, complexFailure); - if (complexFailure != null) { - return failure; + incompleteAssemblies.add(assembly); + RuntimeAssembly previous = engine == null ? null : engine.runtimeAssembly.get(); + if (engine != null) { + engine.runtimeAssembly.set(assembly); } - failure = runCleanup(failure, () -> { - if (assembly.hash32 != null) { - assembly.hash32.cancel(true); + Throwable assemblyFailure; + try { + assemblyFailure = closeAssemblyResources(assembly); + } finally { + if (engine != null) { + if (previous == null) { + engine.runtimeAssembly.remove(); + } else { + engine.runtimeAssembly.set(previous); + } } - }); - if (assembly.mantle != null && assembly.ownsMantle) { - assembly.ownsMantle = false; - failure = runCleanup(failure, assembly.mantle::saveAllNow); - failure = runCleanup(failure, assembly.mantle::close); } - return failure; + if (assemblyFailure == null) { + incompleteAssemblies.remove(assembly); + } else if (engine != null) { + engine.lifecycleState = LifecycleState.FAILED; + } + return appendFailure(failure, assemblyFailure); + } + + void retainUnpublishedRuntime(EngineRuntime runtime) { + unpublishedRuntimes.add(runtime); + } + + boolean retainsData(IrisData data) { + if (engine.runtime != null && engine.runtime.generation().data() == data) { + return true; + } + for (RuntimeAssembly assembly : incompleteAssemblies) { + if (assembly.target.getData() == data) { + return true; + } + } + for (EngineRuntime runtime : unpublishedRuntimes) { + if (runtime.generation().data() == data) { + return true; + } + } + for (EngineTarget target : incompleteTargets.toArray(new EngineTarget[0])) { + if (target.getData() == data) { + return true; + } + } + return false; } Throwable closeRuntime(EngineRuntime engineRuntime, Throwable failure) { @@ -272,8 +292,11 @@ final class EngineShutdownSequence { if (engineRuntime == null) { return failure; } - failure = runCleanup(failure, engineRuntime.worldManager()::close); - failure = runCleanup(failure, engineRuntime.effects()::close); + Throwable serviceFailure = runCleanup(null, engineRuntime.worldManager()::close); + serviceFailure = runCleanup(serviceFailure, engineRuntime.effects()::close); + if (serviceFailure != null) { + return appendFailure(failure, serviceFailure); + } return closeGenerationRuntime(engineRuntime.generation(), retainedMantle, failure); } @@ -289,7 +312,10 @@ final class EngineShutdownSequence { if (generationRuntime == null) { return failure; } - failure = runCleanup(failure, generationRuntime.mode()::close); + Throwable modeFailure = runCleanup(null, generationRuntime.mode()::close); + if (modeFailure != null) { + return appendFailure(failure, modeFailure); + } Throwable complexFailure = runCleanup(null, generationRuntime.complex()::close); failure = appendFailure(failure, complexFailure); if (complexFailure != null) { @@ -297,8 +323,7 @@ final class EngineShutdownSequence { } failure = runCleanup(failure, () -> generationRuntime.hash32().cancel(true)); if (generationRuntime.mantle() != retainedMantle) { - failure = runCleanup(failure, generationRuntime.mantle()::saveAllNow); - failure = runCleanup(failure, generationRuntime.mantle()::close); + failure = appendFailure(failure, closeMantle(generationRuntime.mantle())); } return failure; } @@ -320,6 +345,18 @@ final class EngineShutdownSequence { } Throwable closeDetachedTarget(EngineTarget target, Throwable failure) { + for (RuntimeAssembly assembly : incompleteAssemblies) { + if (assembly.target == target) { + return appendFailure(failure, + new IllegalStateException("Iris runtime assembly still owns the detached target.")); + } + } + for (EngineRuntime runtime : unpublishedRuntimes) { + if (runtime.generation().target() == target) { + return appendFailure(failure, + new IllegalStateException("Iris unpublished runtime still owns the detached target.")); + } + } failure = runCleanup(failure, () -> target.getData().unregisterEngine(engine)); failure = runCleanup(failure, target::close); failure = runCleanup(failure, target.getData()::close); @@ -330,6 +367,10 @@ final class EngineShutdownSequence { if (runtimeReleased) { return failure; } + Throwable incompleteFailure = closeIncompleteRuntimes(); + if (incompleteFailure != null) { + return appendFailure(failure, incompleteFailure); + } Throwable detachedFailure = closeDetachedGenerationRuntimes(null); if (detachedFailure != null) { return appendFailure(failure, detachedFailure); @@ -343,6 +384,86 @@ final class EngineShutdownSequence { return failure; } + private Throwable closeAssemblyResources(RuntimeAssembly assembly) { + Throwable failure = runCleanup(null, () -> { + if (assembly.worldManager != null) { + assembly.worldManager.close(); + } + }); + failure = runCleanup(failure, () -> { + if (assembly.effects != null) { + assembly.effects.close(); + } + }); + if (failure != null) { + return failure; + } + failure = runCleanup(null, () -> { + if (assembly.mode != null) { + assembly.mode.close(); + } + }); + if (failure != null) { + return failure; + } + failure = runCleanup(null, () -> { + if (assembly.complex != null) { + assembly.complex.close(); + } + }); + if (failure != null) { + return failure; + } + failure = runCleanup(null, () -> { + if (assembly.hash32 != null) { + assembly.hash32.cancel(true); + } + }); + if (assembly.mantle != null && assembly.ownsMantle) { + Throwable mantleFailure = closeMantle(assembly.mantle); + if (mantleFailure == null) { + assembly.ownsMantle = false; + } + failure = appendFailure(failure, mantleFailure); + } + return failure; + } + + private Throwable closeIncompleteRuntimes() { + Throwable failure = null; + for (RuntimeAssembly assembly : incompleteAssemblies.toArray(new RuntimeAssembly[0])) { + Throwable assemblyFailure = closeAssembly(assembly, null); + if (assemblyFailure == null && assembly.target != engine.publishedTarget) { + incompleteTargets.add(assembly.target); + } + failure = appendFailure(failure, assemblyFailure); + } + EngineMantle retainedMantle = engine.runtime == null ? null : engine.runtime.generation().mantle(); + for (EngineRuntime unpublished : unpublishedRuntimes.toArray(new EngineRuntime[0])) { + Throwable runtimeFailure = closeRuntime(unpublished, retainedMantle, null); + if (runtimeFailure == null) { + unpublishedRuntimes.remove(unpublished); + if (unpublished.generation().target() != engine.publishedTarget) { + incompleteTargets.add(unpublished.generation().target()); + } + } + failure = appendFailure(failure, runtimeFailure); + } + for (EngineTarget target : incompleteTargets.toArray(new EngineTarget[0])) { + Throwable targetFailure = closeDetachedTarget(target, null); + if (targetFailure == null) { + incompleteTargets.remove(target); + } + failure = appendFailure(failure, targetFailure); + } + return failure; + } + + private static Throwable closeMantle(EngineMantle mantle) { + Throwable failure = runCleanup(null, mantle::saveAllNow); + return failure == null ? runCleanup(null, mantle::close) : failure; + } + private Throwable closeDetachedGenerationRuntimes(Throwable failure) { GenerationRuntime[] detachedRuntimes = engine.detachedGenerationRuntimes.toArray(new GenerationRuntime[0]); for (GenerationRuntime detached : detachedRuntimes) { diff --git a/core/src/main/java/art/arcane/iris/engine/mantle/MatterGenerator.java b/core/src/main/java/art/arcane/iris/engine/mantle/MatterGenerator.java index 0f53e9d6c..a5f2ec507 100644 --- a/core/src/main/java/art/arcane/iris/engine/mantle/MatterGenerator.java +++ b/core/src/main/java/art/arcane/iris/engine/mantle/MatterGenerator.java @@ -22,12 +22,13 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public interface MatterGenerator { MultiBurst DISPATCHER = MultiBurst.burst; - ConcurrentHashMap> IN_FLIGHT_COMPONENTS = new ConcurrentHashMap<>(); + ConcurrentHashMap IN_FLIGHT_COMPONENTS = new ConcurrentHashMap<>(); long COMPONENT_TASK_POLL_MS = 1000L; long COMPONENT_TASK_TIMEOUT_MS = Long.getLong("iris.mantle.componentTimeout", 120000L); @@ -368,38 +369,37 @@ public interface MatterGenerator { ) { MantleFlag flag = component.getFlag(); MatterTaskKey key = new MatterTaskKey(getMantle(), chunkX, chunkZ, flag); + MatterComponentTask task = new MatterComponentTask(key); if (chunk.isFlagged(flag)) { - return new MatterComponentTask(key, CompletableFuture.completedFuture(null), null); + task.finish(null); + return task; } - CompletableFuture future = new CompletableFuture<>(); - CompletableFuture existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, future); + MatterComponentTask existing = IN_FLIGHT_COMPONENTS.putIfAbsent(key, task); if (existing != null) { - return new MatterComponentTask(key, existing, null); + return existing; } try { if (DISPATCHER.ownsCurrentThread()) { // A pool thread runs its claim inline: a task queued behind pool threads that block // on unmanaged waits (structure builds, cache locks) could starve the whole pool. - completeComponentTask(future, key, chunk, component, writer, chunkX, chunkZ, context, IrisContext.get()); - return new MatterComponentTask(key, future, null); + completeComponentTask(task, chunk, component, writer, chunkX, chunkZ, context, IrisContext.get()); + return task; } IrisContext callerContext = IrisContext.get(); - Future submission = DISPATCHER.submit(() -> completeComponentTask( - future, - key, + task.setSubmission(DISPATCHER.submit(() -> completeComponentTask( + task, chunk, component, writer, chunkX, chunkZ, context, - callerContext)); - return new MatterComponentTask(key, future, submission); + callerContext))); + return task; } catch (Throwable throwable) { - IN_FLIGHT_COMPONENTS.remove(key, future); - future.completeExceptionally(throwable); + task.cancel(throwable); throw throwable; } } @@ -449,7 +449,7 @@ public interface MatterGenerator { } private static void awaitComponentTask(MatterComponentTask task) { - CompletableFuture future = task.future(); + CompletableFuture future = task.future; if (future.isDone() && !future.isCompletedExceptionally()) { return; } @@ -459,11 +459,17 @@ public interface MatterGenerator { try { ForkJoinPool.managedBlock(new ComponentTaskBlocker(task)); } catch (InterruptedException interruption) { - Thread.currentThread().interrupt(); + try { + task.await(COMPONENT_TASK_TIMEOUT_MS); + } finally { + Thread.currentThread().interrupt(); + } + } catch (RejectedExecutionException exhaustedPool) { + task.await(COMPONENT_TASK_TIMEOUT_MS); } return; } - awaitComponentTaskBlocking(task); + task.await(COMPONENT_TASK_TIMEOUT_MS); } /** @@ -478,70 +484,22 @@ public interface MatterGenerator { @Override public boolean block() { - awaitComponentTaskBlocking(task); + task.await(COMPONENT_TASK_TIMEOUT_MS); return true; } @Override public boolean isReleasable() { - if (!task.future().isDone()) { + if (!task.future.isDone()) { return false; } - task.future().join(); + task.future.join(); return true; } } - private static void awaitComponentTaskBlocking(MatterComponentTask task) { - CompletableFuture future = task.future(); - long start = System.currentTimeMillis(); - boolean interrupted = false; - - try { - while (true) { - try { - future.get(COMPONENT_TASK_POLL_MS, TimeUnit.MILLISECONDS); - return; - } catch (InterruptedException interruption) { - // The writer cannot close while this task still writes through it, so the wait is - // uninterruptible and the flag is restored on the way out. - interrupted = true; - } catch (TimeoutException timeout) { - Future submission = task.submission(); - boolean dropped = submission != null && submission.isDone(); - if (future.isDone()) { - continue; - } - - long waited = System.currentTimeMillis() - start; - if (!dropped && waited < COMPONENT_TASK_TIMEOUT_MS) { - continue; - } - - // The task can no longer complete (the dispatcher dropped it) or is wedged. Drop - // the dedup entry so later generations do not inherit a future that never - // completes, and release anything else already waiting on it. - IllegalStateException failure = new IllegalStateException("Mantle component " + task.key() - + (dropped ? " was dropped by the dispatcher" : " did not complete in " + waited + "ms")); - IN_FLIGHT_COMPONENTS.remove(task.key(), future); - future.completeExceptionally(failure); - IrisLogging.error(failure.getMessage()); - throw new CompletionException(failure); - } catch (ExecutionException executionFailure) { - Throwable cause = executionFailure.getCause(); - throw new CompletionException(cause != null ? cause : executionFailure); - } - } - } finally { - if (interrupted) { - Thread.currentThread().interrupt(); - } - } - } - private void completeComponentTask( - CompletableFuture future, - MatterTaskKey key, + MatterComponentTask task, MantleChunk chunk, MantleComponent component, MantleWriter writer, @@ -550,14 +508,17 @@ public interface MatterGenerator { ChunkContext context, IrisContext callerContext ) { + if (!task.start()) { + return; + } + Throwable failure = null; try { runComponentWithContext(chunk, component, writer, chunkX, chunkZ, context, callerContext); - future.complete(null); } catch (Throwable throwable) { - future.completeExceptionally(throwable); + failure = throwable; throw throwable; } finally { - IN_FLIGHT_COMPONENTS.remove(key, future); + task.finish(failure); } } @@ -597,11 +558,98 @@ public interface MatterGenerator { )); } - /** - * A launched component task. {@code submission} is the dispatcher handle when this generation - * owns the task, and null when the future belongs to another generation or already completed. - */ - record MatterComponentTask(MatterTaskKey key, CompletableFuture future, Future submission) { + final class MatterComponentTask { + private final MatterTaskKey key; + private final CompletableFuture future = new CompletableFuture<>(); + private volatile Future submission; + private volatile Throwable failure; + private boolean started; + + MatterComponentTask(MatterTaskKey key) { + this.key = key; + } + + synchronized boolean start() { + if (started || future.isDone()) { + return false; + } + started = true; + return true; + } + + synchronized void setSubmission(Future submission) { + this.submission = submission; + if (failure != null) { + submission.cancel(true); + } + } + + synchronized boolean cancel(Throwable cause) { + if (future.isDone() || failure != null) { + return false; + } + failure = cause; + if (!started) { + finish(null); + } + if (submission != null) { + submission.cancel(true); + } + return true; + } + + synchronized void finish(Throwable cause) { + if (failure == null) { + failure = cause; + } else if (cause != null && cause != failure) { + failure.addSuppressed(cause); + } + IN_FLIGHT_COMPONENTS.remove(key, this); + if (failure == null) { + future.complete(null); + } else { + future.completeExceptionally(failure); + } + } + + void await(long timeoutMillis) { + long start = System.nanoTime(); + boolean interrupted = false; + try { + while (true) { + try { + future.get(COMPONENT_TASK_POLL_MS, TimeUnit.MILLISECONDS); + return; + } catch (InterruptedException interruption) { + // The writer cannot close while this task still writes through it, so the wait is + // uninterruptible and the flag is restored on the way out. + interrupted = true; + } catch (TimeoutException timeout) { + if (future.isDone() || failure != null) { + continue; + } + Future submitted = submission; + boolean dropped = submitted != null && submitted.isDone(); + long waited = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + if (!dropped && waited < timeoutMillis) { + continue; + } + IllegalStateException timeoutFailure = new IllegalStateException("Mantle component " + key + + (dropped ? " was dropped by the dispatcher" : " did not complete in " + waited + "ms")); + if (cancel(timeoutFailure)) { + IrisLogging.error(timeoutFailure.getMessage()); + } + } catch (ExecutionException executionFailure) { + Throwable cause = executionFailure.getCause(); + throw new CompletionException(cause != null ? cause : executionFailure); + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } } record MatterPassPlan(MantlePass pass, int passChunkRadius, int downstreamBlockRadius) { diff --git a/core/src/test/java/art/arcane/iris/engine/EngineRuntimePublicationContractTest.java b/core/src/test/java/art/arcane/iris/engine/EngineRuntimePublicationContractTest.java index 5a640bb7d..8015b323a 100644 --- a/core/src/test/java/art/arcane/iris/engine/EngineRuntimePublicationContractTest.java +++ b/core/src/test/java/art/arcane/iris/engine/EngineRuntimePublicationContractTest.java @@ -33,9 +33,9 @@ public class EngineRuntimePublicationContractTest { int publishEnd = source.indexOf("private void scheduleRuntimeTasks", publishStart); String publish = source.substring(publishStart, publishEnd); - assertBefore(publish, "engine.getClosing().set(true)", "closeRuntime(next, e)"); - assertBefore(publish, "closeBackgroundTaskAdmission()", "closeRuntime(next, e)"); - assertBefore(publish, "sealAndAwait(", "closeRuntime(next, e)"); + assertBefore(publish, "engine.getClosing().set(true)", "cleanupFailure = engine.shutdownSequence.closeRuntime(next, null)"); + assertBefore(publish, "closeBackgroundTaskAdmission()", "cleanupFailure = engine.shutdownSequence.closeRuntime(next, null)"); + assertBefore(publish, "sealAndAwait(", "cleanupFailure = engine.shutdownSequence.closeRuntime(next, null)"); } @Test diff --git a/core/src/test/java/art/arcane/iris/engine/EngineShutdownDrainTest.java b/core/src/test/java/art/arcane/iris/engine/EngineShutdownDrainTest.java index 6f17ceb7b..fe8e6f2ed 100644 --- a/core/src/test/java/art/arcane/iris/engine/EngineShutdownDrainTest.java +++ b/core/src/test/java/art/arcane/iris/engine/EngineShutdownDrainTest.java @@ -1,6 +1,15 @@ package art.arcane.iris.engine; +import art.arcane.iris.core.loader.IrisData; import art.arcane.iris.engine.EngineRuntimeBuilder.RuntimeAssembly; +import art.arcane.iris.engine.framework.EngineEffects; +import art.arcane.iris.engine.framework.EnginePlatformHooks; +import art.arcane.iris.engine.framework.EngineTarget; +import art.arcane.iris.engine.framework.EngineWorldManager; +import art.arcane.iris.engine.framework.GenerationSessionException; +import art.arcane.iris.engine.framework.GenerationSessionManager; +import art.arcane.iris.engine.framework.NativeStructureOwnershipStore; +import art.arcane.iris.engine.object.IrisWorld; import art.arcane.iris.engine.framework.EngineMode; import art.arcane.iris.engine.mantle.EngineMantle; import art.arcane.iris.spi.IrisPlatform; @@ -11,17 +20,28 @@ import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; import org.mockito.InOrder; +import org.mockito.MockedStatic; +import java.lang.reflect.Field; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertThrows; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; @@ -99,4 +119,238 @@ public class EngineShutdownDrainTest { assertTrue(assembly.ownsMantle); assertFalse(assembly.hash32.isCancelled()); } + @Test + public void failedAssemblySaveKeepsOwnershipAndDoesNotCloseStorage() { + RuntimeAssembly assembly = mock(RuntimeAssembly.class); + assembly.mantle = mock(EngineMantle.class); + assembly.ownsMantle = true; + IllegalStateException failure = new IllegalStateException("Save failed"); + doThrow(failure).doNothing().when(assembly.mantle).saveAllNow(); + EngineShutdownSequence shutdown = new EngineShutdownSequence(null); + + assertSame(failure, shutdown.closeAssembly(assembly, null)); + assertTrue(assembly.ownsMantle); + verify(assembly.mantle, never()).close(); + assertNull(shutdown.closeAssembly(assembly, null)); + assertFalse(assembly.ownsMantle); + assertNull(shutdown.closeAssembly(assembly, null)); + verify(assembly.mantle, times(2)).saveAllNow(); + verify(assembly.mantle).close(); + } + + @Test + public void failedAssemblyCloseKeepsOwnershipUntilRetrySucceeds() { + RuntimeAssembly assembly = mock(RuntimeAssembly.class); + assembly.mantle = mock(EngineMantle.class); + assembly.ownsMantle = true; + IllegalStateException failure = new IllegalStateException("Close failed"); + doThrow(failure).doNothing().when(assembly.mantle).close(); + EngineShutdownSequence shutdown = new EngineShutdownSequence(null); + + assertSame(failure, shutdown.closeAssembly(assembly, null)); + assertTrue(assembly.ownsMantle); + assertNull(shutdown.closeAssembly(assembly, null)); + assertFalse(assembly.ownsMantle); + verify(assembly.mantle, times(2)).saveAllNow(); + verify(assembly.mantle, times(2)).close(); + } + + @Test + public void failedPartialAssemblyRetainsDataAndRetriesInsideItsAssemblyScope() throws Exception { + ShutdownFixture fixture = new ShutdownFixture(); + RuntimeAssembly assembly = mock(RuntimeAssembly.class); + EngineTarget detachedTarget = target(); + setField(assembly, "target", detachedTarget); + assembly.complex = mock(IrisComplex.class); + assembly.mantle = mock(EngineMantle.class); + assembly.ownsMantle = true; + RuntimeAssembly previous = mock(RuntimeAssembly.class); + fixture.engine.runtimeAssembly.set(previous); + IllegalStateException failure = new IllegalStateException("Planner still active"); + doAnswer(invocation -> { + assertSame(assembly, fixture.engine.runtimeAssembly.get()); + throw failure; + }).when(assembly.complex).close(); + + assertSame(failure, fixture.shutdown.closeAssembly(assembly, null)); + assertSame(previous, fixture.engine.runtimeAssembly.get()); + fixture.engine.runtimeAssembly.remove(); + assertTrue(fixture.shutdown.retainsData(detachedTarget.getData())); + assertTrue(fixture.shutdown.closeDetachedTarget(detachedTarget, null) instanceof IllegalStateException); + verify(detachedTarget, never()).close(); + try (MockedStatic ignored = mockStatic(NativeStructureOwnershipStore.class)) { + assertThrows(IllegalStateException.class, fixture.shutdown::close); + verify(fixture.target, never()).close(); + verifyNoInteractions(assembly.mantle); + doAnswer(invocation -> { + assertSame(assembly, fixture.engine.runtimeAssembly.get()); + return null; + }).when(assembly.complex).close(); + fixture.shutdown.close(); + } + + assertNull(fixture.engine.runtimeAssembly.get()); + assertFalse(assembly.ownsMantle); + assertFalse(fixture.shutdown.retainsData(detachedTarget.getData())); + verify(assembly.mantle).close(); + verify(detachedTarget).close(); + verify(detachedTarget.getData()).close(); + assertTrue(fixture.engine.closed); + assertNull(fixture.engine.runtime); + } + + @Test + public void failedManagerStartKeepsPublishedRuntimeWhenCleanupFails() throws Exception { + ShutdownFixture fixture = new ShutdownFixture(); + fixture.engine.runtime = null; + IllegalStateException startFailure = new IllegalStateException("Manager start failed"); + IllegalStateException closeFailure = new IllegalStateException("Planner still active"); + doThrow(startFailure).when(fixture.manager).start(); + doThrow(closeFailure).when(fixture.complex).close(); + + IllegalStateException failure = assertThrows(IllegalStateException.class, + () -> new EngineRuntimeBuilder(fixture.engine).publishRuntime(fixture.runtime, null)); + + assertSame(startFailure, failure.getCause()); + assertSame(closeFailure, startFailure.getSuppressed()[0]); + assertSame(fixture.runtime, fixture.engine.runtime); + assertTrue(fixture.shutdown.retainsData(fixture.target.getData())); + verify(fixture.mantle, never()).close(); + doNothing().when(fixture.complex).close(); + try (MockedStatic ignored = mockStatic(NativeStructureOwnershipStore.class)) { + fixture.shutdown.close(); + } + assertNull(fixture.engine.runtime); + assertTrue(fixture.engine.closed); + verify(fixture.target).close(); + } + + @Test + public void failedStartDrainRetainsResourcesUntilGenerationStops() throws Exception { + ShutdownFixture fixture = new ShutdownFixture(); + fixture.engine.runtime = null; + doThrow(new IllegalStateException("Manager start failed")).when(fixture.manager).start(); + doThrow(new GenerationSessionException("Generation still active", false)).when(fixture.sessions) + .sealAndAwait("failed world manager start", IrisEngine.SESSION_DRAIN_TIMEOUT_MILLIS, true); + + assertThrows(IllegalStateException.class, + () -> new EngineRuntimeBuilder(fixture.engine).publishRuntime(fixture.runtime, null)); + + assertSame(fixture.runtime, fixture.engine.runtime); + verify(fixture.manager, never()).close(); + verifyNoInteractions(fixture.complex, fixture.mantle); + try (MockedStatic ignored = mockStatic(NativeStructureOwnershipStore.class)) { + fixture.shutdown.close(); + } + assertNull(fixture.engine.runtime); + assertTrue(fixture.engine.closed); + } + + @Test + public void failedRetirementRetainsUnpublishedRuntimeWithoutClosingSharedMantle() throws Exception { + ShutdownFixture fixture = new ShutdownFixture(); + EngineTarget nextTarget = target(); + GenerationRuntime nextGeneration = generation(nextTarget, fixture.mantle); + EngineRuntime next = new EngineRuntime(nextGeneration, mock(EngineEffects.class), mock(EngineWorldManager.class)); + EngineMode nextMode = nextGeneration.mode(); + doThrow(new IllegalStateException("Previous planner active")).when(fixture.complex).close(); + doThrow(new IllegalStateException("Next mode active")).when(nextMode).close(); + + assertThrows(IllegalStateException.class, + () -> new EngineRuntimeBuilder(fixture.engine).publishRuntime(next, fixture.runtime)); + + assertSame(fixture.runtime, fixture.engine.runtime); + assertTrue(fixture.shutdown.retainsData(nextTarget.getData())); + verifyNoInteractions(fixture.mantle); + doNothing().when(fixture.complex).close(); + doNothing().when(nextMode).close(); + try (MockedStatic ignored = mockStatic(NativeStructureOwnershipStore.class)) { + fixture.shutdown.close(); + } + assertFalse(fixture.shutdown.retainsData(nextTarget.getData())); + verify(nextTarget).close(); + verify(fixture.target).close(); + verify(fixture.mantle).saveAllNow(); + verify(fixture.mantle).close(); + assertTrue(fixture.engine.closed); + } + + @Test + public void failedConstructionDrainPreventsAllRuntimeAndTargetRelease() throws Exception { + ShutdownFixture fixture = new ShutdownFixture(); + doThrow(new GenerationSessionException("Generation still active", false)).doNothing().when(fixture.sessions) + .sealAndAwait("failed initialization", 0L, true); + IllegalStateException original = new IllegalStateException("Construction failed"); + try (MockedStatic ignored = mockStatic(NativeStructureOwnershipStore.class)) { + fixture.shutdown.cleanupFailedConstruction(original); + assertSame(fixture.runtime, fixture.engine.runtime); + assertFalse(fixture.engine.closed); + verifyNoInteractions(fixture.complex, fixture.mantle); + verify(fixture.target, never()).close(); + fixture.shutdown.cleanupFailedConstruction(original); + } + assertNull(fixture.engine.runtime); + assertTrue(fixture.engine.closed); + verify(fixture.mantle).close(); + verify(fixture.target).close(); + } + + private static EngineTarget target() { + EngineTarget target = mock(EngineTarget.class); + when(target.getData()).thenReturn(mock(IrisData.class)); + when(target.getWorld()).thenReturn(mock(IrisWorld.class)); + return target; + } + + private static GenerationRuntime generation(EngineTarget target, EngineMantle mantle) { + GenerationRuntime generation = mock(GenerationRuntime.class); + IrisData data = target.getData(); + when(generation.target()).thenReturn(target); + when(generation.data()).thenReturn(data); + when(generation.mantle()).thenReturn(mantle); + when(generation.complex()).thenReturn(mock(IrisComplex.class)); + when(generation.mode()).thenReturn(mock(EngineMode.class)); + when(generation.hash32()).thenReturn(new CompletableFuture<>()); + return generation; + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + + private static final class ShutdownFixture { + private final IrisEngine engine = mock(IrisEngine.class); + private final EngineTarget target = target(); + private final EngineMantle mantle = mock(EngineMantle.class); + private final GenerationRuntime generation = generation(target, mantle); + private final IrisComplex complex = generation.complex(); + private final EngineWorldManager manager = mock(EngineWorldManager.class); + private final EngineRuntime runtime = new EngineRuntime(generation, mock(EngineEffects.class), manager); + private final GenerationSessionManager sessions = mock(GenerationSessionManager.class); + private final EngineShutdownSequence shutdown = new EngineShutdownSequence(engine); + + private ShutdownFixture() throws Exception { + EngineBackgroundTasks background = mock(EngineBackgroundTasks.class); + when(background.drainBackgroundTasks(anyString())) + .thenReturn(new EngineBackgroundTasks.BackgroundTaskDrain(null, true)); + setField(engine, "backgroundTasks", background); + setField(engine, "shutdownSequence", shutdown); + setField(engine, "engineDataStore", mock(EngineDataStore.class)); + setField(engine, "lifecycleLock", new Object()); + setField(engine, "runtimeAssembly", new ThreadLocal()); + setField(engine, "detachedGenerationRuntimes", ConcurrentHashMap.newKeySet()); + when(engine.getClosing()).thenReturn(new AtomicBoolean()); + when(engine.getGenerationSessions()).thenReturn(sessions); + when(engine.getPlatformHooks()).thenReturn(mock(EnginePlatformHooks.class)); + when(engine.getMantle()).thenReturn(mantle); + IrisWorld world = target.getWorld(); + when(engine.getWorld()).thenReturn(world); + when(engine.beginShutdown()).thenReturn(true); + engine.runtime = runtime; + engine.publishedTarget = target; + } + } + } diff --git a/core/src/test/java/art/arcane/iris/engine/EngineShutdownOwnershipContractTest.java b/core/src/test/java/art/arcane/iris/engine/EngineShutdownOwnershipContractTest.java index af03d028d..bdadc9a7c 100644 --- a/core/src/test/java/art/arcane/iris/engine/EngineShutdownOwnershipContractTest.java +++ b/core/src/test/java/art/arcane/iris/engine/EngineShutdownOwnershipContractTest.java @@ -43,14 +43,14 @@ public class EngineShutdownOwnershipContractTest { String cleanup = source.substring(cleanupStart, cleanupEnd); int ownershipClose = cleanup.indexOf("NativeStructureOwnershipStore.close(engine)"); int ownershipGate = cleanup.indexOf("if (ownershipFailure == null)", ownershipClose); - int detachedRelease = cleanup.indexOf("closeDetachedGenerationRuntimes(null)", ownershipGate); - int runtimeRelease = cleanup.indexOf("closeRuntime(engine.runtime, null)", ownershipGate); + int runtimeRelease = cleanup.indexOf("releaseRuntime(cleanupFailure)", ownershipGate); + int targetRelease = cleanup.indexOf("releaseTarget(cleanupFailure)", runtimeRelease); int closedPublication = cleanup.indexOf("engine.closed = true", ownershipGate); assertTrue(ownershipClose >= 0); assertTrue(ownershipGate > ownershipClose); - assertTrue(detachedRelease > ownershipGate); - assertTrue(runtimeRelease > detachedRelease); + assertTrue(runtimeRelease > ownershipGate); + assertTrue(targetRelease > runtimeRelease); assertTrue(closedPublication > ownershipGate); } diff --git a/core/src/test/java/art/arcane/iris/engine/mantle/MatterGeneratorConcurrencyTest.java b/core/src/test/java/art/arcane/iris/engine/mantle/MatterGeneratorConcurrencyTest.java index 86b85cf48..16df4d316 100644 --- a/core/src/test/java/art/arcane/iris/engine/mantle/MatterGeneratorConcurrencyTest.java +++ b/core/src/test/java/art/arcane/iris/engine/mantle/MatterGeneratorConcurrencyTest.java @@ -27,6 +27,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.Future; +import java.util.concurrent.CompletionException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -44,6 +46,9 @@ import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class MatterGeneratorConcurrencyTest { @@ -221,23 +226,153 @@ public class MatterGeneratorConcurrencyTest { @Test public void dispatcherPropagatesAnAlreadyFailedSharedComponent() { IllegalStateException failure = new IllegalStateException("Shared component failed"); - ExecutionException observed = sharedComponentFailure(CompletableFuture.failedFuture(failure), ExecutionException.class); + ExecutionException observed = sharedComponentFailure(failure, ExecutionException.class); assertSame(failure, observed.getCause()); } @Test public void dispatcherPropagatesAnAlreadyCancelledSharedComponent() { - CompletableFuture cancelled = new CompletableFuture<>(); - cancelled.cancel(false); - sharedComponentFailure(cancelled, CancellationException.class); + sharedComponentFailure(new CancellationException("Shared component cancelled"), CancellationException.class); } - private T sharedComponentFailure(CompletableFuture completion, Class expectedFailure) { + @Test + public void timedOutRunningComponentRetainsWriterAndSharedClaimUntilItExits() throws Exception { + GeneratorFixture fixture = new GeneratorFixture(true); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch writersAcquired = new CountDownLatch(2); + CountDownLatch cancelled = new CountDownLatch(1); + AtomicInteger runs = new AtomicInteger(); + MantleChunk chunk = fixture.mantle.getChunk(0, 0); + when(chunk.use()).thenAnswer(invocation -> { + writersAcquired.countDown(); + return chunk; + }); + RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 0) { + @Override + public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) { + runs.incrementAndGet(); + started.countDown(); + await(release); + verify(chunk, never()).release(); + } + }; + TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 0, 0))); + CompletableFuture first = CompletableFuture.runAsync( + () -> generator.generateMatter(0, 0, true, fixture.context)); + await(started); + MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT); + MatterGenerator.MatterComponentTask task = MatterGenerator.IN_FLIGHT_COMPONENTS.get(key); + Future submission = mock(Future.class); + when(submission.cancel(true)).thenAnswer(invocation -> { + cancelled.countDown(); + return true; + }); + task.setSubmission(submission); + CompletableFuture second = MultiBurst.burst.completeValueAsync(() -> { + generator.generateMatter(0, 0, true, fixture.context); + return null; + }); + CompletableFuture timeout = CompletableFuture.runAsync(() -> task.await(0L)); + try { + await(writersAcquired); + await(cancelled); + assertSame(task, MatterGenerator.IN_FLIGHT_COMPONENTS.get(key)); + assertFalse(first.isDone()); + assertFalse(second.isDone()); + assertFalse(timeout.isDone()); + verify(chunk, never()).release(); + assertEquals(1, runs.get()); + } finally { + release.countDown(); + } + assertThrows(ExecutionException.class, () -> timeout.get(5L, TimeUnit.SECONDS)); + assertThrows(ExecutionException.class, () -> first.get(5L, TimeUnit.SECONDS)); + assertThrows(ExecutionException.class, () -> second.get(5L, TimeUnit.SECONDS)); + verify(chunk, times(2)).release(); + assertFalse(MatterGenerator.IN_FLIGHT_COMPONENTS.containsKey(key)); + } + + @Test + public void queuedTimeoutPreventsLateExecutionAndAllowsANewClaim() { + GeneratorFixture fixture = new GeneratorFixture(); + MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT); + MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key); + MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, task); + Future submission = mock(Future.class); + task.setSubmission(submission); + + CompletionException failure = assertThrows(CompletionException.class, () -> task.await(0L)); + + assertTrue(failure.getCause().getMessage().contains("did not complete")); + assertFalse(task.start()); + assertFalse(MatterGenerator.IN_FLIGHT_COMPONENTS.containsKey(key)); + verify(submission).cancel(true); + MatterGenerator.MatterComponentTask replacement = new MatterGenerator.MatterComponentTask(key); + assertNull(MatterGenerator.IN_FLIGHT_COMPONENTS.putIfAbsent(key, replacement)); + task.finish(null); + assertSame(replacement, MatterGenerator.IN_FLIGHT_COMPONENTS.get(key)); + replacement.finish(null); + } + + @Test + public void droppedQueuedSubmissionCannotRunAfterTheBarrier() { + GeneratorFixture fixture = new GeneratorFixture(); + MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT); + MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key); + MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, task); + Future submission = mock(Future.class); + when(submission.isDone()).thenReturn(true); + task.setSubmission(submission); + + CompletionException failure = assertThrows(CompletionException.class, + () -> task.await(MatterGenerator.COMPONENT_TASK_TIMEOUT_MS)); + + assertTrue(failure.getCause().getMessage().contains("was dropped")); + assertFalse(task.start()); + assertFalse(MatterGenerator.IN_FLIGHT_COMPONENTS.containsKey(key)); + } + + @Test + public void interruptedWaiterDrainsRunningTaskAndRestoresInterrupt() throws Exception { + GeneratorFixture fixture = new GeneratorFixture(); + MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT); + MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key); + assertTrue(task.start()); + CountDownLatch waiting = new CountDownLatch(1); + AtomicBoolean restored = new AtomicBoolean(); + CompletableFuture completion = new CompletableFuture<>(); + Thread waiter = new Thread(() -> { + try { + Thread.currentThread().interrupt(); + waiting.countDown(); + task.await(MatterGenerator.COMPONENT_TASK_TIMEOUT_MS); + restored.set(Thread.currentThread().isInterrupted()); + completion.complete(null); + } catch (Throwable failure) { + completion.completeExceptionally(failure); + } + }); + waiter.start(); + await(waiting); + try { + assertFalse(completion.isDone()); + } finally { + task.finish(null); + } + completion.get(5L, TimeUnit.SECONDS); + waiter.join(5_000L); + assertTrue(restored.get()); + } + + private T sharedComponentFailure(Throwable failure, Class expectedFailure) { GeneratorFixture fixture = new GeneratorFixture(true); RecordingComponent component = new RecordingComponent(ReservedFlag.OBJECT, 0, 0); TestMatterGenerator generator = fixture.generator(List.of(new MantlePass(List.of(component), 0, 0))); MatterGenerator.MatterTaskKey key = new MatterGenerator.MatterTaskKey(fixture.mantle, 0, 0, ReservedFlag.OBJECT); - MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, completion); + MatterGenerator.MatterComponentTask task = new MatterGenerator.MatterComponentTask(key); + task.finish(failure); + MatterGenerator.IN_FLIGHT_COMPONENTS.put(key, task); try { CompletableFuture generation = MultiBurst.burst.completeValueAsync(() -> { generator.generateMatter(0, 0, true, fixture.context); @@ -245,7 +380,7 @@ public class MatterGeneratorConcurrencyTest { }); return assertThrows(expectedFailure, () -> generation.get(5L, TimeUnit.SECONDS)); } finally { - MatterGenerator.IN_FLIGHT_COMPONENTS.remove(key, completion); + MatterGenerator.IN_FLIGHT_COMPONENTS.remove(key, task); } }