diff --git a/docs/java/examples/_category_.json b/docs/java/examples/_category_.json new file mode 100644 index 000000000..e05223c10 --- /dev/null +++ b/docs/java/examples/_category_.json @@ -0,0 +1,5 @@ +{ + "label": "Example Applications", + "position": 50 +} + \ No newline at end of file diff --git a/docs/java/examples/assets/widget_store_ui.png b/docs/java/examples/assets/widget_store_ui.png new file mode 100644 index 000000000..a7bc74232 Binary files /dev/null and b/docs/java/examples/assets/widget_store_ui.png differ diff --git a/docs/java/examples/widget-store.md b/docs/java/examples/widget-store.md new file mode 100644 index 000000000..f16dc9075 --- /dev/null +++ b/docs/java/examples/widget-store.md @@ -0,0 +1,19 @@ +--- +displayed_sidebar: examplesSidebar +sidebar_position: 30 +title: Fault-Tolerant Checkout +--- + +:::info +This example is also available in [TypeScript](../../typescript/examples/checkout-tutorial) and [Python](../../python/examples/widget-store.md). +::: + +In this example, we use DBOS and Spring Boot to build an online storefront that's resilient to any failure. + +You can see the application live [here](https://demo-widget-store.cloud.dbos.dev/). +Try playing with it and pressing the crash button as often as you want. +Within a few seconds, the app will recover and resume as if nothing happened. + +All source code is [available on GitHub](https://github.com/dbos-inc/dbos-demo-apps/tree/main/java/widget-store). + +![Widget store UI](./assets/widget_store_ui.png) diff --git a/docs/java/integrating-dbos.md b/docs/java/integrating-dbos.md new file mode 100644 index 000000000..81afaae28 --- /dev/null +++ b/docs/java/integrating-dbos.md @@ -0,0 +1,127 @@ +--- +sidebar_position: 20 +title: Add DBOS To Your App +--- + +This guide shows you how to add the open-source [DBOS Transact](https://github.com/dbos-inc/dbos-transact-java) library to your existing application to **durably execute** it and make it resilient to any failure. + +### 1. Install DBOS + +Add DBOS to your application by including it in your build configuration. + +For **Gradle** (build.gradle): +```groovy +dependencies { + implementation 'dev.dbos:transact:0.5.+' +} +``` + +For **Maven** (pom.xml): +```xml + + + dev.dbos + transact + 0.5.+ + + +``` + +### 2. Add the DBOS Initializer + +Add these lines of code to your program's main method. +They configure and launch DBOS when your program starts. + +```java +import dev.dbos.transact.DBOS; +import dev.dbos.transact.config.DBOSConfig; + +public class MyApp { + public static void main(String[] args) throws Exception { + // Configure DBOS + DBOSConfig config = new DBOSConfig.Builder() + .appName("my-app") + .databaseUrl(System.getenv("DBOS_JDBC_URL")) + .dbUser(System.getenv("PGUSER")) + .dbPassword(System.getenv("PGPASSWORD")) + .build(); + DBOS.configure(config); + + // Register your workflows and queues (see step 4) + + // Launch DBOS + DBOS.launch(); + } +} +``` + +:::info +DBOS uses a Postgres database to durably store workflow and step state. +You can connect to your database by setting these environment variables: +- `DBOS_SYSTEM_JDBC_URL`: The JDBC URL for your Postgres database (e.g., `jdbc:postgresql://localhost:5432/mydb`) +- `PGUSER`: Your Postgres username +- `PGPASSWORD`: Your Postgres password + +If you don't have a Postgres database, you can start one locally with Docker: +```shell +docker run -d --name dbos-postgres -e POSTGRES_PASSWORD=mysecretpassword -p 5432:5432 postgres +``` +::: + +### 3. Start Your Application + +Try starting your application. +If everything is set up correctly, your app should run normally, but log `DBOS started` on startup. +Congratulations! You've integrated DBOS into your application. + +### 4. Start Building With DBOS + +At this point, you can add DBOS workflows and steps to your application. +For example, you can annotate one of your methods as a [workflow](./tutorials/workflow-tutorial.md) and the methods it calls as [steps](./tutorials/step-tutorial.md). +DBOS durably executes the workflow so if it is ever interrupted, upon restart it automatically resumes from the last completed step. + +```java +import dev.dbos.transact.workflow.Workflow; +import dev.dbos.transact.workflow.StepOptions; + +interface MyWorkflows { + void reliableWorkflow(); +} + +class MyWorkflowsImpl implements MyWorkflows { + private void stepOne() { + System.out.println("Step one completed!"); + } + + private void stepTwo() { + System.out.println("Step two completed!"); + } + + @Workflow(name = "reliable-workflow") + public void reliableWorkflow() { + DBOS.runStep(() -> stepOne(), "stepOne"); + DBOS.runStep(() -> stepTwo(), "stepTwo"); + } +} +``` + +To use your workflows, create a proxy before launching DBOS: + +```java +// Create a workflow proxy (before launching DBOS) +MyWorkflows workflows = DBOS.registerWorkflows(MyWorkflows.class, new MyWorkflowsImpl()); + +// Launch DBOS +DBOS.launch(); + +// Now you can call your workflows +workflows.reliableWorkflow(); +``` + +**Important:** You must create all workflow proxies and queues before calling `DBOS.launch()`. +Workflow recovery begins after `DBOS.launch()`, so all workflows must be registered before this point. + +You can add DBOS to your application incrementally—it won't interfere with code that's already there. +It's totally okay for your application to have one DBOS workflow alongside thousands of lines of non-DBOS code. + +To learn more about programming with DBOS, check out [the programming guide](./programming-guide.md). \ No newline at end of file diff --git a/docs/java/programming-guide.md b/docs/java/programming-guide.md new file mode 100644 index 000000000..21557a886 --- /dev/null +++ b/docs/java/programming-guide.md @@ -0,0 +1,335 @@ +--- +sidebar_position: 10 +title: Learn DBOS Java +pagination_prev: quickstart +--- + +This guide shows you how to use DBOS to build Java apps that are **resilient to any failure**. + +## 1. Setting Up Your Environment + +First, initialize a new project with Gradle: + +```shell +gradle init --type java-application --dsl groovy --test-framework junit --package com.example --project-name myapp --no-split-project --java-version 17 +``` + +Then, install DBOS by adding the following lines to your `build.gradle` dependencies: + +``` +implementation 'dev.dbos:transact:0.5.+' +implementation 'ch.qos.logback:logback-classic:1.5.18' +``` + +DBOS requires a Postgres database. + +TODO: Instructions on how to set up Postgres. + +## 2. Workflows and Steps + +DBOS helps you add reliability to Java programs. +The key feature of DBOS is **workflow functions** comprised of **steps**. +DBOS automatically provides durability by checkpointing the state of your workflows and steps to its system database. +If your program crashes or is interrupted, DBOS uses this saved state to recover each of your workflows from its last completed step. +Thus, DBOS makes your application **resilient to any failure**. + +Let's create a simple DBOS program that runs a workflow of two steps. +Add the following code to your `App.java` file: + +```java showLineNumbers title="App.java" +package com.example; + +import org.slf4j.LoggerFactory; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.config.DBOSConfig; +import dev.dbos.transact.workflow.StepOptions; +import dev.dbos.transact.workflow.Workflow; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; + +interface Example { + public void workflow(); +} + +class ExampleImpl implements Example { + + private void stepOne() { + System.out.println("Step one completed!"); + } + + private void stepTwo() { + System.out.println("Step two completed!"); + } + + @Workflow(name = "workflow") + public void workflow() { + DBOS.runStep(() -> stepOne(), "stepOne"); + DBOS.runStep(() -> stepTwo(), "stepTwo"); + } +} + +public class App { + public static void main(String[] args) throws Exception { + Logger root = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME); + root.setLevel(Level.INFO); + DBOSConfig config = new DBOSConfig.Builder() + .appName("dbos-java-starter") + .databaseUrl(System.getenv("DBOS_JDBC_URL")) + .dbUser(System.getenv("PGUSER")) + .dbPassword(System.getenv("PGPASSWORD")) + .build(); + DBOS.configure(config); + Example proxy = DBOS.registerWorkflows(Example.class, new ExampleImpl()); + DBOS.launch(); + proxy.workflow(); + DBOS.shutdown(); + } +} +``` + +Now, build and run this code with: + +```shell +gradle assemble +gradle run +``` + +Your program should print output like: + +```shell +Step one completed! +Step two completed! +``` + +To see durable execution in action, let's modify the app to serve a DBOS workflow from an HTTP endpoint using Javalin. +Replace the contents of `App.java` with: + +```java showLineNumbers title="App.java" +package com.example; + +import java.time.Duration; + +import org.slf4j.LoggerFactory; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.config.DBOSConfig; +import dev.dbos.transact.workflow.StepOptions; +import dev.dbos.transact.workflow.Workflow; +import io.javalin.Javalin; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; + +interface Example { + public void workflow() throws InterruptedException; +} + +class ExampleImpl implements Example { + + private void stepOne() { + System.out.println("Step one completed!"); + } + + private void stepTwo() { + System.out.println("Step two completed!"); + } + + @Workflow(name="workflow") + public void workflow() throws InterruptedException { + DBOS.runStep(() -> stepOne(), "stepOne"); + for (int i = 0; i < 5; i++) { + System.out.println("Press Control + C to stop the app..."); + DBOS.sleep(Duration.ofSeconds(1)); + } + DBOS.runStep(() -> stepTwo(), "stepTwo"); + } +} + + +public class App { + public static void main(String[] args) throws Exception { + Logger root = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME); + root.setLevel(Level.INFO); + DBOSConfig config = new DBOSConfig.Builder() + .appName("dbos-java-starter") + .databaseUrl(System.getenv("DBOS_JDBC_URL")) + .dbUser(System.getenv("PGUSER")) + .dbPassword(System.getenv("PGPASSWORD")) + .build(); + DBOS.configure(config); + Example proxy = DBOS.registerWorkflows(Example.class, new ExampleImpl()); + DBOS.launch(); + Javalin.create().get("/", ctx -> { + proxy.workflow(); + ctx.result("Workflow executed!"); + }).start(8080); + } +} +``` + +Now, add the Javalin web server to your dependencies: + +``` +implementation("io.javalin:javalin:6.7.0") +``` + +Then, build and run this code with: + +```shell +gradle assemble +gradle run +``` + +Next, visit this URL: http://localhost:8080. + +In your terminal, you should see an output like: + +``` +12:45:26.519 [main] INFO io.javalin.Javalin -- Listening on http://localhost:8080/ +Step one completed! +Press Control + C to stop the app... +Press Control + C to stop the app... +Press Control + C to stop the app... +``` + +Now, press CTRL+C stop your app. Then, run `gradle run` to restart it. You should see an output like: + +``` +12:45:37.794 [main] INFO io.javalin.Javalin -- Listening on http://localhost:8080/ +Press Control + C to stop the app... +Press Control + C to stop the app... +Press Control + C to stop the app... +Press Control + C to stop the app... +Step two completed! +``` + +You can see how DBOS **recovers your workflow from the last completed step**, executing step two without re-executing step one. +Learn more about workflows, steps, and their guarantees [here](./tutorials/workflow-tutorial.md). + +## 3. Queues and Parallelism + +To run many functions concurrently, use DBOS _queues_. +To try them out, copy this code into `App.java`: + +```java showLineNumbers title="App.java" +package com.example; + +import java.util.ArrayList; +import java.util.List; + +import org.slf4j.LoggerFactory; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.StartWorkflowOptions; +import dev.dbos.transact.config.DBOSConfig; +import dev.dbos.transact.queue.Queue; +import dev.dbos.transact.workflow.Workflow; +import dev.dbos.transact.workflow.WorkflowHandle; +import io.javalin.Javalin; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; + +interface Example { + public void setProxy(Example proxy); + + public void taskWorkflow(int i) throws InterruptedException; + + public void queueWorkflow() throws InterruptedException; +} + +class ExampleImpl implements Example { + + private final Queue queue; + private Example proxy; + + public ExampleImpl(Queue queue) { + this.queue = queue; + } + + public void setProxy(Example proxy) { + this.proxy = proxy; + } + + @Workflow(name = "task-workflow") + public void taskWorkflow(int i) throws InterruptedException { + Thread.sleep(5000); + System.out.printf("Task %d completed!\n", i); + } + + @Workflow(name = "queue-workflow") + public void queueWorkflow() throws InterruptedException { + List> handles = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + final int index = i; + WorkflowHandle handle = DBOS.startWorkflow( + () -> this.proxy.taskWorkflow(index), + new StartWorkflowOptions().withQueue(this.queue)); + handles.add(handle); + } + for (WorkflowHandle handle : handles) { + handle.getResult(); + } + System.out.printf("Successfully completed %d workflows!", handles.size()); + } +} + +public class App { + public static void main(String[] args) throws Exception { + Logger root = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME); + root.setLevel(Level.INFO); + DBOSConfig config = new DBOSConfig.Builder() + .appName("dbos-java-starter") + .databaseUrl(System.getenv("DBOS_JDBC_URL")) + .dbUser(System.getenv("PGUSER")) + .dbPassword(System.getenv("PGPASSWORD")) + .build(); + DBOS.configure(config); + Queue queue = DBOS.Queue("example-queue").build(); + ExampleImpl impl = new ExampleImpl(queue); + Example proxy = DBOS.registerWorkflows(Example.class, impl); + impl.setProxy(proxy); + DBOS.launch(); + Javalin.create().get("/", ctx -> { + proxy.queueWorkflow(); + ctx.result("Workflow executed!"); + }).start(8080); + } +} +``` + +When you enqueue a function by passing `new StartWorkflowOptions().withQueue(this.queue)` into `DBOS.startWorkflow`, DBOS executes it _asynchronously_, running it in the background without waiting for it to finish. +`DBOS.startWorkflow` returns a handle representing the state of the enqueued function. +This example enqueues ten functions, then waits for them all to finish using `.getResult()` to wait for each of their handles. + +Now, restart your app with: + +```shell +gradle assemble +gradle run +``` + +Then, visit this URL: http://localhost:8080. +Wait five seconds and you should see an output like: + +``` +Task 0 completed! +Task 1 completed! +Task 2 completed! +Task 3 completed! +Task 4 completed! +Task 5 completed! +Task 6 completed! +Task 7 completed! +Task 8 completed! +Task 9 completed! +Successfully completed 10 workflows! +``` + +You can see how all ten steps run concurrently—even though each takes five seconds, they all finish at the same time. +Learn more about DBOS queues [here](./tutorials/queue-tutorial.md). + +Congratulations! You've finished the DBOS Java guide. +You can find the code from this guide in the [DBOS Toolbox](https://github.com/dbos-inc/dbos-demo-apps/tree/main/java/dbos-toolbox) template app. + +Next, to learn how to build more complex applications, check out the Java tutorials and example apps. \ No newline at end of file diff --git a/docs/java/reference/_category_.json b/docs/java/reference/_category_.json new file mode 100644 index 000000000..ebb337b83 --- /dev/null +++ b/docs/java/reference/_category_.json @@ -0,0 +1,4 @@ +{ + "label": "Reference", + "position": 40 +} diff --git a/docs/java/reference/client.md b/docs/java/reference/client.md new file mode 100644 index 000000000..8720c9b6f --- /dev/null +++ b/docs/java/reference/client.md @@ -0,0 +1,157 @@ +--- +sidebar_position: 50 +title: DBOS Client +--- + +`DBOSClient` provides a programmatic way to interact with your DBOS application from external code. + +## DBOSClient + +```java +DBOSClient(String url, String user, String password) +``` + +Construct the DBOSClient. + +**Parameters:** +- **url**: The JDBC URL for your system database. +- **user**: Your Postgres username or role. +- **password**: The password for your Postgres user or role. + +## Workflow Interaction Methods + +### enqueueWorkflow + +```java + WorkflowHandle enqueueWorkflow( + EnqueueOptions options, Object[] args) +``` + +Enqueue a workflow and return a handle to it. + +**Parameters:** +- **options**: Configuration for the enqueued workflow, as defined below. +- **args**: An array of the workflow's arguments. These will be serialized and passed into the workflow when it is dequeued. + +**Example Syntax:** + +This code enqueues workflow `exampleWorkflow` in class `com.example.ExampleImpl` on queue `example-queue` with arguments `argumentOne` and `argumentTwo`. + +```java +var client = new DBOSClient(dbUrl, dbUser, dbPassword); +var options = + new DBOSClient.EnqueueOptions( + "com.example.ExampleImpl", "exampleWorkflow", "example-queue"); +var handle = client.enqueueWorkflow(options, new Object[]{"argumentOne", "argumentTwo"}); +``` + +**Options:** + +```java +public record EnqueueOptions( + String workflowName, + String queueName, + String className, + String instanceName, + String workflowId, + String appVersion, + Duration timeout, + String deduplicationId, + OptionalInt priority +) +``` + +**Constructors:** + +```java +public EnqueueOptions(String className, String workflowName, String queueName) +``` + +Specify the name and class name of the workflow to enqueue and the name of the queue on which it is to be enqueued. + +**Methods:** + +- **`withAppVersion(String appVersion)`**: The version of your application that should process this workflow. +If left undefined, it will be updated to the current version when the workflow is first dequeued. +Please see [Managing Application Versions](../../production/self-hosting/workflow-recovery#managing-application-versions) for more information. +- **`withTimeout(Duration timeout)`**: Set a timeout for the enqueued workflow. When the timeout expires, the workflow and all its children are cancelled. The timeout does not begin until the workflow is dequeued and starts execution. +- **`withDeduplicationId(String deduplicationId)`**: At any given time, only one workflow with a specific deduplication ID can be enqueued in the specified queue. If a workflow with a deduplication ID is currently enqueued or actively executing (status `ENQUEUED` or `PENDING`), subsequent workflow enqueue attempt with the same deduplication ID in the same queue will raise an exception. +- **`withPriority(int priority)`**: The priority of the enqueued workflow in the specified queue. Workflows with the same priority are dequeued in FIFO (first in, first out) order. Priority values can range from `1` to `2,147,483,647`, where a low number indicates a higher priority. Workflows without assigned priorities have the highest priority and are dequeued before workflows with assigned priorities. +- **`withInstanceName`**: The enqueued workflow should run on this particular named class instance. + + + +### send + +```java +send(String destinationId, Object message, String topic, String idempotencyKey) +``` + +Similar to [`DBOS.send`](./methods.md#send). + +### getEvent + +```java +Object getEvent(String targetId, String key, double timeoutSeconds) +``` + +Similar to [`DBOS.getEvent`](./methods.md#getevent). + +## Workflow Management Methods + +### retrieveWorkflow + +```java +WorkflowHandle retrieveWorkflow(String workflowId) +``` + +Similar to [`DBOS.retrieveWorkflow`](./methods.md#retrieveworkflow). + +### getWorkflowStatus + +```java +Optional getWorkflowStatus(String workflowId) +``` + +Retrieve the [`WorkflowStatus`](./methods.md#workflowstatus) of a workflow. + +### listWorkflows + +```java +List listWorkflows(ListWorkflowsInput input) +``` + +Similar to [`DBOS.listWorkflows`](./methods.md#listworkflows). + +### listWorkflowSteps + +```java +List listWorkflowSteps(String workflowId) +``` + +Similar to [`DBOS.listWorkflowSteps`](./methods.md#listworkflowsteps). + +### cancelWorkflow + +```java +void cancelWorkflow(String workflowId) +``` + +Similar to [`DBOS.cancelWorkflow`](./methods.md#cancelworkflow). + +### resumeWorkflow + +```java + WorkflowHandle resumeWorkflow(String workflowId) +``` + +Similar to [`DBOS.resumeWorkflow`](./methods.md#resumeworkflow). + +### forkWorkflow + +```java + WorkflowHandle forkWorkflow( + String originalWorkflowId, int startStep, ForkOptions options) +``` + +Similar to [`DBOS.forkWorkflow`](./methods.md#forkworkflow). diff --git a/docs/java/reference/lifecycle.md b/docs/java/reference/lifecycle.md new file mode 100644 index 000000000..795864132 --- /dev/null +++ b/docs/java/reference/lifecycle.md @@ -0,0 +1,85 @@ +--- +sidebar_position: 10 +title: DBOS Lifecycle +--- + +The DBOS class is a singleton—you should configure and launch it exactly once in a program's lifetime. +You manage and access it through static methods (e.g., `DBOS.configure()`, `DBOS.launch()`). +Here, we document configuration and lifecycle methods. + +### DBOS.configure + +```java +static synchronized Instance configure(DBOSConfig config) +``` + +Configure the DBOS singleton. + +**Options:** + +```java +public record DBOSConfig( + String appName, + String databaseUrl, + String dbUser, + String dbPassword, + int maximumPoolSize, + int connectionTimeout, + boolean adminServer, + int adminServerPort, + String conductorKey, + String appVersion +) +``` + +**Constructors:** + +```java +new DBOSConfig.Builder() +``` + +Create a DBOSConfig object. + +**Methods:** + +- **`appName(String appName)`**: Your application's name. Required. + +- **`databaseUrl(String databaseUrl)`**: The JDBC URL for your system database. Required. A valid JDBC URL is of the form `jdbc:postgresql://host:port/database` + +- **`dbUser(String dbUser)`**: Your Postgres username or role. Required. + +- **`dbPassword(String dbPassword)`**: The password for your Postgres user or role. Required. + +- **`maximumPoolSize(int maximumPoolSize)`**: The maximum size for the system database connection pool created by DBOS. + +- **`connectionTimeout(int connectionTimeout)`**: The connection timeout for the system database connection created by DBOS. + +- **`runAdminServer()`**: Whether to run an [HTTP admin server](../../production/self-hosting/admin-api.md) for workflow management operations. Defaults to true. + +- **`adminServerPort(int port)`**: The port on which the admin server runs. Defaults to 3001. + +- **`conductorKey(String key)`**: An API key for [DBOS Conductor](../../production/self-hosting/conductor.md). If provided, application is connected to Conductor. API keys can be created from the [DBOS console](https://console.dbos.dev). + +- **`appVersion(String appVersion)`**: The code version for this application and its workflows. Workflow versioning is documented [here](../tutorials/workflow-tutorial.md#workflow-versioning-and-recovery). + +- **`build()`**: Build the configuration object. Must be called after all parameters are set. + +### DBOS.launch + +```java +static void launch() +``` + +Launch DBOS, initializing database connections and beginning workflow recovery and queue processing. +This should be called after all workflows and queues are registered. +**You should not call a DBOS workflow until after DBOS is launched.** + +### DBOS.shutdown + +```java +static void shutdown() +``` + +Destroy the DBOS singleton. +After DBOS is shut down, a new singleton can be configured and launched. +This may be useful for testing DBOS applications. \ No newline at end of file diff --git a/docs/java/reference/methods.md b/docs/java/reference/methods.md new file mode 100644 index 000000000..5c86ba128 --- /dev/null +++ b/docs/java/reference/methods.md @@ -0,0 +1,353 @@ +--- +sidebar_position: 30 +title: DBOS Methods & Variables +toc_max_heading_level: 3 +--- + +## DBOS Methods + +### getEvent + +```java +static Object getEvent(String workflowId, String key, Duration timeout) +``` + +Retrieve the latest value of an event published by the workflow identified by `workflowId` to the key `key`. +If the event does not yet exist, wait for it to be published, an error if the wait times out. + +**Parameters:** +- **workflowId**: The identifier of the workflow whose events to retrieve. +- **key**: The key of the event to retrieve. +- **timeout**: A timeout duration. If the wait times out, return null. + +### setEvent + +```java +static void setEvent(String key, Object value) +``` +Create and associate with this workflow an event with key `key` and value `value`. +If the event already exists, update its value. +Can only be called from within a workflow. + +**Parameters:** +- **key**: The key of the event. +- **message**: The value of the event. Must be serializable. + +### send + +```java +static void send(String destinationId, Object message, String topic) +``` + +Send a message to the workflow identified by `destinationID`. +Messages can optionally be associated with a topic. + +**Parameters:** +- **destinationId**: The workflow to which to send the message. +- **message**: The message to send. Must be serializable. +- **topic**: A topic with which to associate the message. Messages are enqueued per-topic on the receiver. + +### recv + +```java +static Object recv(String topic, Duration timeout) +``` + +Receive and return a message sent to this workflow. +Can only be called from within a workflow. +Messages are dequeued first-in, first-out from a queue associated with the topic. +Calls to `recv` wait for the next message in the queue, returning an error if the wait times out. + +**Parameters:** +- **topic**: A topic queue on which to wait. +- **timeout**: A timeout duration. If the wait times out, return null. + +### sleep + +```java +static void sleep(Duration sleepduration) +``` + +Sleep for the given duration. +If called from within a workflow, this sleep is durable—it records its intended wake-up time in the database so if it is interrupted and recovers, it still wakes up at the intended time. +If called from outside a workflow, or from within a step, it behaves like a regular sleep. + +**Parameters:** +- **sleepduration**: The duration to sleep. + +### retrieveWorkflow + +```go +static WorkflowHandle retrieveWorkflow(String workflowId) +``` + +Retrieve the [handle](./workflows-steps.md#workflowhandle) of a workflow. + +**Parameters**: +- **workflowId**: The ID of the workflow whose handle to retrieve. + +## Workflow Management Methods + +### listWorkflows + +```java +static List listWorkflows(ListWorkflowsInput input) +``` + +Retrieve a list of [`WorkflowStatus`](#workflowstatus) of all workflows matching specified criteria. + +#### ListWorkflowsInput + +`ListWorkflowsInput` is a builder-based configuration record for filtering and customizing workflow queries. All fields are optional. + +**Builder Methods:** + +##### workflowId +```java +Builder workflowId(String workflowId) +``` +Add a workflow ID to filter by. + +##### workflowIds +```java +Builder workflowIds(List workflowIDs) +``` +Add multiple workflow IDs to filter by. + +##### className +```java +Builder className(String className) +``` +Filter workflows by the class name containing the workflow function. + +##### instanceName +```java +Builder instanceName(String instanceName) +``` +Filter workflows by the instance name of the class. + +##### workflowName +```java +Builder workflowName(String workflowName) +``` +Filter workflows by the workflow function name. + +##### authenticatedUser +```java +Builder authenticatedUser(String authenticatedUser) +``` +Filter workflows run by this authenticated user. + +##### startTime +```java +Builder startTime(OffsetDateTime startTime) +``` +Retrieve workflows started after this timestamp. + +##### endTime +```java +Builder endTime(OffsetDateTime endTime) +``` +Retrieve workflows started before this timestamp. + +##### status +```java +Builder status(WorkflowState status) +Builder status(String status) +``` +Filter workflows by status. Status must be one of: `ENQUEUED`, `PENDING`, `SUCCESS`, `ERROR`, `CANCELLED`, or `MAX_RECOVERY_ATTEMPTS_EXCEEDED`. + +##### applicationVersion +```java +Builder applicationVersion(String applicationVersion) +``` +Retrieve workflows tagged with this application version. + +##### limit +```java +Builder limit(Integer limit) +``` +Retrieve up to this many workflows. + +##### offset +```java +Builder offset(Integer offset) +``` +Skip this many workflows from the results returned (for pagination). + +##### sortDesc +```java +Builder sortDesc(Boolean sortDesc) +``` +Sort the results in descending (true) or ascending (false) order by workflow start time. + +##### executorId +```java +Builder executorId(String executorId) +``` +Retrieve workflows that ran on this executor process. + +##### queueName +```java +Builder queueName(String queueName) +``` +Retrieve workflows that were enqueued on this queue. + +##### workflowIdPrefix +```java +Builder workflowIdPrefix(String workflowIdPrefix) +``` +Filter workflows whose IDs start with the specified prefix. + +##### workflowIdPrefix +```java +Builder workflowIdPrefix(String workflowIdPrefix) +``` +Filter workflows whose IDs start with the specified prefix. + +##### queuedOnly +```java +Builder queuedOnly(Boolean queuedOnly) +``` +Select only workflows that are currently enqueued (status `PENDING` or `ENQUEUED` and on a queue). + +##### loadInput +```java +Builder loadInput(Boolean value) +``` +Controls whether to load workflow input data (default: true). + +##### loadOutput +```java +Builder loadOutput(Boolean value) +``` +Controls whether to load workflow output data (default: true). + + +### listWorkflowSteps + +```java +static List listWorkflowSteps(String workflowId) +``` + +Retrieve the execution steps of a workflow. +This is a list of `StepInfo` objects, with the following structure: + +```java +StepInfo( + // The sequential ID of the step within the workflow + int functionId, + // The name of the step function + String functionName, + // The output returned by the step, if any + Object output, + // The error returned by the step, if any + ErrorResult error, + // If the step starts or retrieves the result of a workflow, its ID + String childWorkflowId +) +``` + +### cancelWorkflow + +```java +static cancelWorkflow(String workflowId) +``` + +Cancel a workflow. This sets its status to `CANCELLED`, removes it from its queue (if it is enqueued) and preempts its execution (interrupting it at the beginning of its next step). + +### resumeWorkflow + +```java +static WorkflowHandle resumeWorkflow(String workflowId) +``` + +Resume a workflow. This immediately starts it from its last completed step. You can use this to resume workflows that are cancelled or have exceeded their maximum recovery attempts. You can also use this to start an enqueued workflow immediately, bypassing its queue. + +### forkWorkflow + +```java +static WorkflowHandle forkWorkflow( + String workflowId, + int startStep, + ForkOptions options +) +``` + +```java +public record ForkOptions( + String forkedWorkflowId, + String applicationVersion, + Duration timeout +) +``` + +Start a new execution of a workflow from a specific step. The input step ID (`startStep`) must match the step number of the step returned by workflow introspection. The specified `startStep` is the step from which the new workflow will start, so any steps whose ID is less than `startStep` will not be re-executed. + +**Parameters:** +- **workflowId**: The ID of the workflow to fork +- **startStep**: The step from which to fork the workflow +- **options**: + - **forkedWorkflowId**: The workflow ID for the newly forked workflow (if not provided, generate a UUID) + - **applicationVersion**: The application version for the forked workflow (inherited from the original if not provided) + - **timeout**: A timeout for the forked workflow. + +### WorkflowStatus + +Some workflow introspection and management methods return a `WorkflowStatus`. +This object has the following definition: + +```java +public record WorkflowStatus( + // The workflow ID + String workflowId, + // The workflow status. Must be one of ENQUEUED, PENDING, SUCCESS, ERROR, CANCELLED, or MAX_RECOVERY_ATTEMPTS_EXCEEDED + String status, + // The name of the workflow function + String name, + // The class of the workflow function + String className, + // The name given to the class instance, if any + String instanceName, + // The deserialized workflow input object + Object[] input, + // The workflow's output, if any + Object output, + // The error the workflow threw, if any + ErrorResult error, + // Workflow start time, as a Unix epoch timestamp in ms + Long createdAt, + // The last time the workflow status was updated, as a Unix epoch timestamp in ms + Long updatedAt, + // If this workflow was enqueued, on which queue + String queueName, + // The ID of the executor (process) that most recently executed this workflow + String executorId, + // The application version on which this workflow was started + String appVersion, + // The workflow timeout, if any + Long workflowTimeoutMs, + // The Unix epoch timestamp at which this workflow will time out, if any + Long workflowDeadlineEpochMs, + // The number of times this workflow has been started + Integer recoveryAttempts +) +``` + +## DBOS Variables + +### workflowId + +```java +static String workflowId() +``` + +Retrieve the ID of the current workflow. Returns `null` if not called from a workflow or step. + +### stepId + +```java +static Integer stepId() +``` + +Returns the unique ID of the current step within its workflow. Returns `null` if not called from a step. diff --git a/docs/java/reference/queues.md b/docs/java/reference/queues.md new file mode 100644 index 000000000..318a317eb --- /dev/null +++ b/docs/java/reference/queues.md @@ -0,0 +1,44 @@ +--- +sidebar_position: 40 +title: Queues +--- + +Workflow queues allow you to ensure that workflow functions will be run, without starting them immediately. +Queues are useful for controlling the number of workflows run in parallel, or the rate at which they are started. + +All queues should be created before DBOS is launched. + +### Queue + +```java +QueueBuilder Queue(String name) +``` + +```java +public class QueueBuilder { + public QueueBuilder concurrency(Integer concurrency) + public QueueBuilder workerConcurrency(Integer workerConcurrency) + public QueueBuilder limit(int limit, double period) + public QueueBuilder priorityEnabled(boolean priorityEnabled) + public Queue build(); +} +``` + +Create a new workflow queue with the specified name and optional configuration parameters. +Queues must be created before DBOS is launched. +You can enqueue a workflow using the `withQueue` parameter of [`startWorkflow`](./workflows-steps.md#startworkflow). + +**Parameters:** +- **name**: The name of the queue. Must be unique among all queues in the application. +- **workerConcurrency**: The maximum number of workflows from this queue that may run concurrently within a single DBOS process. +- **concurrency**: The maximum number of workflows from this queue that may run concurrently. This concurrency limit is global across all DBOS processes using this queue. +- **limit**: A limit on the maximum number of functions (`limit`) that may be started in a given period (`period`). +- **priorityEnabled**: Enable setting priority for workflows on this queue. + +**Example Syntax:** + +```java +Queue queue = DBOS.Queue("example-queue") + .workerConcurrency(1) + .build(); +``` \ No newline at end of file diff --git a/docs/java/reference/workflows-steps.md b/docs/java/reference/workflows-steps.md new file mode 100644 index 000000000..bc628d5c9 --- /dev/null +++ b/docs/java/reference/workflows-steps.md @@ -0,0 +1,232 @@ +--- +sidebar_position: 20 +title: Workflows & Steps +toc_max_heading_level: 3 +--- + +## Annotations + +### @Workflow + +```java +public @interface Workflow { + String name(); + + int maxRecoveryAttempts(); +} +``` + +An annotation that can be applied to a class method to mark it as a durable workflow. + +**Parameters:** +- **name**: The workflow name. Must be unique. +- **maxRecoveryAttempts**: Optionally configure the maximum number of times execution of a workflow may be attempted. +This acts as a [dead letter queue](https://en.wikipedia.org/wiki/Dead_letter_queue) so that a buggy workflow that crashes its application (for example, by running it out of memory) does not do so infinitely. +If a workflow exceeds this limit, its status is set to `MAX_RECOVERY_ATTEMPTS_EXCEEDED` and it may no longer be executed. + +## Methods + +### registerWorkflows + +```java +static T registerWorkflows(Class interfaceClass, T implementation) +static T registerWorkflows(Class interfaceClass, T implementation, String instanceName) +``` + +Proxy a class whose methods contain `@Workflow` annotations. +Methods can then be invoked as durable workflows using the proxy. +All proxies must be created before DBOS is launched. + +**Example Syntax:** + +```java +interface Example { + public void workflow(); +} + +class ExampleImpl implements Example { + @Workflow(name="workflow") + public void workflow() { + return; + } +} + +Example proxy = DBOS.registerWorkflows(Example.class, new ExampleImpl()); +proxy.workflow(); +``` + +**Parameters:** +- **interfaceClass**: The interface class to be proxied. +- **implementation**: An instance of the class to proxy. +- **instanceName**: A unique name for this class instance. Use only when you are creating multiple instances of a class and your workflow depends on class instance variables. When DBOS needs to recover a workflow belonging to that class, it looks up the class instance using `instanceName` so it can recover the workflow using the right instance of its class. + + +### startWorkflow + +```java +static WorkflowHandle startWorkflow( + ThrowingSupplier workflow, + StartWorkflowOptions options +) +``` + +Start a workflow in the background and return a handle to it. +Optionally enqueue it on a DBOS queue. +The `startWorkflow` method resolves after the workflow is durably started; at this point the workflow is guaranteed to run to completion even if the app is interrupted. + +**Example Syntax**: + +```java +interface Example { + public void workflow(); +} + +class ExampleImpl implements Example { + @Workflow(name="workflow") + public void workflow() { + return; + } +} + +Example proxy = DBOS.registerWorkflows(Example.class, new ExampleImpl()); +DBOS.startWorkflow(() -> proxy.workflow(), new StartWorkflowOptions()); +``` + +**Options:** + +```java +public record StartWorkflowOptions( + String workflowId, + Duration timeout, + String queueName, + String deduplicationId, + OptionalInt priority +) +``` + +**Constructors:** +```java +new StartWorkflowOptions() +``` +Create workflow options with all fields set to their defaults. + +**Methods:** +- **`withWorkflowId(String workflowId)`** - Set the workflow ID of this workflow. + +- **`withTimeout(Duration timeout)`** / **`withTimeout(long value, TimeUnit unit)`** - Set a timeout for this workflow. When the timeout expires, the workflow **and all its children** are cancelled. Cancelling a workflow sets its status to `CANCELLED` and preempts its execution at the beginning of its next step. + + Timeouts are **start-to-completion**: if a workflow is enqueued, the timeout does not begin until the workflow is dequeued and starts execution. Also, timeouts are **durable**: they are stored in the database and persist across restarts, so workflows can have very long timeouts. + + Timeout deadlines are propagated to child workflows by default, so when a workflow's deadline expires all of its child workflows (and their children, and so on) are also cancelled. If you want to detach a child workflow from its parent's timeout, you can start it with `SetWorkflowTimeout(custom_timeout)` to override the propagated timeout. You can use `SetWorkflowTimeout(None)` to start a child workflow with no timeout. + +- **`withQueue(Queue queue)`** - Instead of starting the workflow directly, enqueue it on this queue. + +- **`withDeduplicationId(String deduplicationId)`** - May only be used when enqueueing. At any given time, only one workflow with a specific deduplication ID can be enqueued in the specified queue. If a workflow with a deduplication ID is currently enqueued or actively executing (status `ENQUEUED` or `PENDING`), subsequent workflow enqueue attempts with the same deduplication ID in the same queue will raise an exception. + +- **`withPriority(int priority)`** - May only be used when enqueueing. The priority of the enqueued workflow in the specified queue. Workflows with the same priority are dequeued in FIFO (first in, first out) order. Priority values can range from `1` to `2,147,483,647`, where a low number indicates a higher priority. Workflows without assigned priorities have the highest priority and are dequeued before workflows with assigned priorities. + +### runStep + +```java +static T runStep( + ThrowingSupplier stepfunc, + StepOptions opts +) throws E + +static T runStep( + ThrowingSupplier stepfunc, + String stepName +) throws E +``` + +Run a function as a step. If called from within a workflow, the result is durably stored. +Returns the output of the step. + +**Example Syntax:** + +```java +class ExampleImpl implements Example { + + private void stepOne() { + System.out.println("Step one completed!"); + } + + private void stepTwo() { + System.out.println("Step two completed!"); + } + + @Workflow(name="workflow") + public void workflow() throws InterruptedException { + DBOS.runStep(() -> stepOne(), "stepOne"); + DBOS.runStep(() -> stepTwo(), new StepOptions("stepTwo").withRetriesAllowed(false)); + } +} +``` + +**Options:** + +```java +public record StepOptions( + String name, + boolean retriesAllowed, + int maxAttempts, + double intervalSeconds, + double backOffRate +) +``` + +**Constructors:** +```java +new StepOptions(String name) +``` +Create step options and provide a name for this step. + +**Methods:** +- **`withRetriesAllowed(boolean b)`** - Whether to retry the step if it throws an exception. Defaults to false. + +- **`withMaxAttempts(int n)`** - How many times to retry a step that is throwing exceptions. + +- **`withIntervalSeconds(double t)`** - How long to wait before the initial retry. + +- **`withBackoffRate(double t)`** - How much to multiplicatively increase `intervalSeconds` between retries. + +### WorkflowHandle + +```java +public interface WorkflowHandle { + + String getWorkflowId(); + + T getResult() throws E; + + WorkflowStatus getStatus(); +} +``` + +WorkflowHandle provides methods to interact with a running or completed workflow. +The type parameters `T` and `E` represents the expected return type of the workflow and the checked exceptions it may throw. +Handles can be used to wait for workflow completion, check status, and retrieve results. + +#### WorkflowHandle.getResult + +```java +T getResult() throws E; +``` + +Wait for the workflow to complete and return its result. + +#### WorkflowHandle.getStatus + +```java +WorkflowStatus getStatus(); +``` + +Retrieve the WorkflowStatus of the workflow. + +#### WorkflowHandle.getWorkflowId + +```java +String getWorkflowId(); +``` + +Retrieve the ID of the workflow. \ No newline at end of file diff --git a/docs/java/tutorials/_category_.json b/docs/java/tutorials/_category_.json new file mode 100644 index 000000000..dbc40b5e1 --- /dev/null +++ b/docs/java/tutorials/_category_.json @@ -0,0 +1,4 @@ +{ + "label": "Tutorials", + "position": 30 +} diff --git a/docs/java/tutorials/queue-tutorial.md b/docs/java/tutorials/queue-tutorial.md new file mode 100644 index 000000000..a08a0875c --- /dev/null +++ b/docs/java/tutorials/queue-tutorial.md @@ -0,0 +1,261 @@ +--- +sidebar_position: 30 +title: Queues & Concurrency +toc_max_heading_level: 3 +--- + +You can use queues to run many workflows at once with managed concurrency. +Queues provide _flow control_, letting you manage how many workflows run at once or how often workflows are started. + +To create a queue, use [`Queue`](../reference/queues.md#queue). +All queues should be created before DBOS is launched. + +```java +Queue queue = DBOS.Queue("example-queue").build(); +``` + +You can then enqueue any workflow using [`withQueue`](../reference/workflows-steps.md#startworkflow) when calling `startWorkflow`. +Enqueuing a workflow submits it for execution and returns a [handle](../reference/workflows-steps.md#workflowhandle) to it. +Queued tasks are started in first-in, first-out (FIFO) order. + +```java +class ExampleImpl implements Example { + @Workflow(name = "processTask") + public String processTask(String task) { + // Process the task... + return "Processed: " + task; + } +} + +public String example(Queue queue, Example proxy, String task) throws Exception { + // Enqueue a workflow + WorkflowHandle handle = DBOS.startWorkflow( + () -> proxy.processTask(task), + new StartWorkflowOptions().withQueue(queue) + ); + + // Get the result + String result = handle.getResult(); + System.out.println("Task result: " + result); + return result; +} +``` + +## Queue Example + +Here's an example of a workflow using a queue to process tasks concurrently: + +```java +interface Example { + public void setProxy(Example proxy); + public String taskWorkflow(String task); + public List queueWorkflow(String[] tasks) throws Exception; +} + +class ExampleImpl implements Example { + + private final Queue queue; + private Example proxy; + + public ExampleImpl(Queue queue) { + this.queue = queue; + } + + public void setProxy(Example proxy) { + this.proxy = proxy; + } + + @Workflow(name = "taskWorkflow") + public String taskWorkflow(String task) { + // Process the task... + return "Processed: " + task; + } + + @Workflow(name = "queueWorkflow") + public List queueWorkflow(String[] tasks) throws Exception { + // Enqueue each task so all tasks are processed concurrently + List> handles = new ArrayList<>(); + for (String task : tasks) { + WorkflowHandle handle = DBOS.startWorkflow( + () -> proxy.taskWorkflow(task), + new StartWorkflowOptions().withQueue(queue) + ); + handles.add(handle); + } + + // Wait for each task to complete and retrieve its result + List results = new ArrayList<>(); + for (WorkflowHandle handle : handles) { + String result = handle.getResult(); + results.add(result); + } + + return results; + } +} + +public class App { + public static void main(String[] args) throws Exception { + DBOSConfig config = ... + DBOS.configure(config); + + Queue queue = DBOS.Queue("example-queue").build(); + ExampleImpl impl = new ExampleImpl(queue); + Example proxy = dbos.registerWorkflows(Example.class, impl); + impl.setProxy(proxy); + + DBOS.launch(); + + // Run the queue workflow + String[] tasks = {"task1", "task2", "task3", "task4", "task5"}; + List results = proxy.queueWorkflow(tasks); + for (String result : results) { + System.out.println(result); + } + } +} +``` + +## Enqueueing from Another Application + +Often, you want to enqueue a workflow from outside your DBOS application. +For example, let's say you have an API server and a data processing service. +You're using DBOS to build a durable data pipeline in the data processing service. +When the API server receives a request, it should enqueue the data pipeline for execution on the data processing service. + +You can use the [DBOS Client](../reference/client.md) to enqueue workflows from outside your DBOS application by connecting directly to your DBOS application's system database. +Since the DBOS Client is designed to be used from outside your DBOS application, workflow and queue metadata must be specified explicitly. + +For example, this code enqueues the `dataPipeline` workflow on the `pipelineQueue` queue with arguments: + +```java +var client = new DBOSClient(dbUrl, dbUser, dbPassword); + +var options = new DBOSClient.EnqueueOptions( + "com.example.DataPipelineImpl", // Class name + "dataPipeline", // Workflow name + "pipelineQueue" // Queue name +); + +var handle = client.enqueueWorkflow( + options, + new Object[]{"task-123", "data"} // Workflow arguments +); + +// Optionally wait for the result +Object result = handle.getResult(); +``` + +## Managing Concurrency + +You can control how many workflows from a queue run simultaneously by configuring concurrency limits. +This helps prevent resource exhaustion when workflows consume significant memory or processing power. + +### Worker Concurrency + +Worker concurrency sets the maximum number of workflows from a queue that can run concurrently on a single DBOS process. +This is particularly useful for resource-intensive workflows to avoid exhausting the resources of any process. +For example, this queue has a worker concurrency of 5, so each process will run at most 5 workflows from this queue simultaneously: + +```java +Queue queue = DBOS.Queue("example-queue") + .workerConcurrency(5) + .build(); +``` + +### Global Concurrency + +Global concurrency limits the total number of workflows from a queue that can run concurrently across all DBOS processes in your application. +For example, this queue will have a maximum of 10 workflows running simultaneously across your entire application. + +:::warning +Worker concurrency limits are recommended for most use cases. +Take care when using a global concurrency limit as any `PENDING` workflow on the queue counts toward the limit, including workflows from previous application versions. +::: + +```java +Queue queue = DBOS.Queue("example-queue") + .concurrency(10) + .build(); +``` + +## Rate Limiting + +You can set _rate limits_ for a queue, limiting the number of workflows that it can start in a given period. +Rate limits are global across all DBOS processes using this queue. +For example, this queue has a limit of 100 workflows with a period of 60 seconds, so it may not start more than 100 workflows in 60 seconds: + +```java +Queue queue = DBOS.Queue("example-queue") + .limit(100, 60.0) // 100 workflows per 60 seconds + .build(); +``` + +Rate limits are especially useful when working with a rate-limited API, such as many LLM APIs. + +## Deduplication + +You can set a deduplication ID for an enqueued workflow using [`withQueue`](../reference/workflows-steps.md#startworkflow) when calling `startWorkflow`. +At any given time, only one workflow with a specific deduplication ID can be enqueued in the specified queue. +If a workflow with a deduplication ID is currently enqueued or actively executing (status `ENQUEUED` or `PENDING`), subsequent workflow enqueue attempts with the same deduplication ID in the same queue will raise an exception. + +For example, this is useful if you only want to have one workflow active at a time per user—set the deduplication ID to the user's ID. + +**Example syntax:** + +```java +@Workflow(name = "taskWorkflow") +public String taskWorkflow(String task) { + // Process the task... + return "completed"; +} + +public void example(Example proxy, String task, String userID) throws Exception { + // Use user ID for deduplication + WorkflowHandle handle = DBOS.startWorkflow( + () -> proxy.taskWorkflow(task), + new StartWorkflowOptions().withQueue(queue).withDeduplicationId(userID) + ); + + String result = handle.getResult(); + System.out.println("Workflow completed: " + result); +} +``` + +## Priority + +You can set a priority for an enqueued workflow using [`withQueue`](../reference/workflows-steps.md#startworkflow) when calling `startWorkflow`. +Workflows with the same priority are dequeued in **FIFO (first in, first out)** order. Priority values can range from `1` to `2,147,483,647`, where **a low number indicates a higher priority**. +If using priority, you must set [`priorityEnabled`](../reference/queues.md#queue) on your queue. + +:::tip +Workflows without assigned priorities have the highest priority and are dequeued before workflows with assigned priorities. +::: + +To use priorities in a queue, you must enable it when creating the queue: + +```java +Queue queue = DBOS.Queue("example-queue") + .priorityEnabled(true) + .build(); +``` + +**Example syntax:** + +```java +@Workflow(name = "taskWorkflow") +public String taskWorkflow(String task) { + // Process the task... + return "completed"; +} + +public void example(Example proxy, String task, int priority) throws Exception { + WorkflowHandle handle = DBOS.startWorkflow( + () -> proxy.taskWorkflow(task), + new StartWorkflowOptions().withQueue(queue).withPriority(priority) + ); + + String result = handle.getResult(); + System.out.println("Workflow completed: " + result); +} +``` diff --git a/docs/java/tutorials/step-tutorial.md b/docs/java/tutorials/step-tutorial.md new file mode 100644 index 000000000..bdf65b977 --- /dev/null +++ b/docs/java/tutorials/step-tutorial.md @@ -0,0 +1,90 @@ +--- +sidebar_position: 20 +title: Steps +--- + +When using DBOS workflows, you should call any method that performs complex operations or accesses external APIs or services as a _step_. +If a workflow is interrupted, upon restart it automatically resumes execution from the **last completed step**. + +You can use [`runStep`](../reference/workflows-steps.md#runstep) to call a method as a step. +A step can return any serializable value and may throw checked or unchecked exceptions. + +Here's a simple example: + +```java +class ExampleImpl implements Example { + + private int generateRandomNumber(int n) { + return new Random().nextInt(n); + } + + @Workflow(name = "workflowFunction") + public int workflowFunction(int n) { + int randomNumber = DBOS.runStep( + () -> generateRandomNumber(n), // Code to run as a checkpointed step + "generateRandomNumber" // StepOptions can also be used here + ); + return randomNumber; + } +} +``` + +You should make a method a step if you're using it in a DBOS workflow and it performs a [**nondeterministic**](./workflow-tutorial.md#determinism) operation. +A nondeterministic operation is one that may return different outputs given the same inputs. +Common nondeterministic operations include: + +- Accessing an external API or service, like serving a file from [AWS S3](https://aws.amazon.com/s3/), calling an external API like [Stripe](https://stripe.com/), or accessing an external data store like [Elasticsearch](https://www.elastic.co/elasticsearch/). +- Accessing files on disk. +- Generating a random number. +- Getting the current time. + +You **cannot** call, start, or enqueue workflows from within steps. +These operations should be performed from workflow methods. +You can call one step from another step, but the called step becomes part of the calling step's execution rather than functioning as a separate step. + +## Configurable Retries + +You can optionally configure a step to automatically retry any error a set number of times with exponential backoff. +This is useful for automatically handling transient failures, like making requests to unreliable APIs. +Retries are configurable through step options that can be passed to `runStep`. + +Available retry configuration options include: +- [`withRetriesAllowed`](../reference/workflows-steps.md#runstep) - Whether to retry the step if it throws an exception (default: false). +- [`withMaxAttempts`](../reference/workflows-steps.md#runstep) - Maximum number of times this step is automatically retried on failure. +- [`withIntervalSeconds`](../reference/workflows-steps.md#runstep) - Initial delay between retries in seconds. +- [`withBackoffRate`](../reference/workflows-steps.md#runstep) - Exponential backoff multiplier between retries. + +For example, let's configure this step to retry failures (such as if the site to be fetched is temporarily down) up to 10 times: + +```java +class ExampleImpl implements Example { + + private String fetchStep(String url) throws Exception { + HttpClient client = HttpClient.newHttpClient(); + HttpRequest request = HttpRequest.newBuilder() + .uri(URI.create(url)) + .build(); + + HttpResponse response = client.send( + request, + HttpResponse.BodyHandlers.ofString() + ); + + return response.body(); + } + + @Workflow(name = "fetchWorkflow") + public String fetchWorkflow(String inputURL) throws Exception { + return DBOS.runStep( + () -> fetchStep(inputURL), + new StepOptions("fetchFunction") + .withRetriesAllowed(true) + .withMaxAttempts(10) + .withIntervalSeconds(0.5) + .withBackoffRate(2.0) + ); + } +} +``` + +If a step exhausts all retry attempts, it throws an exception to the calling workflow. diff --git a/docs/java/tutorials/workflow-communication.md b/docs/java/tutorials/workflow-communication.md new file mode 100644 index 000000000..c379aae3d --- /dev/null +++ b/docs/java/tutorials/workflow-communication.md @@ -0,0 +1,167 @@ +--- +sidebar_position: 50 +title: Communicating with Workflows +--- + +DBOS provides a few different ways to communicate with your workflows. +You can: + +- [Send messages to workflows](#workflow-messaging-and-notifications) +- [Publish events from workflows for clients to read](#workflow-events) + + +## Workflow Messaging and Notifications +You can send messages to a specific workflow. +This is useful for signaling a workflow or sending notifications to it while it's running. + +DBOS Steps + +#### Send + +```java +static void send(String destinationId, Object message, String topic) +``` + +You can call `dbos.send()` to send a message to a workflow. +Messages can optionally be associated with a topic and are queued on the receiver per topic. + +You can also call [`send`](../reference/client.md#send) from outside of your DBOS application with the [DBOS Client](../reference/client.md). + +#### Recv + +```java +static Object recv(String topic, Duration timeout) +``` + +Workflows can call `dbos.recv()` to receive messages sent to them, optionally for a particular topic. +Each call to `recv()` waits for and consumes the next message to arrive in the queue for the specified topic, returning `null` if the wait times out. +If the topic is not specified, this method only receives messages sent without a topic. + +#### Messages Example + +Messages are especially useful for sending notifications to a workflow. +For example, in a payments system, after redirecting customers to a payments page, the checkout workflow must wait for a notification that the user has paid. + +To wait for this notification, the payments workflow uses `recv()`, executing failure-handling code if the notification doesn't arrive in time: + +```java +interface Checkout { + void checkoutWorkflow(); +} + +class CheckoutImpl implements Checkout { + private static final String PAYMENT_STATUS = "payment_status"; + + @Workflow(name = "checkout-workflow") + public void checkoutWorkflow() { + // Validate the order, redirect the customer to a payments page, + // then wait for a notification. + String paymentStatus = (String) DBOS.recv(PAYMENT_STATUS, Duration.ofSeconds(60)); + if (paymentStatus != null && paymentStatus.equals("paid")) { + // Handle a successful payment. + } else { + // Handle a failed payment or timeout. + } + } +} +``` + +An endpoint waits for the payment processor to send the notification, then uses `send()` to forward it to the workflow: + +```java +// Using Javalin for the HTTP endpoint +app.post("/payment_webhook/{workflow_id}/{payment_status}", ctx -> { + String workflowId = ctx.pathParam("workflow_id"); + String paymentStatus = ctx.pathParam("payment_status"); + // Send the payment status to the checkout workflow. + DBOS.send(workflowId, paymentStatus, PAYMENT_STATUS); + ctx.result("Payment status sent"); +}); +``` + +#### Reliability Guarantees + +All messages are persisted to the database, so if `send` completes successfully, the destination workflow is guaranteed to be able to `recv` it. +If you're sending a message from a workflow, DBOS guarantees exactly-once delivery. +If you're sending a message from normal Java code, you can use a unique workflow ID to guarantee exactly-once delivery. + +## Workflow Events + +Workflows can publish _events_, which are key-value pairs associated with the workflow. +They are useful for publishing information about the status of a workflow or to send a result to clients while the workflow is running. + +DBOS Steps + +#### setEvent + +```java +static void setEvent(String key, Object value) +``` + +Any workflow can call [`dbos.setEvent`](../reference/methods.md#setevent) to publish a key-value pair, or update its value if it has already been published. + +#### getEvent + +```java +static Object getEvent(String workflowId, String key, Duration timeout) +``` + +You can call [`dbos.getEvent`](../reference/methods.md#getevent) to retrieve the value published by a particular workflow identity for a particular key. +If the event does not yet exist, this call waits for it to be published, returning `null` if the wait times out. + +You can also call [`getEvent`](../reference/client.md#getevent) from outside of your DBOS application with [DBOS Client](../reference/client.md). + +#### Events Example + +Events are especially useful for writing interactive workflows that communicate information to their caller. +For example, in a checkout system, after validating an order, the checkout workflow needs to send the customer a unique payment ID. +To communicate the payment ID to the customer, it uses events. + +The payments workflow emits the payment ID using `setEvent()`: + +```java +interface Checkout { + void checkoutWorkflow(); +} + +class CheckoutImpl implements Checkout { + private static final String PAYMENT_ID = "payment_id"; + + @Workflow(name = "checkout-workflow") + public void checkoutWorkflow() { + // ... validation logic + String paymentId = generatePaymentId(); + DBOS.setEvent(PAYMENT_ID, paymentId); + // ... continue processing + } +} +``` + +The Javalin handler that originally started the workflow uses `getEvent()` to await this payment ID, then returns it: + +```java +// Using Javalin for the HTTP endpoint +app.post("/checkout/{idempotency_key}", ctx -> { + String idempotencyKey = ctx.pathParam("idempotency_key"); + + // Idempotently start the checkout workflow in the background. + WorkflowHandle handle = DBOS.startWorkflow( + () -> checkoutProxy.checkoutWorkflow(), + new StartWorkflowOptions().withWorkflowId(idempotencyKey) + ); + + // Wait for the checkout workflow to send a payment ID, then return it. + String paymentId = (String) DBOS.getEvent(handle.getWorkflowId(), PAYMENT_ID, Duration.ofSeconds(60)); + if (paymentId == null) { + ctx.status(404); + ctx.result("Checkout failed to start"); + } else { + ctx.result(paymentId); + } +}); +``` + +#### Reliability Guarantees + +All events are persisted to the database, so the latest version of an event is always retrievable. +Additionally, if `getEvent` is called in a workflow, the retrieved value is persisted in the database so workflow recovery can use that value, even if the event is later updated. diff --git a/docs/java/tutorials/workflow-management.md b/docs/java/tutorials/workflow-management.md new file mode 100644 index 000000000..26a168454 --- /dev/null +++ b/docs/java/tutorials/workflow-management.md @@ -0,0 +1,49 @@ +--- +sidebar_position: 60 +title: Workflow Management +--- + +You can view and manage your durable workflow executions via a web UI ([self-hosted](../../production/self-hosting/workflow-management.md), [DBOS Cloud](../../production/dbos-cloud/workflow-management.md)) or programmatically. + +## Listing Workflows + +You can list your application's workflows programmatically via [`dbos.listWorkflows`](../reference/methods.md#listworkflows) or using the [`DBOSClient`](../reference/client.md#listworkflows). + +You can also view a searchable and expandable list of your application's workflows from its page on the DBOS Console (either [self-hosted](../../production/self-hosting/workflow-management.md) or on [DBOS Cloud](../../production/dbos-cloud/workflow-management.md)). + +Workflow List + +## Listing Workflow Steps + +You can list the steps of a workflow programmatically via [`dbos.listWorkflowSteps`](../reference/methods.md#listworkflowsteps) or using the [`DBOSClient`](../reference/client.md#listworkflowsteps). + +You can also visualize a workflow's execution graph (including the workflow, its steps, and its child workflows and their steps) from its page on the DBOS Console (either [self-hosted](../../production/self-hosting/workflow-management.md) or on [DBOS Cloud](../../production/dbos-cloud/workflow-management.md)). +For example, here is the graph of a workflow that processes multiple tasks concurrently by enqueueing child workflows: + +Workflow List + +## Cancelling Workflows + +You can cancel the execution of a workflow from the web UI, programmatically via [`dbos.cancelWorkflow`](../reference/methods.md#cancelworkflow), or using the [`DBOSClient`](../reference/client.md#cancelworkflow). + +If the workflow is currently executing, cancelling it preempts its execution (interrupting it at the beginning of its next step). +If the workflow is enqueued, cancelling removes it from the queue. + +## Resuming Workflows + +You can resume a workflow from its last completed step from the web UI, programmatically via [`dbos.resumeWorkflow`](../reference/methods.md#resumeworkflow), or using the [`DBOSClient`](../reference/client.md#resumeworkflow). + +You can use this to resume workflows that are cancelled or that have exceeded their maximum recovery attempts. +You can also use this to start an enqueued workflow immediately, bypassing its queue. + +## Forking Workflows + +You can start a new execution of a workflow by **forking** it from a specific step. +When you fork a workflow, DBOS generates a new workflow with a new workflow ID, copies to that workflow the original workflow's inputs and all its steps up to the selected step, then begins executing the new workflow from the selected step. + +Forking a workflow is useful for recovering from outages in downstream services (by forking from the step that failed after the outage is resolved) or for "patching" workflows that failed due to a bug in a previous application version (by forking from the bugged step to an application version on which the bug is fixed). + +You can fork a workflow programmatically using [`dbos.forkWorkflow`](../reference/methods.md#forkworkflow) or using the [`DBOSClient`](../reference/client.md#forkworkflow). +You can also fork a workflow from a step from the web UI by clicking on that step in the workflow's graph visualization: + +Workflow List diff --git a/docs/java/tutorials/workflow-tutorial.md b/docs/java/tutorials/workflow-tutorial.md new file mode 100644 index 000000000..26cf1661c --- /dev/null +++ b/docs/java/tutorials/workflow-tutorial.md @@ -0,0 +1,248 @@ +--- +sidebar_position: 10 +title: Workflows +toc_max_heading_level: 3 +--- + +Workflows provide **durable execution** so you can write programs that are **resilient to any failure**. +Workflows are comprised of [steps](./step-tutorial.md), which wrap ordinary Java methods. +If a workflow is interrupted for any reason (e.g., an executor restarts or crashes), when your program restarts the workflow automatically resumes execution from the last completed step. + +To write a workflow, annotate a method with [`@Workflow`](../reference/workflows-steps.md#workflow). +All workflow methods must be registered with DBOS by creating a proxy before DBOS is launched. +A workflow method can have any parameters and return type (including void), as long as they are serializable. + +Here's an example of a workflow: + +```java +interface Example { + public String workflow(); +} + +class ExampleImpl implements Example { + + private String stepOne() { + System.out.println("Step one completed"); + return "success"; + } + + private String stepTwo() { + System.out.println("Step two completed"); + return "success"; + } + + @Workflow(name = "workflow") + public String workflow() { + DBOS.runStep(() -> stepOne(), "stepOne"); + DBOS.runStep(() -> stepTwo(), "stepTwo"); + return "success"; + } +} + +public class App { + public static void main(String[] args) throws Exception { + // Configure DBOS + DBOSConfig config = ... + DBOS.configure(config); + + // Create the workflow proxy + Example proxy = DBOS.registerWorkflows(Example.class, new ExampleImpl()); + + // Launch DBOS after registering all workflows + DBOS.launch(); + + // Call the workflow + String result = proxy.workflow(); + System.out.println("Workflow result: " + result); + } +} +``` + +## Starting Workflows In The Background + +One common use-case for workflows is building reliable background tasks that keep running even when your program is interrupted, restarted, or crashes. +You can use [`startWorkflow`](../reference/workflows-steps.md#startworkflow) to start a workflow in the background. +When you start a workflow this way, it returns a [workflow handle](../reference/workflows-steps.md#workflowhandle), from which you can access information about the workflow or wait for it to complete and retrieve its result. + +Here's an example: + +```java +class ExampleImpl implements Example { + @Workflow(name = "backgroundTask") + public String backgroundTask(String input) { + // ... + return output; + } +} + +public void runWorkflowExample(Example proxy) throws Exception { + // Start the background task + WorkflowHandle handle = DBOS.startWorkflow( + () -> proxy.backgroundTask("input"), + new StartWorkflowOptions() + ); + // Wait for the background task to complete and retrieve its result + String result = handle.getResult(); + System.out.println("Workflow result: " + result); +} +``` + +After starting a workflow in the background, you can use [`retrieveWorkflow`](../reference/methods.md#retrieveworkflow) to retrieve a workflow's handle from its ID. +You can also retrieve a workflow's handle from outside of your DBOS application with [`DBOSClient.retrieveWorkflow`](../reference/client.md#retrieveworkflow). + +If you need to run many workflows in the background and manage their concurrency or flow control, use [queues](./queue-tutorial.md). + +## Workflow IDs and Idempotency + +Every time you execute a workflow, that execution is assigned a unique ID, by default a [UUID](https://en.wikipedia.org/wiki/Universally_unique_identifier). +You can access this ID from the [`DBOS.workflowId`](../reference/methods.md#workflowid-1) method. +Workflow IDs are useful for communicating with workflows and developing interactive workflows. + +You can set the workflow ID of a workflow using `withWorkflowId` when calling `startWorkflow`. +Workflow IDs are **globally unique** within your application. +An assigned workflow ID acts as an idempotency key: if a workflow is called multiple times with the same ID, it executes only once. +This is useful if your operations have side effects like making a payment or sending an email. +For example: + +```java +class ExampleImpl implements Example { + @Workflow(name = "exampleWorkflow") + public String exampleWorkflow() { + System.out.println("Running workflow with ID: " + DBOS.workflowId()); + // ... + return "success"; + } +} + +public void example(Example proxy) throws Exception { + String myID = "unique-workflow-id-123"; + WorkflowHandle handle = DBOS.startWorkflow( + () -> proxy.exampleWorkflow(), + new StartWorkflowOptions().withWorkflowId(myID) + ); + String result = handle.getResult(); + System.out.println("Result: " + result); +} +``` + +## Determinism + +Workflows are in most respects normal Java methods. +They can have loops, branches, conditionals, and so on. +However, a workflow method must be **deterministic**: if called multiple times with the same inputs, it should invoke the same steps with the same inputs in the same order (given the same return values from those steps). +If you need to perform a non-deterministic operation like accessing the database, calling a third-party API, generating a random number, or getting the local time, you shouldn't do it directly in a workflow method. +Instead, you should do all non-deterministic operations in [steps](./step-tutorial.md). + +:::warning +Java's threading and concurrency APIs are non-deterministic. You should use them only inside steps. +::: + +For example, **don't do this**: + +```java +@Workflow(name = "exampleWorkflow") +public String exampleWorkflow() { + int randomChoice = new Random().nextInt(2); + if (randomChoice == 0) { + return DBOS.runStep(() -> stepOne(), "stepOne"); + } else { + return DBOS.runStep(() -> stepTwo(), "stepTwo"); + } +} +``` + +Instead, do this: + +```java +private int generateChoice() { + return new Random().nextInt(2); +} + +@Workflow(name = "exampleWorkflow") +public String exampleWorkflow() { + int randomChoice = DBOS.runStep(() -> generateChoice(), "generateChoice"); + if (randomChoice == 0) { + return DBOS.runStep(() -> stepOne(), "stepOne"); + } else { + return DBOS.runStep(() -> stepTwo(), "stepTwo"); + } +} +``` + +## Workflow Timeouts + +You can set a timeout for a workflow using [`withTimeout`](../reference/workflows-steps.md#startworkflow) in `StartWorkflowOptions`. + +When the timeout expires, the workflow and all its children are cancelled. Cancelling a workflow sets its status to CANCELLED and preempts its execution at the beginning of its next step. You can detach a child workflow from its parent's timeout by starting it with a custom timeout using `withTimeout`. + +Timeouts are **start-to-completion**: if a workflow is [enqueued](./queue-tutorial.md), the timeout does not begin until the workflow is dequeued and starts execution. Also, timeouts are durable: they are stored in the database and persist across restarts, so workflows can have very long timeouts. + +```java +@Workflow(name = "exampleWorkflow") +public void exampleWorkflow() throws InterruptedException { + // Workflow implementation +} + +WorkflowHandle handle = DBOS.startWorkflow( + () -> proxy.exampleWorkflow(), + new StartWorkflowOptions().withTimeout(Duration.ofHours(12)) +); +``` + +You can also manually cancel the workflow by calling [`cancelWorkflow`](../reference/methods.md#cancelworkflow). + + +## Durable Sleep + +You can use [`sleep`](../reference/methods.md#sleep) to put your workflow to sleep for any period of time. +This sleep is **durable**—DBOS saves the wakeup time in the database so that even if the workflow is interrupted and restarted multiple times while sleeping, it still wakes up on schedule. + +Sleeping is useful for scheduling work to run in the future (even days, weeks, or months from now). +For example: + +```java +public String runTask(String task) { + // Execute the task... + return "task completed"; +} + +@Workflow(name = "exampleWorkflow") +public String exampleWorkflow(float timeToSleepSeconds, String task) throws InterruptedException { + // Sleep for the specified duration + DBOS.sleep(Duration.ofMillis((long)(timeToSleepSeconds*1000))); + + // Execute the task after sleeping + String result = DBOS.runStep( + () -> runTask(task), + "runTask" + ); + + return result; +} +``` + +## Workflow Guarantees + +Workflows provide the following reliability guarantees. +These guarantees assume that the application and database may crash and go offline at any point in time, but are always restarted and return online. + +1. Workflows always run to completion. If a DBOS process is interrupted while executing a workflow and restarts, it resumes the workflow from the last completed step. +2. [Steps](./step-tutorial.md) are tried _at least once_ but are never re-executed after they complete. If a failure occurs inside a step, the step may be retried, but once a step has completed (returned a value or thrown an exception to the calling workflow), it will never be re-executed. + +If an exception is thrown from a workflow, the workflow **terminates**—DBOS records the exception, sets the workflow status to `ERROR`, and **does not recover the workflow**. +This is because uncaught exceptions are assumed to be nonrecoverable. +If your workflow performs operations that may transiently fail (for example, sending HTTP requests to unreliable services), those should be performed in [steps with configured retries](./step-tutorial.md#configurable-retries). +DBOS provides [tooling](../reference/methods.md#workflow-management-methods) to help you identify failed workflows and examine the specific uncaught exceptions. + +## Workflow Versioning and Recovery + +Because DBOS recovers workflows by re-executing them using information saved in the database, a workflow cannot safely be recovered if its code has changed since the workflow was started. +To guard against this, DBOS _versions_ applications and their workflows. +When DBOS is launched, it computes an application version from a hash of the application source code (this can be overridden through [configuration](../reference/lifecycle.md)). +All workflows are tagged with the application version on which they started. + +When DBOS tries to recover workflows, it only recovers workflows whose version matches the current application version. +This prevents unsafe recovery of workflows that depend on different code. +You cannot change the version of a workflow, but you can use [`forkWorkflow`](../reference/methods.md#forkworkflow) to restart a workflow from a specific step on a specific code version. + +For more information on managing workflow recovery when self-hosting production DBOS applications, check out [the guide](../../production/self-hosting/workflow-recovery.md). diff --git a/docs/python/tutorials/queue-tutorial.md b/docs/python/tutorials/queue-tutorial.md index 7cc66d924..3d2617610 100644 --- a/docs/python/tutorials/queue-tutorial.md +++ b/docs/python/tutorials/queue-tutorial.md @@ -204,7 +204,7 @@ def on_user_task_submission(user_id: str, task: Task): ## Deduplication -You can set a deduplication ID for an enqueued workflow with [`SetEnqueueOptions`](../reference/contexts.md#setenqueueoptions). +You can set a deduplication ID for an enqueued workflow with [`SetEnqueueOptions`](../reference/queues.md#setenqueueoptions). At any given time, only one workflow with a specific deduplication ID can be enqueued in the specified queue. If a workflow with a deduplication ID is currently enqueued or actively executing (status `ENQUEUED` or `PENDING`), subsequent workflow enqueue attempt with the same deduplication ID in the same queue will raise a `DBOSQueueDeduplicatedError` exception. @@ -227,7 +227,7 @@ with SetEnqueueOptions(deduplication_id="my_dedup_id"): ## Priority -You can set a priority for an enqueued workflow with [`SetEnqueueOptions`](../reference/contexts.md#setenqueueoptions). +You can set a priority for an enqueued workflow with [`SetEnqueueOptions`](../reference/queues.md#setenqueueoptions). Workflows with the same priority are dequeued in **FIFO (first in, first out)** order. Priority values can range from `1` to `2,147,483,647`, where **a low number indicates a higher priority**. If using priority, you must set `priority_enabled=True` on your queue. diff --git a/docusaurus.config.js b/docusaurus.config.js index e716b1b29..3f833000f 100644 --- a/docusaurus.config.js +++ b/docusaurus.config.js @@ -299,7 +299,7 @@ const config = { }, }, prism: { - additionalLanguages: ['bash'], + additionalLanguages: ['bash', 'java'], theme: { ...prismThemes.okaidia, styles: [ diff --git a/sidebars.js b/sidebars.js index 70ba9c204..05f8c6436 100644 --- a/sidebars.js +++ b/sidebars.js @@ -64,6 +64,16 @@ const sidebars = { } ], }, + { + type: 'category', + label: 'Develop with Java', + items: [ + { + type: 'autogenerated', + dirName: 'java', + } + ], + }, { type: 'category', label: 'Deploy To Production', @@ -139,6 +149,17 @@ const sidebars = { } ], collapsed: false, + }, + { + type: 'category', + label: 'Java Examples', + items: [ + { + type: 'autogenerated', + dirName: 'java/examples', + } + ], + collapsed: false, } ], };