From 41eff8c878a653a3e616c7412d846035c22830c4 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Thu, 12 Nov 2026 13:07:33 -0800 Subject: [PATCH 1/3] Leave DBOS unlaunched when launch() fails launch() published its executor before running migrations and starting it. When either threw, the instance kept that executor: a second launch() returned at once with nothing running, and other calls failed with a NullPointerException on what it never started. Every deployment that runs with migrations off against a schema below the required version takes that path. Now a launch that throws shuts down whatever it started, clears the executor and rethrows, so the instance is as if launch() had never been called and a later launch() tries again from the start. Closing an executor that failed before creating its thread pools no longer throws. --- .../src/main/java/dev/dbos/transact/DBOS.java | 54 ++++++--- .../dbos/transact/execution/DBOSExecutor.java | 11 +- .../dev/dbos/transact/FailedLaunchTest.java | 106 ++++++++++++++++++ 3 files changed, 150 insertions(+), 21 deletions(-) create mode 100644 transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java diff --git a/transact/src/main/java/dev/dbos/transact/DBOS.java b/transact/src/main/java/dev/dbos/transact/DBOS.java index 4af7494a..fe4e9f1e 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOS.java +++ b/transact/src/main/java/dev/dbos/transact/DBOS.java @@ -380,7 +380,13 @@ private void registerStepCheckpointStore(StepCheckpointStore store) { stepCheckpointStores.add(store); } - /** Launch DBOS, and start recovery. */ + /** + * Launch DBOS, and start recovery. + * + *

If launch throws, for example because migrations are off and the system database is missing + * a migration, whatever it started is shut down and this instance is left as if launch had never + * been called, so a later {@code launch()} tries again from the start. + */ public void launch() { logger.info("Launching DBOS v{}", DBOS.version()); @@ -388,24 +394,36 @@ public void launch() { var executor = new DBOSExecutor(config); if (dbosExecutor.compareAndSet(null, executor)) { - if (config.migrate()) { - MigrationManager.runMigrations(config); - } else { - // The deployment owns the schema, but it must have every migration this SDK defines. - MigrationManager.validateSysDbVersion(config); + try { + if (config.migrate()) { + MigrationManager.runMigrations(config); + } else { + // The deployment owns the schema, but it must have every migration this SDK defines. + MigrationManager.validateSysDbVersion(config); + } + + executor.start( + this, + new HashSet<>(this.lifecycleRegistry), + workflowRegistry.getWorkflowSnapshot(), + workflowRegistry.getInstanceSnapshot(), + workflowRegistry.getInternalWorkflowSnapshot(), + workflowRegistry.getInternalInstanceSnapshot(), + queueRegistry.getSnapshot(), + Set.copyOf(stepCheckpointStores), + keepCompletedStepCheckpoints, + alertHandler); + } catch (RuntimeException | Error e) { + // Left in place, the executor would make every later launch() return at once with + // nothing running, and every other call fail on what it never started. + try { + executor.close(); + } catch (RuntimeException closeFailure) { + e.addSuppressed(closeFailure); + } + dbosExecutor.compareAndSet(executor, null); + throw e; } - - executor.start( - this, - new HashSet<>(this.lifecycleRegistry), - workflowRegistry.getWorkflowSnapshot(), - workflowRegistry.getInstanceSnapshot(), - workflowRegistry.getInternalWorkflowSnapshot(), - workflowRegistry.getInternalInstanceSnapshot(), - queueRegistry.getSnapshot(), - Set.copyOf(stepCheckpointStores), - keepCompletedStepCheckpoints, - alertHandler); } } } diff --git a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java index 3ddcb55e..fc1a40cb 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -484,9 +484,14 @@ public void close() { } systemDatabase = null; - timeoutScheduler.shutdownNow(); - var notRun = executorService.shutdownNow(); - logger.debug("Shutting down DBOS Executor threadpool. Tasks not run {}", notRun.size()); + // Both are null when start() failed before creating them. + if (timeoutScheduler != null) { + timeoutScheduler.shutdownNow(); + } + if (executorService != null) { + var notRun = executorService.shutdownNow(); + logger.debug("Shutting down DBOS Executor threadpool. Tasks not run {}", notRun.size()); + } this.workflowMap = null; this.instanceMap = null; diff --git a/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java b/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java new file mode 100644 index 00000000..b188b8cf --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java @@ -0,0 +1,106 @@ +package dev.dbos.transact; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.dbos.transact.config.DBOSConfig; +import dev.dbos.transact.execution.DBOSLifecycleListener; +import dev.dbos.transact.migrations.MigrationManager; +import dev.dbos.transact.utils.PgContainer; +import dev.dbos.transact.workflow.Workflow; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.Test; + +/** + * A launch that throws leaves the instance as if it had never been launched: a later launch tries + * again from the start, and other calls fail as they do before any launch. + */ +public class FailedLaunchTest { + + @AutoClose final PgContainer pgContainer = new PgContainer(); + + @AutoClose DBOS dbos; + + private FailedLaunchService proxy; + + private void create(DBOSConfig config) { + dbos = new DBOS(config); + proxy = dbos.registerProxy(FailedLaunchService.class, new FailedLaunchServiceImpl()); + } + + private void assertNotLaunched() { + var e = assertThrows(IllegalStateException.class, () -> dbos.getWorkflowStatus("any")); + assertTrue(e.getMessage().contains("before DBOS is launched"), e.getMessage()); + } + + @Test + public void aLaunchRefusedForAnUnmigratedSchemaCanBeRetriedOnceItIsMigrated() throws Exception { + // A schema no migration has run on, which a launch with migrations off must refuse. + var schema = "failed_launch_%d".formatted(System.nanoTime()); + var config = pgContainer.dbosConfig().withDatabaseSchema(schema).withMigrate(false); + create(config); + + var first = assertThrows(IllegalStateException.class, dbos::launch); + assertNotLaunched(); + // Tried again for real, not skipped because an executor is already in place. + var second = assertThrows(IllegalStateException.class, dbos::launch); + assertEquals(first.getMessage(), second.getMessage()); + + MigrationManager.runMigrations(config.withMigrate(true)); + dbos.launch(); + var handle = + dbos.startWorkflow(() -> proxy.echo("after"), new StartWorkflowOptions("echo-after")); + assertEquals("after", handle.getResult()); + } + + @Test + public void aLaunchThatFailsPartWayShutsDownWhatItStartedAndCanBeRetried() throws Exception { + create(pgContainer.dbosConfig()); + var listener = new FailingOnceListener(); + dbos.integration().registerLifecycleListener(listener); + + var e = assertThrows(IllegalStateException.class, dbos::launch); + assertEquals("listener failed", e.getMessage()); + assertEquals(1, listener.shutDowns.get(), "what the failed launch started was shut down"); + assertNotLaunched(); + + dbos.launch(); + assertEquals(2, listener.launches.get()); + var handle = dbos.startWorkflow(() -> proxy.echo("again"), new StartWorkflowOptions("again")); + assertEquals("again", handle.getResult()); + } + + /** Fails the first launch, after the executor has started its services, and not the next. */ + private static final class FailingOnceListener implements DBOSLifecycleListener { + final AtomicInteger launches = new AtomicInteger(); + final AtomicInteger shutDowns = new AtomicInteger(); + + @Override + public void dbosLaunched(DBOS dbos) { + if (launches.incrementAndGet() == 1) { + throw new IllegalStateException("listener failed"); + } + } + + @Override + public void dbosShutDown() { + shutDowns.incrementAndGet(); + } + } +} + +interface FailedLaunchService { + String echo(String value); +} + +class FailedLaunchServiceImpl implements FailedLaunchService { + @Override + @Workflow + public String echo(String value) { + return value; + } +} From df8e6f9af07e6597fb047d64bd4f942144de96f4 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Thu, 12 Nov 2026 13:12:55 -0800 Subject: [PATCH 2/3] Reset a failed launch whatever it throws, and say when listeners hear of it Catch Throwable rather than RuntimeException and Error, and clear the executor in a finally, so a checked exception thrown undeclared by a listener or Conductor provider, or an Error from close(), can't leave the half-started executor in place. Document that dbosShutDown can come without dbosLaunched when launch fails, and give the test that migrates a schema from empty the five-minute timeout CustomSchemaTest uses on slow CockroachDB runners. --- transact/src/main/java/dev/dbos/transact/DBOS.java | 11 +++++++---- .../transact/execution/DBOSLifecycleListener.java | 8 +++++++- .../test/java/dev/dbos/transact/FailedLaunchTest.java | 5 +++++ 3 files changed, 19 insertions(+), 5 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/DBOS.java b/transact/src/main/java/dev/dbos/transact/DBOS.java index fe4e9f1e..6ab1698e 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOS.java +++ b/transact/src/main/java/dev/dbos/transact/DBOS.java @@ -413,15 +413,18 @@ public void launch() { Set.copyOf(stepCheckpointStores), keepCompletedStepCheckpoints, alertHandler); - } catch (RuntimeException | Error e) { + } catch (Throwable e) { // Left in place, the executor would make every later launch() return at once with - // nothing running, and every other call fail on what it never started. + // nothing running, and every other call fail on what it never started. Throwable, not + // Exception: a listener or Conductor provider written in another JVM language can throw + // a checked exception undeclared, and the reset must happen whatever close() throws. try { executor.close(); - } catch (RuntimeException closeFailure) { + } catch (Throwable closeFailure) { e.addSuppressed(closeFailure); + } finally { + dbosExecutor.compareAndSet(executor, null); } - dbosExecutor.compareAndSet(executor, null); throw e; } } diff --git a/transact/src/main/java/dev/dbos/transact/execution/DBOSLifecycleListener.java b/transact/src/main/java/dev/dbos/transact/execution/DBOSLifecycleListener.java index 325e187f..022e6dfe 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSLifecycleListener.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSLifecycleListener.java @@ -10,6 +10,12 @@ public interface DBOSLifecycleListener { /** Called from within DBOS.launch, after workflow processing is allowed */ void dbosLaunched(DBOS dbos); - /** Called from within DBOS.shutdown, before workflow processing is stopped */ + /** + * Called from within DBOS.shutdown, before workflow processing is stopped. + * + *

Also called when {@code DBOS.launch()} fails, including before {@link #dbosLaunched} was + * called on this listener, or when this listener's own {@code dbosLaunched} threw. A later + * successful launch calls {@link #dbosLaunched} again. + */ void dbosShutDown(); } diff --git a/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java b/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java index b188b8cf..ad056c6e 100644 --- a/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java +++ b/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java @@ -10,10 +10,12 @@ import dev.dbos.transact.utils.PgContainer; import dev.dbos.transact.workflow.Workflow; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.AutoClose; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; /** * A launch that throws leaves the instance as if it had never been launched: a later launch tries @@ -37,7 +39,10 @@ private void assertNotLaunched() { assertTrue(e.getMessage().contains("before DBOS is launched"), e.getMessage()); } + // Migrates a schema from empty, which takes minutes on a slow CockroachDB runner, as in + // CustomSchemaTest. @Test + @Timeout(value = 5, unit = TimeUnit.MINUTES) public void aLaunchRefusedForAnUnmigratedSchemaCanBeRetriedOnceItIsMigrated() throws Exception { // A schema no migration has run on, which a launch with migrations off must refuse. var schema = "failed_launch_%d".formatted(System.nanoTime()); From 9b36fca0786598eac7dd94241db01a42c8e0fc08 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Thu, 12 Nov 2026 13:25:44 -0800 Subject: [PATCH 3/3] Stop every executor component even when one fails to stop DBOSExecutor.close() returned at its first exception, so a Conductor client or lifecycle listener that threw on stop left the queue service, scheduler, system database pool and thread pools running. A failed launch now cleans up through close() and then clears the executor, so a retry could run next to them. close() now runs every step, then throws the first failure with the rest attached. DBOS.shutdown() clears the executor even when close() throws, and a failed launch no longer tries to attach an exception to itself when close() throws the same one. --- .../src/main/java/dev/dbos/transact/DBOS.java | 15 +++- .../dbos/transact/execution/DBOSExecutor.java | 74 ++++++++++++++----- .../dev/dbos/transact/FailedLaunchTest.java | 47 ++++++++++++ 3 files changed, 115 insertions(+), 21 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/DBOS.java b/transact/src/main/java/dev/dbos/transact/DBOS.java index 6ab1698e..85fc675d 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOS.java +++ b/transact/src/main/java/dev/dbos/transact/DBOS.java @@ -421,7 +421,9 @@ public void launch() { try { executor.close(); } catch (Throwable closeFailure) { - e.addSuppressed(closeFailure); + if (closeFailure != e) { + e.addSuppressed(closeFailure); + } } finally { dbosExecutor.compareAndSet(executor, null); } @@ -450,9 +452,14 @@ void keepCompletedStepCheckpoints() { public void shutdown() { var current = dbosExecutor.get(); if (current != null) { - current.close(); - if (!dbosExecutor.compareAndSet(current, null)) { - logger.error("failed to set DBOS executor to null on shut down"); + try { + current.close(); + } finally { + // Cleared even when closing threw: close() has stopped all it could, and a closed + // executor left in place would make launch() return at once with nothing running. + if (!dbosExecutor.compareAndSet(current, null)) { + logger.error("failed to set DBOS executor to null on shut down"); + } } } logger.info("DBOS shut down"); diff --git a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java index fc1a40cb..73f35002 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -459,38 +459,53 @@ public void close() { if (isRunning.compareAndSet(true, false)) { logger.info("DBOS Executor stopping"); - if (adminServer != null) { - adminServer.stop(); - adminServer = null; + // Every step runs even when an earlier one throws, so one component that fails to stop + // leaves none of the others running. The first failure is thrown at the end, with the rest + // attached to it. + var failures = new ArrayList(); + + var admin = adminServer; + adminServer = null; + if (admin != null) { + closeStep(failures, admin::stop); } - if (conductor != null) { - conductor.stop(); - conductor = null; + var conductorConnection = conductor; + conductor = null; + if (conductorConnection != null) { + closeStep(failures, conductorConnection::stop); } - shutdownLifecycleListeners(); + closeStep(failures, this::shutdownLifecycleListeners); - if (schedulerService != null) { - schedulerService.close(); - } + var scheduler = schedulerService; schedulerService = null; - if (queueService != null) { - queueService.close(); + if (scheduler != null) { + closeStep(failures, scheduler::close); } + var queues = queueService; queueService = null; - if (systemDatabase != null) { - systemDatabase.close(); + if (queues != null) { + closeStep(failures, queues::close); } + var sysdb = systemDatabase; systemDatabase = null; + if (sysdb != null) { + closeStep(failures, sysdb::close); + } // Both are null when start() failed before creating them. if (timeoutScheduler != null) { - timeoutScheduler.shutdownNow(); + closeStep(failures, timeoutScheduler::shutdownNow); } if (executorService != null) { - var notRun = executorService.shutdownNow(); - logger.debug("Shutting down DBOS Executor threadpool. Tasks not run {}", notRun.size()); + closeStep( + failures, + () -> { + var notRun = executorService.shutdownNow(); + logger.debug( + "Shutting down DBOS Executor threadpool. Tasks not run {}", notRun.size()); + }); } this.workflowMap = null; @@ -498,6 +513,31 @@ public void close() { this.internalWorkflowMap = null; logger.debug("DBOS Executor stopped"); + + if (!failures.isEmpty()) { + var first = failures.get(0); + for (var other : failures.subList(1, failures.size())) { + if (other != first) { + first.addSuppressed(other); + } + } + if (first instanceof RuntimeException e) { + throw e; + } + if (first instanceof Error e) { + throw e; + } + // A checked exception a listener or Conductor provider threw undeclared. + throw new RuntimeException(first); + } + } + } + + private static void closeStep(List failures, Runnable step) { + try { + step.run(); + } catch (Throwable t) { + failures.add(t); } } diff --git a/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java b/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java index ad056c6e..49f195ad 100644 --- a/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java +++ b/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java @@ -6,6 +6,7 @@ import dev.dbos.transact.config.DBOSConfig; import dev.dbos.transact.execution.DBOSLifecycleListener; +import dev.dbos.transact.execution.QueueService; import dev.dbos.transact.migrations.MigrationManager; import dev.dbos.transact.utils.PgContainer; import dev.dbos.transact.workflow.Workflow; @@ -79,6 +80,52 @@ public void aLaunchThatFailsPartWayShutsDownWhatItStartedAndCanBeRetried() throw assertEquals("again", handle.getResult()); } + @Test + public void aShutdownStepThatThrowsLeavesNothingElseRunning() throws Exception { + create(pgContainer.dbosConfig()); + var listener = new FailingShutdownListener(); + dbos.integration().registerLifecycleListener(listener); + + var e = assertThrows(IllegalStateException.class, dbos::launch); + assertEquals("listener failed", e.getMessage()); + // The listener's shutdown failed, and the launch's own failure is still the one thrown. + assertEquals(1, e.getSuppressed().length); + assertEquals("shutdown failed", e.getSuppressed()[0].getMessage()); + assertTrue( + listener.failedLaunchQueues.isStopped(), + "the services after the failing step were still stopped"); + assertNotLaunched(); + + dbos.launch(); + var queues = DBOSTestAccess.getQueueService(dbos); + var thrown = assertThrows(AssertionError.class, dbos::shutdown); + assertEquals("shutdown failed", thrown.getMessage()); + assertTrue(queues.isStopped(), "a normal shutdown also stops what comes after the failure"); + assertNotLaunched(); + } + + /** + * Fails the first launch, and its own shutdown on every launch, with an Error, which nothing in + * DBOS catches for it. + */ + private static final class FailingShutdownListener implements DBOSLifecycleListener { + final AtomicInteger launches = new AtomicInteger(); + QueueService failedLaunchQueues; + + @Override + public void dbosLaunched(DBOS dbos) { + if (launches.incrementAndGet() == 1) { + failedLaunchQueues = DBOSTestAccess.getQueueService(dbos); + throw new IllegalStateException("listener failed"); + } + } + + @Override + public void dbosShutDown() { + throw new AssertionError("shutdown failed"); + } + } + /** Fails the first launch, after the executor has started its services, and not the next. */ private static final class FailingOnceListener implements DBOSLifecycleListener { final AtomicInteger launches = new AtomicInteger();