Skip to content

Commit 92c574f

Browse files
authored
DBOS Instance support (#291)
ensures DBOS.Instance is usable and that all DBOS statics use the global DBOS instance. * DBOS.Instance is publicly constructable. * DBOS.Instance.setConfig removed - config is passed to the ctor * added a DBOS.Instance parameter to DBOSLifecycleListener.launch * updated DBOS internals to use DBOS.Instance instead of DBOS statics * changed WorkflowHandleFuture and WorkflowHandleDBPoll to use an executor instance * Added ClientWorkflowHandle that uses a sys db instance * DBOS.globalInstance is now an AtomicReference, eliminating the need for `syncronized` methods to set it * DBOS.configure sets the global instance if it is not null * DBOS.reinitialize unconditionally sets the global instance. This method is now package private as it is primarily intended for our test usage * DBOS static methods all forward to the global instance. * The global instance is *NOT* created on demand. If it is not set via DBOS.configure (or reinitialize) the static methods fail * Migrations are run on launch instead of when calling DBOS.configure * hard coded internal queue support in queue registry to eliminate need to register internal queue * minor cleanups (registerQueue no longer returns the queue, registerWorkflowMethod no longer returns the method's name) * added JavaDocs to all the DBOS.Instance public methods Note: currently, there's no DBOS.Instance mechanism for something like using DBOS.startWorkflow to start a child workflow. Opened #296 to track fixing this.
1 parent 7b8ee81 commit 92c574f

50 files changed

Lines changed: 1785 additions & 463 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 739 additions & 263 deletions
Large diffs are not rendered by default.

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

Lines changed: 56 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,33 @@
2929
import org.jspecify.annotations.NonNull;
3030
import org.jspecify.annotations.Nullable;
3131

32+
class ClientWorkflowHandle<T, E extends Exception> implements WorkflowHandle<T, E> {
33+
34+
private final SystemDatabase systemDatabase;
35+
private final String workflowId;
36+
37+
public ClientWorkflowHandle(SystemDatabase systemDatabase, String workflowId) {
38+
this.systemDatabase = systemDatabase;
39+
this.workflowId = workflowId;
40+
}
41+
42+
@Override
43+
public String workflowId() {
44+
return workflowId;
45+
}
46+
47+
@Override
48+
public T getResult() throws E {
49+
var result = systemDatabase.<T>awaitWorkflowResult(workflowId);
50+
return Result.<T, E>process(result);
51+
}
52+
53+
@Override
54+
public WorkflowStatus getStatus() {
55+
return systemDatabase.getWorkflowStatus(workflowId);
56+
}
57+
}
58+
3259
/**
3360
* DBOSClient allows external programs to interact with DBOS apps via direct system database access.
3461
* Example interactions: Start/enqueue a workflow, and get the result Get events and send messages
@@ -470,31 +497,36 @@ public EnqueueOptions(
470497
String serializationFormat =
471498
options.serialization() != null ? options.serialization().formatName() : null;
472499

473-
return DBOSExecutor.enqueueWorkflow(
474-
Objects.requireNonNull(
475-
options.workflowName(), "EnqueueOptions workflowName must not be null"),
476-
Objects.requireNonNull(options.className(), "EnqueueOptions className must not be null"),
477-
Objects.requireNonNullElse(options.instanceName(), ""),
478-
null,
479-
args,
480-
new DBOSExecutor.ExecutionOptions(
481-
Objects.requireNonNullElseGet(options.workflowId(), () -> UUID.randomUUID().toString()),
482-
Timeout.of(options.timeout()),
483-
options.deadline,
500+
var workflowId =
501+
DBOSExecutor.enqueueWorkflow(
502+
Objects.requireNonNull(
503+
options.workflowName(), "EnqueueOptions workflowName must not be null"),
484504
Objects.requireNonNull(
485-
options.queueName(), "EnqueueOptions queueName must not be null"),
486-
options.deduplicationId,
487-
options.priority,
488-
options.queuePartitionKey,
489-
false,
490-
false,
491-
serializationFormat),
492-
null,
493-
null,
494-
null,
495-
options.appVersion,
496-
systemDatabase,
497-
this.serializer);
505+
options.className(), "EnqueueOptions className must not be null"),
506+
Objects.requireNonNullElse(options.instanceName(), ""),
507+
null,
508+
args,
509+
new DBOSExecutor.ExecutionOptions(
510+
Objects.requireNonNullElseGet(
511+
options.workflowId(), () -> UUID.randomUUID().toString()),
512+
Timeout.of(options.timeout()),
513+
options.deadline,
514+
Objects.requireNonNull(
515+
options.queueName(), "EnqueueOptions queueName must not be null"),
516+
options.deduplicationId,
517+
options.priority,
518+
options.queuePartitionKey,
519+
false,
520+
false,
521+
serializationFormat),
522+
null,
523+
null,
524+
null,
525+
options.appVersion,
526+
systemDatabase,
527+
this.serializer);
528+
529+
return new ClientWorkflowHandle<>(systemDatabase, workflowId);
498530
}
499531

500532
/**

transact/src/main/java/dev/dbos/transact/context/DBOSContext.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -163,4 +163,9 @@ public static boolean inStep() {
163163
var ctx = DBOSContextHolder.get();
164164
return ctx == null ? false : ctx.isInStep();
165165
}
166+
167+
public static SerializationStrategy serializationStrategy() {
168+
var ctx = DBOSContextHolder.get();
169+
return ctx != null ? ctx.getSerialization() : null;
170+
}
166171
}

transact/src/main/java/dev/dbos/transact/database/WorkflowDAO.java

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -812,7 +812,7 @@ void resumeWorkflow(String workflowId) throws SQLException {
812812
String forkWorkflow(String originalWorkflowId, int startStep, ForkOptions options)
813813
throws SQLException {
814814

815-
Objects.requireNonNull(options);
815+
options = Objects.requireNonNullElseGet(options, ForkOptions::new);
816816

817817
var status = getWorkflowStatus(originalWorkflowId);
818818
if (status == null) {
@@ -897,7 +897,11 @@ private static void insertForkedWorkflowStatus(
897897
stmt.setString(6, applicationVersion);
898898
stmt.setString(7, originalStatus.appId());
899899
stmt.setString(8, originalStatus.authenticatedUser());
900-
stmt.setString(9, JSONUtil.serializeArray(originalStatus.authenticatedRoles()));
900+
stmt.setString(
901+
9,
902+
originalStatus.authenticatedRoles() == null
903+
? null
904+
: JSONUtil.serializeArray(originalStatus.authenticatedRoles()));
901905
stmt.setString(10, originalStatus.assumedRole());
902906
stmt.setString(11, Constants.DBOS_INTERNAL_QUEUE);
903907
stmt.setString(

transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java

Lines changed: 25 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -196,7 +196,7 @@ public ExecutorService get() {
196196
listeners.add(schedulerService);
197197

198198
for (var listener : listeners) {
199-
listener.dbosLaunched();
199+
listener.dbosLaunched(dbos);
200200
}
201201

202202
var recoveryTask =
@@ -650,7 +650,7 @@ public <T, E extends Exception> T runStepInternal(
650650
/** Retrieve the workflowHandle for the workflowId */
651651
public <R, E extends Exception> WorkflowHandle<R, E> retrieveWorkflow(String workflowId) {
652652
logger.debug("retrieveWorkflow {}", workflowId);
653-
return new WorkflowHandleDBPoll<R, E>(workflowId);
653+
return new WorkflowHandleDBPoll<>(this, workflowId);
654654
}
655655

656656
public void sleep(Duration duration) {
@@ -1089,6 +1089,10 @@ public <T, E extends Exception> WorkflowHandle<T, E> startWorkflow(
10891089
logger.debug("startWorkflow {}", options);
10901090

10911091
var invocation = captureInvocation(supplier);
1092+
if (invocation.executor() != this) {
1093+
throw new IllegalStateException(
1094+
"The @Workflow method must be called on the DBOS instance passed to the startWorkflow lambda");
1095+
}
10921096
var workflow = getWorkflow(invocation);
10931097

10941098
var ctx = DBOSContextHolder.get();
@@ -1278,19 +1282,21 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
12781282
"queue %s does not exist".formatted(options.queueName()));
12791283
}
12801284

1281-
return enqueueWorkflow(
1282-
workflow.name(),
1283-
workflow.className(),
1284-
workflow.instanceName(),
1285-
maxRetries,
1286-
args,
1287-
options,
1288-
parent,
1289-
executorId(),
1290-
appVersion(),
1291-
appId(),
1292-
systemDatabase,
1293-
this.serializer);
1285+
var workflowId =
1286+
enqueueWorkflow(
1287+
workflow.name(),
1288+
workflow.className(),
1289+
workflow.instanceName(),
1290+
maxRetries,
1291+
args,
1292+
options,
1293+
parent,
1294+
executorId(),
1295+
appVersion(),
1296+
appId(),
1297+
systemDatabase,
1298+
this.serializer);
1299+
return new WorkflowHandleDBPoll<>(this, workflowId);
12941300
}
12951301

12961302
logger.debug("executeWorkflow {}({}) {}", workflow.fullyQualifiedName(), args, options);
@@ -1413,10 +1419,10 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
14131419
TimeUnit.MILLISECONDS);
14141420
}
14151421

