Create the factory before calling {@link dev.dbos.transact.DBOS#launch()}: DBOS needs to know + * every factory when it launches, so the constructor throws {@link IllegalStateException} after + * launch. + * *
{@code
* JdbiStepFactory factory = new JdbiStepFactory(dbos, Jdbi.create(dataSource));
*
@@ -232,6 +237,45 @@ protected void recordError(String workflowId, int stepId, Exception exception) {
}
}
+ @Override
+ protected void deleteCheckpoints(String workflowId, int fromStepId) {
+ inOwnTransaction(
+ h ->
+ h.createUpdate(TxStepSchema.deleteFromStepSql(schema))
+ .bind(0, workflowId)
+ .bind(1, fromStepId)
+ .execute());
+ }
+
+ @Override
+ protected void deleteCheckpoints(Collection workflowIds) {
+ inOwnTransaction(
+ h ->
+ h.createUpdate(TxStepSchema.deleteWorkflowsSql(schema))
+ .bindArray(0, String.class, workflowIds)
+ .execute());
+ }
+
+ // Jdbi's useTransaction does not commit a connection that arrives with autocommit off: it takes
+ // that for a transaction already in progress and leaves the commit to someone else. begin() and
+ // commit() on the handle commit either way.
+ private void inOwnTransaction(HandleConsumer work) {
+ try (var handle = jdbi.open()) {
+ handle.begin();
+ try {
+ work.useHandle(handle);
+ handle.commit();
+ } catch (RuntimeException e) {
+ try {
+ handle.rollback();
+ } catch (RuntimeException rollbackFailure) {
+ e.addSuppressed(rollbackFailure);
+ }
+ throw e;
+ }
+ }
+ }
+
private void recordResult(
Handle handle,
String workflowId,
diff --git a/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/DBOSTestAccess.java b/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/DBOSTestAccess.java
new file mode 100644
index 000000000..ff6e5faba
--- /dev/null
+++ b/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/DBOSTestAccess.java
@@ -0,0 +1,11 @@
+package dev.dbos.transact;
+
+import dev.dbos.transact.execution.DBOSExecutor;
+
+// Helper class to retrieve DBOS internals via package private methods
+public class DBOSTestAccess {
+
+ public static DBOSExecutor getDbosExecutor(DBOS dbos) {
+ return dbos.getDbosExecutor();
+ }
+}
diff --git a/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/jdbi/JdbiStepFactoryNoAutoCommitTest.java b/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/jdbi/JdbiStepFactoryNoAutoCommitTest.java
new file mode 100644
index 000000000..27cc4cc15
--- /dev/null
+++ b/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/jdbi/JdbiStepFactoryNoAutoCommitTest.java
@@ -0,0 +1,91 @@
+package dev.dbos.transact.jdbi;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.dbos.transact.DBOS;
+import dev.dbos.transact.DBOSTestAccess;
+import dev.dbos.transact.utils.PgContainer;
+
+import java.sql.SQLException;
+import java.util.List;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import org.jdbi.v3.core.Jdbi;
+import org.junit.jupiter.api.AutoClose;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * JdbiStepFactory on a pool configured without autocommit. A delete that did not commit explicitly
+ * would be rolled back when its connection went back to the pool.
+ */
+public class JdbiStepFactoryNoAutoCommitTest {
+ private static final String SCHEMA = "jdbi_no_autocommit";
+
+ @AutoClose final PgContainer pgContainer = new PgContainer();
+
+ @AutoClose DBOS dbos;
+ @AutoClose HikariDataSource dataSource;
+ @AutoClose HikariDataSource noAutoCommitDataSource;
+
+ @BeforeEach
+ void beforeEach() {
+ dataSource = pgContainer.dataSource();
+ var cfg = new HikariConfig();
+ cfg.setJdbcUrl(pgContainer.jdbcUrl());
+ cfg.setUsername(pgContainer.username());
+ cfg.setPassword(pgContainer.password());
+ cfg.setAutoCommit(false);
+ noAutoCommitDataSource = new HikariDataSource(cfg);
+
+ dbos = new DBOS(pgContainer.dbosConfig());
+ new JdbiStepFactory(dbos, Jdbi.create(noAutoCommitDataSource), SCHEMA);
+ dbos.launch();
+ }
+
+ private void insertCheckpoints(String workflowId, int... stepIds) throws SQLException {
+ try (var conn = dataSource.getConnection();
+ var stmt =
+ conn.prepareStatement(
+ "INSERT INTO %s.tx_step_outputs (workflow_id, step_id) VALUES (?, ?)"
+ .formatted(SCHEMA))) {
+ for (var stepId : stepIds) {
+ stmt.setString(1, workflowId);
+ stmt.setInt(2, stepId);
+ stmt.executeUpdate();
+ }
+ }
+ }
+
+ private int checkpoints(String workflowId) throws SQLException {
+ try (var conn = dataSource.getConnection();
+ var stmt =
+ conn.prepareStatement(
+ "SELECT COUNT(*) FROM %s.tx_step_outputs WHERE workflow_id = ?"
+ .formatted(SCHEMA))) {
+ stmt.setString(1, workflowId);
+ try (var rs = stmt.executeQuery()) {
+ rs.next();
+ return rs.getInt(1);
+ }
+ }
+ }
+
+ @Test
+ public void deletesCommit() throws Exception {
+ insertCheckpoints("wf-one", 0, 1, 2);
+ insertCheckpoints("wf-a", 0);
+ insertCheckpoints("wf-b", 0);
+ insertCheckpoints("wf-keep", 0);
+ var executor = DBOSTestAccess.getDbosExecutor(dbos);
+
+ executor.deleteStepCheckpoints("wf-one", 1);
+ assertEquals(1, checkpoints("wf-one"));
+
+ executor.deleteStepCheckpoints(List.of("wf-a", "wf-b"));
+ assertEquals(0, checkpoints("wf-a"));
+ assertEquals(0, checkpoints("wf-b"));
+ assertEquals(1, checkpoints("wf-keep"));
+ }
+}
diff --git a/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/jdbi/JdbiStepFactoryTest.java b/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/jdbi/JdbiStepFactoryTest.java
index 901a76be3..82d81cb7d 100644
--- a/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/jdbi/JdbiStepFactoryTest.java
+++ b/transact-jdbi-step-factory/src/test/java/dev/dbos/transact/jdbi/JdbiStepFactoryTest.java
@@ -7,6 +7,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import dev.dbos.transact.DBOS;
+import dev.dbos.transact.DBOSTestAccess;
import dev.dbos.transact.config.DBOSConfig;
import dev.dbos.transact.context.WorkflowOptions;
import dev.dbos.transact.database.SystemDatabase;
@@ -17,6 +18,7 @@
import dev.dbos.transact.workflow.WorkflowHandle;
import java.sql.SQLException;
+import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
@@ -228,6 +230,30 @@ private int getGreetCount(String user) throws SQLException {
}
}
+ @Test
+ public void testDeleteCheckpointsThroughDbos() throws Exception {
+ var wfid = "wf-delete";
+ try (var _o = new WorkflowOptions(wfid).setContext()) {
+ proxy.insertWorkflow("deleteUser");
+ }
+ assertEquals(1, DBUtils.getTxStepRows(dataSource, wfid).size());
+
+ // DBOS reaches the factory's checkpoints through the store it registered, and the delete
+ // commits through the library's own transaction.
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints(wfid, 0);
+ assertEquals(0, DBUtils.getTxStepRows(dataSource, wfid).size());
+
+ // The batch form, which binds the workflow IDs as an array.
+ for (var id : List.of("wf-batch-1", "wf-batch-2")) {
+ try (var _o = new WorkflowOptions(id).setContext()) {
+ proxy.insertWorkflow("deleteUser");
+ }
+ }
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints(List.of("wf-batch-1", "wf-batch-2"));
+ assertEquals(0, DBUtils.getTxStepRows(dataSource, "wf-batch-1").size());
+ assertEquals(0, DBUtils.getTxStepRows(dataSource, "wf-batch-2").size());
+ }
+
@Test
public void testInsert() throws Exception {
var wfid = "wf1";
diff --git a/transact-jooq-step-factory/src/main/java/dev/dbos/transact/jooq/JooqStepFactory.java b/transact-jooq-step-factory/src/main/java/dev/dbos/transact/jooq/JooqStepFactory.java
index a0bd1829e..c38cc44a2 100644
--- a/transact-jooq-step-factory/src/main/java/dev/dbos/transact/jooq/JooqStepFactory.java
+++ b/transact-jooq-step-factory/src/main/java/dev/dbos/transact/jooq/JooqStepFactory.java
@@ -9,6 +9,7 @@
import dev.dbos.transact.txstep.TxStepSchema;
import dev.dbos.transact.workflow.internal.StepResult;
+import java.util.Collection;
import java.util.Objects;
import java.util.Optional;
@@ -27,6 +28,10 @@
* org.jooq.Configuration} with a transaction already open; they must not commit or close the
* underlying connection themselves.
*
+ * Create the factory before calling {@link dev.dbos.transact.DBOS#launch()}: DBOS needs to know
+ * every factory when it launches, so the constructor throws {@link IllegalStateException} after
+ * launch.
+ *
*
{@code
* JooqStepFactory factory = new JooqStepFactory(dbos, dslContext);
*
@@ -208,6 +213,21 @@ protected void recordError(String workflowId, int stepId, Exception exception) {
}
}
+ @Override
+ protected void deleteCheckpoints(String workflowId, int fromStepId) {
+ dsl.transaction(
+ trx -> trx.dsl().execute(TxStepSchema.deleteFromStepSql(schema), workflowId, fromStepId));
+ }
+
+ @Override
+ protected void deleteCheckpoints(Collection workflowIds) {
+ // Bound on the JDBC connection: jOOQ can only bind an array when the dialect is set to one
+ // that has arrays, and the DSLContext may have been built without one.
+ dsl.transaction(
+ trx ->
+ trx.dsl().connection(conn -> TxStepSchema.deleteWorkflows(conn, schema, workflowIds)));
+ }
+
private void recordResult(
DSLContext ctx,
String workflowId,
diff --git a/transact-jooq-step-factory/src/test/java/dev/dbos/transact/DBOSTestAccess.java b/transact-jooq-step-factory/src/test/java/dev/dbos/transact/DBOSTestAccess.java
new file mode 100644
index 000000000..ff6e5faba
--- /dev/null
+++ b/transact-jooq-step-factory/src/test/java/dev/dbos/transact/DBOSTestAccess.java
@@ -0,0 +1,11 @@
+package dev.dbos.transact;
+
+import dev.dbos.transact.execution.DBOSExecutor;
+
+// Helper class to retrieve DBOS internals via package private methods
+public class DBOSTestAccess {
+
+ public static DBOSExecutor getDbosExecutor(DBOS dbos) {
+ return dbos.getDbosExecutor();
+ }
+}
diff --git a/transact-jooq-step-factory/src/test/java/dev/dbos/transact/jooq/JooqStepFactoryNoAutoCommitTest.java b/transact-jooq-step-factory/src/test/java/dev/dbos/transact/jooq/JooqStepFactoryNoAutoCommitTest.java
new file mode 100644
index 000000000..58e788d68
--- /dev/null
+++ b/transact-jooq-step-factory/src/test/java/dev/dbos/transact/jooq/JooqStepFactoryNoAutoCommitTest.java
@@ -0,0 +1,93 @@
+package dev.dbos.transact.jooq;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import dev.dbos.transact.DBOS;
+import dev.dbos.transact.DBOSTestAccess;
+import dev.dbos.transact.utils.PgContainer;
+
+import java.sql.SQLException;
+import java.util.List;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import org.jooq.SQLDialect;
+import org.jooq.impl.DSL;
+import org.junit.jupiter.api.AutoClose;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * JooqStepFactory on a pool configured without autocommit, and with jOOQ's default dialect, which
+ * cannot bind arrays itself. A delete that did not commit explicitly would be rolled back when its
+ * connection went back to the pool.
+ */
+public class JooqStepFactoryNoAutoCommitTest {
+ private static final String SCHEMA = "jooq_no_autocommit";
+
+ @AutoClose final PgContainer pgContainer = new PgContainer();
+
+ @AutoClose DBOS dbos;
+ @AutoClose HikariDataSource dataSource;
+ @AutoClose HikariDataSource noAutoCommitDataSource;
+
+ @BeforeEach
+ void beforeEach() {
+ dataSource = pgContainer.dataSource();
+ var cfg = new HikariConfig();
+ cfg.setJdbcUrl(pgContainer.jdbcUrl());
+ cfg.setUsername(pgContainer.username());
+ cfg.setPassword(pgContainer.password());
+ cfg.setAutoCommit(false);
+ noAutoCommitDataSource = new HikariDataSource(cfg);
+
+ dbos = new DBOS(pgContainer.dbosConfig());
+ new JooqStepFactory(dbos, DSL.using(noAutoCommitDataSource, SQLDialect.DEFAULT), SCHEMA);
+ dbos.launch();
+ }
+
+ private void insertCheckpoints(String workflowId, int... stepIds) throws SQLException {
+ try (var conn = dataSource.getConnection();
+ var stmt =
+ conn.prepareStatement(
+ "INSERT INTO %s.tx_step_outputs (workflow_id, step_id) VALUES (?, ?)"
+ .formatted(SCHEMA))) {
+ for (var stepId : stepIds) {
+ stmt.setString(1, workflowId);
+ stmt.setInt(2, stepId);
+ stmt.executeUpdate();
+ }
+ }
+ }
+
+ private int checkpoints(String workflowId) throws SQLException {
+ try (var conn = dataSource.getConnection();
+ var stmt =
+ conn.prepareStatement(
+ "SELECT COUNT(*) FROM %s.tx_step_outputs WHERE workflow_id = ?"
+ .formatted(SCHEMA))) {
+ stmt.setString(1, workflowId);
+ try (var rs = stmt.executeQuery()) {
+ rs.next();
+ return rs.getInt(1);
+ }
+ }
+ }
+
+ @Test
+ public void deletesCommit() throws Exception {
+ insertCheckpoints("wf-one", 0, 1, 2);
+ insertCheckpoints("wf-a", 0);
+ insertCheckpoints("wf-b", 0);
+ insertCheckpoints("wf-keep", 0);
+ var executor = DBOSTestAccess.getDbosExecutor(dbos);
+
+ executor.deleteStepCheckpoints("wf-one", 1);
+ assertEquals(1, checkpoints("wf-one"));
+
+ executor.deleteStepCheckpoints(List.of("wf-a", "wf-b"));
+ assertEquals(0, checkpoints("wf-a"));
+ assertEquals(0, checkpoints("wf-b"));
+ assertEquals(1, checkpoints("wf-keep"));
+ }
+}
diff --git a/transact-jooq-step-factory/src/test/java/dev/dbos/transact/jooq/JooqStepFactoryTest.java b/transact-jooq-step-factory/src/test/java/dev/dbos/transact/jooq/JooqStepFactoryTest.java
index 65753ff39..c0dc12993 100644
--- a/transact-jooq-step-factory/src/test/java/dev/dbos/transact/jooq/JooqStepFactoryTest.java
+++ b/transact-jooq-step-factory/src/test/java/dev/dbos/transact/jooq/JooqStepFactoryTest.java
@@ -7,6 +7,7 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import dev.dbos.transact.DBOS;
+import dev.dbos.transact.DBOSTestAccess;
import dev.dbos.transact.config.DBOSConfig;
import dev.dbos.transact.context.WorkflowOptions;
import dev.dbos.transact.database.SystemDatabase;
@@ -19,6 +20,7 @@
import dev.dbos.transact.workflow.WorkflowHandle;
import java.sql.SQLException;
+import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
@@ -218,6 +220,30 @@ private int getGreetCount(String user) throws SQLException {
}
}
+ @Test
+ public void testDeleteCheckpointsThroughDbos() throws Exception {
+ var wfid = "wf-delete";
+ try (var _o = new WorkflowOptions(wfid).setContext()) {
+ proxy.insertWorkflow("deleteUser");
+ }
+ assertEquals(1, DBUtils.getTxStepRows(dataSource, wfid).size());
+
+ // DBOS reaches the factory's checkpoints through the store it registered, and the delete
+ // commits through the library's own transaction.
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints(wfid, 0);
+ assertEquals(0, DBUtils.getTxStepRows(dataSource, wfid).size());
+
+ // The batch form, which binds the workflow IDs as an array.
+ for (var id : List.of("wf-batch-1", "wf-batch-2")) {
+ try (var _o = new WorkflowOptions(id).setContext()) {
+ proxy.insertWorkflow("deleteUser");
+ }
+ }
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints(List.of("wf-batch-1", "wf-batch-2"));
+ assertEquals(0, DBUtils.getTxStepRows(dataSource, "wf-batch-1").size());
+ assertEquals(0, DBUtils.getTxStepRows(dataSource, "wf-batch-2").size());
+ }
+
@Test
public void testInsert() throws Exception {
var wfid = "wf1";
diff --git a/transact-spring-txstep-starter/src/main/java/dev/dbos/transact/spring/txstep/TransactionalStepFactory.java b/transact-spring-txstep-starter/src/main/java/dev/dbos/transact/spring/txstep/TransactionalStepFactory.java
index 159e2f5a7..a59faee4c 100644
--- a/transact-spring-txstep-starter/src/main/java/dev/dbos/transact/spring/txstep/TransactionalStepFactory.java
+++ b/transact-spring-txstep-starter/src/main/java/dev/dbos/transact/spring/txstep/TransactionalStepFactory.java
@@ -6,6 +6,7 @@
import dev.dbos.transact.DBOS;
import dev.dbos.transact.database.SystemDatabase;
import dev.dbos.transact.execution.ThrowingSupplier;
+import dev.dbos.transact.internal.StepCheckpointStore;
import dev.dbos.transact.json.DBOSSerializer;
import dev.dbos.transact.json.SerializationUtil;
import dev.dbos.transact.txstep.PostgresStepFactory;
@@ -14,6 +15,7 @@
import java.sql.Connection;
import java.sql.SQLException;
+import java.util.Collection;
import java.util.Objects;
import java.util.Optional;
@@ -44,6 +46,10 @@ public class TransactionalStepFactory {
private final PlatformTransactionManager txManager;
private final String schema;
private final DBOSSerializer serializer;
+ // The store DBOS calls. A private class rather than this factory implementing the interface, so
+ // the interface's methods do not become public methods of the factory. One object, so a second
+ // initialize() does not add a second entry.
+ private final StepCheckpointStore checkpointStore = new CheckpointStore();
public TransactionalStepFactory(
DBOS dbos, DataSource dataSource, PlatformTransactionManager txManager, String schema) {
@@ -57,8 +63,12 @@ public TransactionalStepFactory(
/**
* Verifies the datasource is PostgreSQL and creates the {@code tx_step_outputs} table if it does
- * not already exist. Called lazily by {@code TransactionalStepRegistrar} only when annotated
- * methods are found — avoids any DB contact for applications that never use this starter.
+ * not already exist, then registers the factory's checkpoints with DBOS so it can delete them
+ * when it discards a workflow's history. Called lazily by {@code TransactionalStepRegistrar} only
+ * when annotated methods are found — avoids any DB contact for applications that never use this
+ * starter. The registrar runs before DBOS launches.
+ *
+ * @throws IllegalStateException if DBOS has already been launched
*/
public void initialize() {
try (var conn = dataSource.getConnection()) {
@@ -67,6 +77,42 @@ public void initialize() {
} catch (SQLException e) {
throw new RuntimeException(e);
}
+
+ // Registered here rather than in the constructor: until now the table may not exist.
+ dbos.integration().registerStepCheckpointStore(checkpointStore);
+ }
+
+ private final class CheckpointStore implements StepCheckpointStore {
+ @Override
+ public void deleteCheckpoints(String workflowId, int fromStepId) throws SQLException {
+ inTransaction(conn -> TxStepSchema.deleteFromStep(conn, schema, workflowId, fromStepId));
+ }
+
+ @Override
+ public void deleteCheckpoints(Collection workflowIds) throws SQLException {
+ if (!workflowIds.isEmpty()) {
+ inTransaction(conn -> TxStepSchema.deleteWorkflows(conn, schema, workflowIds));
+ }
+ }
+ }
+
+ @FunctionalInterface
+ private interface SqlWork {
+ void run(Connection conn) throws SQLException;
+ }
+
+ // A fresh connection with an explicit commit, as recordError does.
+ private void inTransaction(SqlWork work) throws SQLException {
+ try (var conn = dataSource.getConnection()) {
+ conn.setAutoCommit(false);
+ try {
+ work.run(conn);
+ conn.commit();
+ } catch (SQLException ex) {
+ conn.rollback();
+ throw ex;
+ }
+ }
}
private Optional checkExecution(String workflowId, int stepId, String stepName) {
diff --git a/transact-spring-txstep-starter/src/test/java/dev/dbos/transact/DBOSTestAccess.java b/transact-spring-txstep-starter/src/test/java/dev/dbos/transact/DBOSTestAccess.java
new file mode 100644
index 000000000..ff6e5faba
--- /dev/null
+++ b/transact-spring-txstep-starter/src/test/java/dev/dbos/transact/DBOSTestAccess.java
@@ -0,0 +1,11 @@
+package dev.dbos.transact;
+
+import dev.dbos.transact.execution.DBOSExecutor;
+
+// Helper class to retrieve DBOS internals via package private methods
+public class DBOSTestAccess {
+
+ public static DBOSExecutor getDbosExecutor(DBOS dbos) {
+ return dbos.getDbosExecutor();
+ }
+}
diff --git a/transact-spring-txstep-starter/src/test/java/dev/dbos/transact/spring/txstep/TransactionalStepJdbcIntegrationTest.java b/transact-spring-txstep-starter/src/test/java/dev/dbos/transact/spring/txstep/TransactionalStepJdbcIntegrationTest.java
index f6e938502..2777fadd0 100644
--- a/transact-spring-txstep-starter/src/test/java/dev/dbos/transact/spring/txstep/TransactionalStepJdbcIntegrationTest.java
+++ b/transact-spring-txstep-starter/src/test/java/dev/dbos/transact/spring/txstep/TransactionalStepJdbcIntegrationTest.java
@@ -4,6 +4,7 @@
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import dev.dbos.transact.DBOS;
+import dev.dbos.transact.DBOSTestAccess;
import dev.dbos.transact.context.WorkflowOptions;
import dev.dbos.transact.database.SystemDatabase;
import dev.dbos.transact.spring.DBOSAutoConfiguration;
@@ -195,6 +196,39 @@ void goldenPath() throws SQLException {
}
}
+ @Test
+ void deleteCheckpointsThroughDbos() throws SQLException {
+ try (var db = new TransactionalStepTest.TestDatabase()) {
+ runner(db)
+ .run(
+ ctx -> {
+ assertThat(ctx).hasNotFailed();
+ var workflow = ctx.getBean(OrderWorkflowService.class);
+ var wfid = "wf-jdbc-int-delete";
+
+ try (var _o = new WorkflowOptions(wfid).setContext()) {
+ workflow.processOrder("ord-d", "Widget", 1);
+ }
+ assertThat(TransactionalStepTest.getTxRows(db.dataSource, wfid)).hasSize(1);
+
+ // The registrar initialized the factory before launch, which registered its
+ // checkpoints with DBOS.
+ var executor = DBOSTestAccess.getDbosExecutor(ctx.getBean(DBOS.class));
+ executor.deleteStepCheckpoints(wfid, 0);
+ assertThat(TransactionalStepTest.getTxRows(db.dataSource, wfid)).isEmpty();
+
+ // The batch form.
+ var batchId = wfid + "-batch";
+ try (var _o = new WorkflowOptions(batchId).setContext()) {
+ workflow.processOrder("ord-d2", "Widget", 1);
+ }
+ assertThat(TransactionalStepTest.getTxRows(db.dataSource, batchId)).hasSize(1);
+ executor.deleteStepCheckpoints(java.util.List.of(batchId));
+ assertThat(TransactionalStepTest.getTxRows(db.dataSource, batchId)).isEmpty();
+ });
+ }
+ }
+
@Test
void isolationLevel() {
try (var db = new TransactionalStepTest.TestDatabase()) {
diff --git a/transact/src/main/java/dev/dbos/transact/DBOS.java b/transact/src/main/java/dev/dbos/transact/DBOS.java
index 3065d1fe1..2aa6966c6 100644
--- a/transact/src/main/java/dev/dbos/transact/DBOS.java
+++ b/transact/src/main/java/dev/dbos/transact/DBOS.java
@@ -9,6 +9,7 @@
import dev.dbos.transact.internal.DBOSIntegration;
import dev.dbos.transact.internal.DBOSInvocationHandler;
import dev.dbos.transact.internal.QueueRegistry;
+import dev.dbos.transact.internal.StepCheckpointStore;
import dev.dbos.transact.internal.WorkflowRegistry;
import dev.dbos.transact.migrations.MigrationManager;
import dev.dbos.transact.workflow.Debouncer;
@@ -64,6 +65,7 @@ public class DBOS implements AutoCloseable {
private final WorkflowRegistry workflowRegistry = new WorkflowRegistry();
private final QueueRegistry queueRegistry = new QueueRegistry();
private final Set lifecycleRegistry = ConcurrentHashMap.newKeySet();
+ private final Set stepCheckpointStores = ConcurrentHashMap.newKeySet();
private final DBOSConfig config;
private final AtomicReference dbosExecutor = new AtomicReference<>();
private final DBOSIntegration integration;
@@ -88,7 +90,11 @@ public DBOS(@NonNull DBOSConfig config) {
this.config = new DBOSConfig(config);
this.integration =
new DBOSIntegration(
- this.config, this.workflowRegistry, dbosExecutor::get, this::registerLifecycleListener);
+ this.config,
+ this.workflowRegistry,
+ dbosExecutor::get,
+ this::registerLifecycleListener,
+ this::registerStepCheckpointStore);
// Register the built-in debouncer workflow directly (without a proxy). Debouncer no
// longer starts it, but one an older SDK version enqueued can still be recovered here when the
// application version is pinned across the upgrade.
@@ -354,6 +360,16 @@ public void registerAlertHandler(AlertHandler handler) {
return dbosExecutor.get();
}
+ // The executor takes a snapshot of the stores at launch, so a factory created afterwards would
+ // never have its checkpoints deleted.
+ private void registerStepCheckpointStore(StepCheckpointStore store) {
+ if (dbosExecutor.get() != null) {
+ throw new IllegalStateException(
+ "Transactional step factories must be created before DBOS is launched");
+ }
+ stepCheckpointStores.add(store);
+ }
+
/** Launch DBOS, and start recovery. */
public void launch() {
logger.info("Launching DBOS v{}", DBOS.version());
@@ -377,6 +393,7 @@ public void launch() {
workflowRegistry.getInternalWorkflowSnapshot(),
workflowRegistry.getInternalInstanceSnapshot(),
queueRegistry.getSnapshot(),
+ Set.copyOf(stepCheckpointStores),
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 e7ca77c3f..96c715eff 100644
--- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java
+++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java
@@ -27,6 +27,7 @@
import dev.dbos.transact.exceptions.DBOSWorkflowExecutionConflictException;
import dev.dbos.transact.exceptions.DBOSWorkflowFunctionNotFoundException;
import dev.dbos.transact.internal.AppVersionComputer;
+import dev.dbos.transact.internal.StepCheckpointStore;
import dev.dbos.transact.internal.Validation;
import dev.dbos.transact.internal.WorkflowRegistry;
import dev.dbos.transact.json.DBOSSerializer;
@@ -62,6 +63,7 @@
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+import java.sql.SQLException;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
@@ -173,6 +175,7 @@ public void close() {
private ExecutorService executorService;
private ScheduledExecutorService timeoutScheduler;
private AlertHandler alertHandler;
+ private Set stepCheckpointStores = Set.of();
public DBOSExecutor(DBOSConfig config) {
this.config = config;
@@ -240,6 +243,7 @@ public void start(
Map internalWorkflowMap,
Map internalInstanceMap,
List queues,
+ Set stepCheckpointStores,
AlertHandler alertHandler) {
if (isRunning.compareAndSet(false, true)) {
@@ -251,6 +255,7 @@ public void start(
this.queueMap =
queues.stream().collect(Collectors.toUnmodifiableMap(Queue::name, queue -> queue));
this.listeners = listenerSet;
+ this.stepCheckpointStores = stepCheckpointStores;
this.alertHandler = alertHandler;
if (this.appVersion == null || this.appVersion.isEmpty()) {
@@ -1038,6 +1043,55 @@ public void updateWorkflowAttributes(String workflowId, Map attr
null);
}
+ /** Whether any transactional step factory has registered its checkpoints. */
+ public boolean hasStepCheckpointStores() {
+ return !stepCheckpointStores.isEmpty();
+ }
+
+ /**
+ * Deletes a workflow's checkpoints from {@code fromStepId} on, in every registered transactional
+ * step factory. Each store commits its own delete, so a failure part way leaves the stores before
+ * it cleared; deleting again is safe.
+ */
+ public void deleteStepCheckpoints(String workflowId, int fromStepId) {
+ for (var store : stepCheckpointStores) {
+ try {
+ store.deleteCheckpoints(workflowId, fromStepId);
+ } catch (SQLException | RuntimeException e) {
+ throw new RuntimeException(
+ "Failed to delete the transactional step checkpoints of workflow " + workflowId, e);
+ }
+ }
+ }
+
+ /**
+ * Deletes every checkpoint of the given workflows, in every registered transactional step
+ * factory. As with {@link #deleteStepCheckpoints(String, int)}, a failure part way leaves the
+ * stores before it cleared, and deleting again is safe.
+ */
+ public void deleteStepCheckpoints(Collection workflowIds) {
+ for (var store : stepCheckpointStores) {
+ try {
+ store.deleteCheckpoints(workflowIds);
+ } catch (SQLException | RuntimeException e) {
+ throw new RuntimeException(
+ "Failed to delete the transactional step checkpoints of "
+ + describeWorkflows(workflowIds),
+ e);
+ }
+ }
+ }
+
+ // A retention batch can hold thousands of IDs, so name only the first few.
+ private static String describeWorkflows(Collection workflowIds) {
+ var shown = workflowIds.stream().limit(5).toList();
+ var more = workflowIds.size() - shown.size();
+ return workflowIds.size()
+ + " workflows "
+ + shown
+ + (more > 0 ? " and %d more".formatted(more) : "");
+ }
+
public WorkflowHandle forkWorkflow(
String workflowId, int startStep, ForkOptions options) {
diff --git a/transact/src/main/java/dev/dbos/transact/internal/DBOSIntegration.java b/transact/src/main/java/dev/dbos/transact/internal/DBOSIntegration.java
index 7c0f271e8..c13a773cc 100644
--- a/transact/src/main/java/dev/dbos/transact/internal/DBOSIntegration.java
+++ b/transact/src/main/java/dev/dbos/transact/internal/DBOSIntegration.java
@@ -36,16 +36,19 @@ public class DBOSIntegration {
private final WorkflowRegistry workflowRegistry;
private final Supplier executorSupplier;
private final Consumer listenerConsumer;
+ private final Consumer checkpointStoreConsumer;
public DBOSIntegration(
@NonNull DBOSConfig config,
@NonNull WorkflowRegistry workflowRegistry,
@NonNull Supplier executorSupplier,
- @NonNull Consumer lifecycleConsumer) {
+ @NonNull Consumer lifecycleConsumer,
+ @NonNull Consumer checkpointStoreConsumer) {
this.config = Objects.requireNonNull(config);
this.workflowRegistry = Objects.requireNonNull(workflowRegistry);
this.executorSupplier = Objects.requireNonNull(executorSupplier);
this.listenerConsumer = Objects.requireNonNull(lifecycleConsumer);
+ this.checkpointStoreConsumer = Objects.requireNonNull(checkpointStoreConsumer);
}
private DBOSExecutor executor(String caller) {
@@ -75,6 +78,18 @@ public void registerLifecycleListener(@NonNull DBOSLifecycleListener listener) {
listenerConsumer.accept(listener);
}
+ /**
+ * Register the checkpoints a transactional step factory keeps in its own database, so DBOS can
+ * delete a workflow's checkpoints there when it discards that workflow's history. Registering the
+ * same store again has no effect.
+ *
+ * @param store the factory's checkpoints; must not be {@code null}
+ * @throws IllegalStateException if called after DBOS is launched
+ */
+ public void registerStepCheckpointStore(@NonNull StepCheckpointStore store) {
+ checkpointStoreConsumer.accept(Objects.requireNonNull(store));
+ }
+
/**
* Register a workflow method with DBOS. This method is used internally by the proxy registration
* process and should not typically be called directly by application code.
diff --git a/transact/src/main/java/dev/dbos/transact/internal/StepCheckpointStore.java b/transact/src/main/java/dev/dbos/transact/internal/StepCheckpointStore.java
new file mode 100644
index 000000000..a968f013c
--- /dev/null
+++ b/transact/src/main/java/dev/dbos/transact/internal/StepCheckpointStore.java
@@ -0,0 +1,32 @@
+package dev.dbos.transact.internal;
+
+import java.sql.SQLException;
+import java.util.Collection;
+
+/**
+ * Checkpoints a transactional step factory keeps in its own database, outside the system database.
+ * Each factory registers one through {@link DBOSIntegration#registerStepCheckpointStore}, so DBOS
+ * can drop a workflow's checkpoints when it discards that workflow's history.
+ *
+ * This interface is not part of the primary public API and may change without
+ * notice.
+ */
+public interface StepCheckpointStore {
+
+ /**
+ * Deletes a workflow's checkpoints from {@code fromStepId} on. Deleting checkpoints that are
+ * already gone is not an error.
+ *
+ * @param workflowId the workflow whose checkpoints to delete
+ * @param fromStepId the first step ID to delete; 0 deletes them all
+ */
+ void deleteCheckpoints(String workflowId, int fromStepId) throws SQLException;
+
+ /**
+ * Deletes every checkpoint of the given workflows, in one statement. Workflows with no
+ * checkpoints are not an error.
+ *
+ * @param workflowIds the workflows whose checkpoints to delete
+ */
+ void deleteCheckpoints(Collection workflowIds) throws SQLException;
+}
diff --git a/transact/src/main/java/dev/dbos/transact/txstep/JdbcStepFactory.java b/transact/src/main/java/dev/dbos/transact/txstep/JdbcStepFactory.java
index 7218e1eb8..e93ab8366 100644
--- a/transact/src/main/java/dev/dbos/transact/txstep/JdbcStepFactory.java
+++ b/transact/src/main/java/dev/dbos/transact/txstep/JdbcStepFactory.java
@@ -9,6 +9,7 @@
import java.sql.Connection;
import java.sql.SQLException;
+import java.util.Collection;
import java.util.Objects;
import java.util.Optional;
@@ -22,6 +23,10 @@
* User lambdas passed to {@code txStep} receive a {@link Connection} with a transaction already
* started; they should not call {@code commit} or {@code close} themselves.
*
+ * Create the factory before calling {@link dev.dbos.transact.DBOS#launch()}: DBOS needs to know
+ * every factory when it launches, so the constructor throws {@link IllegalStateException} after
+ * launch.
+ *
*
{@code
* JdbcStepFactory factory = new JdbcStepFactory(dbos, dataSource);
*
@@ -200,6 +205,26 @@ public void txStep(
options);
}
+ @Override
+ protected void deleteCheckpoints(String workflowId, int fromStepId) throws SQLException {
+ executeTransaction(
+ dataSource,
+ (Connection conn) -> {
+ TxStepSchema.deleteFromStep(conn, schema, workflowId, fromStepId);
+ return null;
+ });
+ }
+
+ @Override
+ protected void deleteCheckpoints(Collection workflowIds) throws SQLException {
+ executeTransaction(
+ dataSource,
+ (Connection conn) -> {
+ TxStepSchema.deleteWorkflows(conn, schema, workflowIds);
+ return null;
+ });
+ }
+
private static R executeTransaction(
final DataSource ds, TransactionalFunction func) throws X {
return executeTransaction(ds, IsolationLevel.DEFAULT, func);
diff --git a/transact/src/main/java/dev/dbos/transact/txstep/PostgresStepFactory.java b/transact/src/main/java/dev/dbos/transact/txstep/PostgresStepFactory.java
index 75a638f5b..db9088519 100644
--- a/transact/src/main/java/dev/dbos/transact/txstep/PostgresStepFactory.java
+++ b/transact/src/main/java/dev/dbos/transact/txstep/PostgresStepFactory.java
@@ -1,12 +1,15 @@
package dev.dbos.transact.txstep;
import dev.dbos.transact.DBOS;
+import dev.dbos.transact.database.SqlTransaction;
import dev.dbos.transact.database.SystemDatabase;
+import dev.dbos.transact.internal.StepCheckpointStore;
import dev.dbos.transact.json.DBOSSerializer;
import dev.dbos.transact.workflow.internal.StepResult;
import java.sql.Connection;
import java.sql.SQLException;
+import java.util.Collection;
import java.util.Objects;
import java.util.Optional;
@@ -19,13 +22,23 @@
* with the DBOS runtime.
*
* The constructor verifies that the datasource is PostgreSQL and creates the {@code
- * tx_step_outputs} table (and its enclosing schema) if they do not already exist.
+ * tx_step_outputs} table (and its enclosing schema) if they do not already exist. It then registers
+ * the factory's checkpoints with DBOS, which must not be launched yet, so DBOS can delete them when
+ * it discards a workflow's history.
+ *
+ *
Subclasses: the factory is registered before the subclass constructor finishes, and stays
+ * registered even if that constructor then throws, so a subclass constructor should not fail after
+ * calling {@code super}.
*/
public abstract class PostgresStepFactory {
protected final DBOS dbos;
protected final String schema;
protected final DBOSSerializer serializer;
+ private final ConnectionOpener opener;
+ // The store DBOS calls. A private class rather than this factory implementing the interface, so
+ // the interface's methods do not become public methods of every factory.
+ private final StepCheckpointStore checkpointStore = new CheckpointStore();
@FunctionalInterface
protected interface ConnectionOpener {
@@ -38,6 +51,7 @@ protected PostgresStepFactory(
var config = dbos.integration().config();
this.schema = SystemDatabase.sanitizeSchema(schema == null ? config.databaseSchema() : schema);
this.serializer = serializer == null ? config.serializer() : serializer;
+ this.opener = opener;
try (var conn = opener.open()) {
TxStepSchema.verifyPostgres(conn);
@@ -45,6 +59,48 @@ protected PostgresStepFactory(
} catch (SQLException e) {
throw new RuntimeException(e);
}
+
+ dbos.integration().registerStepCheckpointStore(checkpointStore);
+ }
+
+ private final class CheckpointStore implements StepCheckpointStore {
+ @Override
+ public void deleteCheckpoints(String workflowId, int fromStepId) throws SQLException {
+ PostgresStepFactory.this.deleteCheckpoints(workflowId, fromStepId);
+ }
+
+ @Override
+ public void deleteCheckpoints(Collection workflowIds) throws SQLException {
+ if (!workflowIds.isEmpty()) {
+ PostgresStepFactory.this.deleteCheckpoints(workflowIds);
+ }
+ }
+ }
+
+ /**
+ * Deletes a workflow's checkpoints from {@code fromStepId} on, in a transaction of its own that
+ * commits before this returns. DBOS calls it when it discards the workflow's history.
+ *
+ * This default opens a connection and commits explicitly. Subclasses override it to go through
+ * their database library's own transaction handling.
+ */
+ protected void deleteCheckpoints(String workflowId, int fromStepId) throws SQLException {
+ try (var conn = opener.open()) {
+ SqlTransaction.run(conn, c -> TxStepSchema.deleteFromStep(c, schema, workflowId, fromStepId));
+ }
+ }
+
+ /**
+ * Deletes every checkpoint of the given workflows, never empty, in a transaction of its own that
+ * commits before this returns. DBOS calls it when it deletes the workflows.
+ *
+ *
This default opens a connection and commits explicitly. Subclasses override it to go through
+ * their database library's own transaction handling.
+ */
+ protected void deleteCheckpoints(Collection workflowIds) throws SQLException {
+ try (var conn = opener.open()) {
+ SqlTransaction.run(conn, c -> TxStepSchema.deleteWorkflows(c, schema, workflowIds));
+ }
}
protected abstract Optional checkExecution(
diff --git a/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java b/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java
index 4d7947de8..94a0ca6eb 100644
--- a/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java
+++ b/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java
@@ -6,6 +6,7 @@
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.Collection;
import java.util.Optional;
/** Shared SQL DDL and query constants for the {@code tx_step_outputs} table. */
@@ -38,6 +39,10 @@ PRIMARY KEY (workflow_id, step_id)
.formatted(schema));
stmt.executeBatch();
}
+ // A pool configured without autocommit would otherwise roll the DDL back on close.
+ if (!conn.getAutoCommit()) {
+ conn.commit();
+ }
}
public static String checkSql(String schema) {
@@ -49,6 +54,46 @@ public static String checkSql(String schema) {
.formatted(schema);
}
+ /** Deletes a workflow's checkpoints from a step on: parameters are the workflow and step IDs. */
+ public static String deleteFromStepSql(String schema) {
+ return """
+ DELETE FROM "%s".tx_step_outputs
+ WHERE workflow_id = ? AND step_id >= ?
+ """
+ .formatted(schema);
+ }
+
+ /** Runs {@link #deleteFromStepSql} on a connection; the caller owns the transaction. */
+ public static void deleteFromStep(
+ Connection conn, String schema, String workflowId, int fromStepId) throws SQLException {
+ try (var stmt = conn.prepareStatement(deleteFromStepSql(schema))) {
+ stmt.setString(1, workflowId);
+ stmt.setInt(2, fromStepId);
+ stmt.executeUpdate();
+ }
+ }
+
+ /** Deletes every checkpoint of a set of workflows: the parameter is a text[] of workflow IDs. */
+ public static String deleteWorkflowsSql(String schema) {
+ return """
+ DELETE FROM "%s".tx_step_outputs
+ WHERE workflow_id = ANY(?)
+ """
+ .formatted(schema);
+ }
+
+ /** Runs {@link #deleteWorkflowsSql} on a connection; the caller owns the transaction. */
+ public static void deleteWorkflows(Connection conn, String schema, Collection workflowIds)
+ throws SQLException {
+ var ids = conn.createArrayOf("text", workflowIds.toArray(String[]::new));
+ try (var stmt = conn.prepareStatement(deleteWorkflowsSql(schema))) {
+ stmt.setArray(1, ids);
+ stmt.executeUpdate();
+ } finally {
+ ids.free();
+ }
+ }
+
public static String upsertSql(String schema) {
return """
INSERT INTO "%s".tx_step_outputs
diff --git a/transact/src/test/java/dev/dbos/transact/txstep/StepCheckpointRegistryTest.java b/transact/src/test/java/dev/dbos/transact/txstep/StepCheckpointRegistryTest.java
new file mode 100644
index 000000000..836162cdd
--- /dev/null
+++ b/transact/src/test/java/dev/dbos/transact/txstep/StepCheckpointRegistryTest.java
@@ -0,0 +1,301 @@
+package dev.dbos.transact.txstep;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import dev.dbos.transact.DBOS;
+import dev.dbos.transact.DBOSTestAccess;
+import dev.dbos.transact.config.DBOSConfig;
+import dev.dbos.transact.context.WorkflowOptions;
+import dev.dbos.transact.internal.StepCheckpointStore;
+import dev.dbos.transact.utils.DBUtils;
+import dev.dbos.transact.utils.PgContainer;
+import dev.dbos.transact.utils.TxStepOutputRow;
+import dev.dbos.transact.workflow.Workflow;
+import dev.dbos.transact.workflow.internal.StepResult;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import org.junit.jupiter.api.AutoClose;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+interface CheckpointService {
+ void threeSteps() throws SQLException;
+}
+
+class CheckpointServiceImpl implements CheckpointService {
+ private final JdbcStepFactory factory;
+
+ CheckpointServiceImpl(JdbcStepFactory factory) {
+ this.factory = factory;
+ }
+
+ @Override
+ @Workflow
+ public void threeSteps() throws SQLException {
+ for (var name : List.of("one", "two", "three")) {
+ factory.txStep((Connection conn) -> {}, name);
+ }
+ }
+}
+
+public class StepCheckpointRegistryTest {
+
+ @AutoClose final PgContainer pgContainer = new PgContainer();
+
+ DBOSConfig dbosConfig;
+ @AutoClose DBOS dbos;
+ @AutoClose HikariDataSource dataSource;
+ // A delete that relied on autocommit would be rolled back when its connection went back here.
+ @AutoClose HikariDataSource noAutoCommitDataSource;
+
+ @BeforeEach
+ void beforeEach() {
+ dbosConfig = pgContainer.dbosConfig();
+ dataSource = pgContainer.dataSource();
+ var cfg = new HikariConfig();
+ cfg.setJdbcUrl(pgContainer.jdbcUrl());
+ cfg.setUsername(pgContainer.username());
+ cfg.setPassword(pgContainer.password());
+ cfg.setAutoCommit(false);
+ noAutoCommitDataSource = new HikariDataSource(cfg);
+ dbos = new DBOS(dbosConfig);
+ }
+
+ private List checkpoints(String schema, String workflowId) throws SQLException {
+ return DBUtils.getTxStepRows(dataSource, workflowId, schema).stream()
+ .map(TxStepOutputRow::stepId)
+ .sorted()
+ .toList();
+ }
+
+ @Test
+ public void deletesAWorkflowsCheckpointsFromAStepOn() throws Exception {
+ var factory = new JdbcStepFactory(dbos, noAutoCommitDataSource, "registry_jdbc");
+ var proxy = dbos.registerProxy(CheckpointService.class, new CheckpointServiceImpl(factory));
+ dbos.launch();
+
+ try (var o = new WorkflowOptions("wf-registry").setContext()) {
+ proxy.threeSteps();
+ }
+ assertEquals(List.of(0, 1, 2), checkpoints("registry_jdbc", "wf-registry"));
+
+ var executor = DBOSTestAccess.getDbosExecutor(dbos);
+ assertTrue(executor.hasStepCheckpointStores());
+ executor.deleteStepCheckpoints("wf-registry", 1);
+ assertEquals(List.of(0), checkpoints("registry_jdbc", "wf-registry"));
+
+ // Deleting again is not an error, and 0 deletes the rest.
+ executor.deleteStepCheckpoints("wf-registry", 1);
+ executor.deleteStepCheckpoints("wf-registry", 0);
+ assertEquals(List.of(), checkpoints("registry_jdbc", "wf-registry"));
+ }
+
+ @Test
+ public void theDefaultDeleteCommits() throws Exception {
+ // A factory that inherits PostgresStepFactory's own delete rather than overriding it.
+ new MinimalStepFactory(dbos, noAutoCommitDataSource, "registry_default");
+ dbos.launch();
+ try (var conn = dataSource.getConnection();
+ var stmt = conn.createStatement()) {
+ stmt.execute(
+ "INSERT INTO registry_default.tx_step_outputs (workflow_id, step_id) VALUES"
+ + " ('wf-default', 0), ('wf-default', 1), ('wf-other', 1)");
+ }
+
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints("wf-default", 1);
+
+ assertEquals(List.of(0), checkpoints("registry_default", "wf-default"));
+ assertEquals(List.of(1), checkpoints("registry_default", "wf-other"));
+ }
+
+ @Test
+ public void aFactoryCreatedAfterLaunchIsRefused() {
+ dbos.launch();
+ var refused =
+ assertThrows(
+ IllegalStateException.class,
+ () -> new JdbcStepFactory(dbos, dataSource, "registry_late"));
+ assertTrue(refused.getMessage().contains("before DBOS is launched"), refused.getMessage());
+ assertFalse(DBOSTestAccess.getDbosExecutor(dbos).hasStepCheckpointStores());
+ }
+
+ @Test
+ public void aStoreRegisteredTwiceHasOneEntry() {
+ var deletes = new AtomicInteger();
+ var store = new RecordingStore(deletes, null);
+ dbos.integration().registerStepCheckpointStore(store);
+ dbos.integration().registerStepCheckpointStore(store);
+ dbos.launch();
+
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints("wf", 0);
+ assertEquals(1, deletes.get());
+ }
+
+ @Test
+ public void aFailedDeleteNamesTheWorkflow() {
+ dbos.integration()
+ .registerStepCheckpointStore(
+ new RecordingStore(new AtomicInteger(), new SQLException("datasource down")));
+ dbos.launch();
+
+ var thrown =
+ assertThrows(
+ RuntimeException.class,
+ () -> DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints("wf-down", 0));
+ assertTrue(thrown.getMessage().contains("wf-down"), thrown.getMessage());
+ assertEquals("datasource down", thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void anUncheckedFailureNamesTheWorkflowToo() {
+ // Jdbi and jOOQ report database errors as unchecked exceptions.
+ dbos.integration()
+ .registerStepCheckpointStore(
+ new RecordingStore(new AtomicInteger(), new IllegalStateException("pool closed")));
+ dbos.launch();
+
+ var thrown =
+ assertThrows(
+ RuntimeException.class,
+ () -> DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints("wf-closed", 0));
+ assertTrue(thrown.getMessage().contains("wf-closed"), thrown.getMessage());
+ assertEquals("pool closed", thrown.getCause().getMessage());
+ }
+
+ @Test
+ public void aFailedBatchNamesOnlyTheFirstFewWorkflows() {
+ dbos.integration()
+ .registerStepCheckpointStore(
+ new RecordingStore(new AtomicInteger(), new SQLException("datasource down")));
+ dbos.launch();
+
+ var ids = java.util.stream.IntStream.range(0, 12).mapToObj(i -> "wf-" + i).toList();
+ var thrown =
+ assertThrows(
+ RuntimeException.class,
+ () -> DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints(ids));
+ var message = thrown.getMessage();
+ assertTrue(message.contains("12 workflows"), message);
+ assertTrue(message.contains("wf-0") && message.contains("wf-4"), message);
+ assertFalse(message.contains("wf-5"), message);
+ assertTrue(message.contains("and 7 more"), message);
+ }
+
+ @Test
+ public void deletesEveryCheckpointOfTheGivenWorkflows() throws Exception {
+ var factory = new JdbcStepFactory(dbos, noAutoCommitDataSource, "registry_batch");
+ var proxy = dbos.registerProxy(CheckpointService.class, new CheckpointServiceImpl(factory));
+ dbos.launch();
+
+ for (var id : List.of("wf-a", "wf-b", "wf-c")) {
+ try (var o = new WorkflowOptions(id).setContext()) {
+ proxy.threeSteps();
+ }
+ }
+
+ var executor = DBOSTestAccess.getDbosExecutor(dbos);
+ executor.deleteStepCheckpoints(List.of("wf-a", "wf-c", "wf-missing"));
+ assertEquals(List.of(), checkpoints("registry_batch", "wf-a"));
+ assertEquals(List.of(0, 1, 2), checkpoints("registry_batch", "wf-b"));
+ assertEquals(List.of(), checkpoints("registry_batch", "wf-c"));
+
+ executor.deleteStepCheckpoints(List.of());
+ assertEquals(List.of(0, 1, 2), checkpoints("registry_batch", "wf-b"));
+ }
+
+ @Test
+ public void anEmptyBatchDoesNotReachTheDatabase() {
+ var opens = new AtomicInteger();
+ new MinimalStepFactory(
+ dbos,
+ () -> {
+ opens.incrementAndGet();
+ return dataSource.getConnection();
+ },
+ "registry_empty");
+ dbos.launch();
+ var opensAfterCreate = opens.get();
+
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints(List.of());
+ assertEquals(opensAfterCreate, opens.get());
+ }
+
+ @Test
+ public void theDefaultBatchDeleteCommits() throws Exception {
+ new MinimalStepFactory(dbos, noAutoCommitDataSource, "registry_default_batch");
+ dbos.launch();
+ try (var conn = dataSource.getConnection();
+ var stmt = conn.createStatement()) {
+ stmt.execute(
+ "INSERT INTO registry_default_batch.tx_step_outputs (workflow_id, step_id) VALUES"
+ + " ('wf-a', 0), ('wf-a', 1), ('wf-b', 0), ('wf-c', 0)");
+ }
+
+ DBOSTestAccess.getDbosExecutor(dbos).deleteStepCheckpoints(List.of("wf-a", "wf-b"));
+
+ assertEquals(List.of(), checkpoints("registry_default_batch", "wf-a"));
+ assertEquals(List.of(), checkpoints("registry_default_batch", "wf-b"));
+ assertEquals(List.of(0), checkpoints("registry_default_batch", "wf-c"));
+ }
+
+ /** Counts its calls, and fails every one with {@code failure} when it is not null. */
+ static class RecordingStore implements StepCheckpointStore {
+ private final AtomicInteger calls;
+ private final Exception failure;
+
+ RecordingStore(AtomicInteger calls, Exception failure) {
+ this.calls = calls;
+ this.failure = failure;
+ }
+
+ @Override
+ public void deleteCheckpoints(String workflowId, int fromStepId) throws SQLException {
+ record();
+ }
+
+ @Override
+ public void deleteCheckpoints(Collection workflowIds) throws SQLException {
+ record();
+ }
+
+ private void record() throws SQLException {
+ calls.incrementAndGet();
+ if (failure instanceof SQLException e) {
+ throw e;
+ }
+ if (failure instanceof RuntimeException e) {
+ throw e;
+ }
+ }
+ }
+
+ /** The smallest PostgresStepFactory: it runs no steps, and only inherits the default delete. */
+ static class MinimalStepFactory extends PostgresStepFactory {
+ MinimalStepFactory(DBOS dbos, javax.sql.DataSource dataSource, String schema) {
+ this(dbos, dataSource::getConnection, schema);
+ }
+
+ MinimalStepFactory(DBOS dbos, ConnectionOpener opener, String schema) {
+ super(dbos, schema, null, opener);
+ }
+
+ @Override
+ protected Optional checkExecution(String workflowId, int stepId, String stepName) {
+ return Optional.empty();
+ }
+
+ @Override
+ protected void recordError(String workflowId, int stepId, Exception exception) {}
+ }
+}