diff --git a/transact/src/main/java/dev/dbos/transact/DBOS.java b/transact/src/main/java/dev/dbos/transact/DBOS.java index 4af7494a..85fc675d 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,41 @@ 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 (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. 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 (Throwable closeFailure) { + if (closeFailure != e) { + e.addSuppressed(closeFailure); + } + } finally { + 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); } } } @@ -429,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 3ddcb55e..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,40 +459,85 @@ 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); + } - 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) { + closeStep(failures, timeoutScheduler::shutdownNow); + } + if (executorService != null) { + closeStep( + failures, + () -> { + var notRun = executorService.shutdownNow(); + logger.debug( + "Shutting down DBOS Executor threadpool. Tasks not run {}", notRun.size()); + }); + } this.workflowMap = null; this.instanceMap = null; 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/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 new file mode 100644 index 00000000..49f195ad --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/FailedLaunchTest.java @@ -0,0 +1,158 @@ +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.execution.QueueService; +import dev.dbos.transact.migrations.MigrationManager; +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 + * 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()); + } + + // 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()); + 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()); + } + + @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(); + 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; + } +}