1416-
return new WorkflowHandleFuture<T, E>(workflowId, future, this);
1422+
return new WorkflowHandleFuture<T, E>(this, workflowId, future);
14171423
}
14181424

1419-
public static <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflow(
1425+
public static String enqueueWorkflow(
14201426
String name,
14211427
String className,
14221428
String instanceName,
@@ -1468,10 +1474,10 @@ public static <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflow(
14681474
options.isDequeuedRequest,
14691475
options.serialization(),
14701476
serializer);
1471-
return new WorkflowHandleDBPoll<T, E>(workflowId);
1477+
return workflowId;
14721478
} catch (DBOSWorkflowExecutionConflictException e) {
14731479
logger.debug("Workflow execution conflict for workflowId {}", workflowId);
1474-
return new WorkflowHandleDBPoll<T, E>(workflowId);
1480+
return workflowId;
14751481
} catch (DBOSQueueDuplicatedException e) {
14761482
logger.debug(
14771483
"Workflow queue {} reused deduplicationId {}", e.queueName(), e.deduplicationId());

transact/src/main/java/dev/dbos/transact/execution/DBOSLifecycleListener.java

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,14 @@
11
package dev.dbos.transact.execution;
22

3+
import dev.dbos.transact.DBOS;
4+
35
/**
46
* For registering callbacks that hear about `DBOS.launch()` and `DBOS.shutdown()`. At this point,
57
* DBOS is ready to run workflows, and no additional registrations are allowed.
68
*/
79
public interface DBOSLifecycleListener {
810
/** Called from within DBOS.launch, after workflow processing is allowed */
9-
void dbosLaunched();
11+
void dbosLaunched(DBOS.Instance dbos);
1012

1113
/** Called from within DBOS.shutdown, before workflow processing is stopped */
1214
void dbosShutDown();

0 commit comments

Comments
 (0)