Leave DBOS unlaunched when launch() fails

transact/src/main/java/dev/dbos/transact/DBOS.java

@@ -380,32 +380,55 @@ private void registerStepCheckpointStore(StepCheckpointStore store) {
     stepCheckpointStores.add(store);
   }
 
-  /** Launch DBOS, and start recovery. */
+  /**
+   * Launch DBOS, and start recovery.
+   *
+   * <p>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());
 
     if (dbosExecutor.get() == null) {
       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");

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<Throwable>();
+
+      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<Throwable> failures, Runnable step) {
+    try {
+      step.run();
+    } catch (Throwable t) {
+      failures.add(t);
     }
   }
 

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.
+   *
+   * <p>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();
 }

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;
+  }
+}