Merge pull request #15 from dbos-inc/manoj/sleep

Manoj/sleep

Author: manojdbos

DEVELOPING.md

@@ -0,0 +1,38 @@
+# dbos-transact-java
+DBOS Transact Java SDK
+
+## Setting up dev environment
+
+Install a recent OpenJDK. I use OpenJDK 21.    
+https://adoptium.net/en-GB/temurin/releases/?os=any&arch=any&version=21
+
+Recommended IDE IntelliJ (Community edition is fine).
+But feel free to use vi or VSCode, if you are more comfortable with it.
+
+Postgres docker container with
+localhost   
+port 5432   
+user postgres
+
+export PGPASSWORD = password for postgres user
+
+## build
+
+./gradlew clean build
+
+## run tests
+
+./gradlew clean test
+
+## publish to local maven repository
+
+./gradlew publishToMavenLocal
+
+## import transact into your application
+
+Add to your build.gradle.kts
+
+implementation("dev.dbos:transact:1.0-SNAPSHOT")      
+implementation("ch.qos.logback:logback-classic:1.5.6")
+
+Annotations @Workflow, @Transaction, @Step need to be on implementation class methods. 

README.md

@@ -1,41 +1,222 @@
-# dbos-transact-java
-DBOS Transact Java SDK 
+<div align="center">
 
-## Setting up dev environment
+# DBOS Transact: Lightweight Durable Workflows
 
-Install a recent OpenJDK. I use OpenJDK 21.    
-https://adoptium.net/en-GB/temurin/releases/?os=any&arch=any&version=21
+#### [Documentation](https://docs.dbos.dev/) &nbsp;&nbsp;•&nbsp;&nbsp;  [Examples](https://docs.dbos.dev/examples) &nbsp;&nbsp;•&nbsp;&nbsp; [Github](https://github.com/dbos-inc) &nbsp;&nbsp;•&nbsp;&nbsp; [Discord](https://discord.com/invite/jsmC6pXGgX)
+</div>
 
-Recommended IDE IntelliJ (Community edition is fine).
-But feel free to use vi or VSCode, if you are more comfortable with it.  
+---
 
-Postgres docker container with
-localhost   
-port 5432   
-user postgres
+## What is DBOS?
 
-export PGPASSWORD = password for postgres user  
+DBOS provides lightweight durable workflows built on top of Postgres.
+Instead of managing your own workflow orchestrator or task queue system, you can use DBOS to add durable workflows and queues to your program in just a few lines of code.
 
-## build
+To get started, follow the [quickstart](https://docs.dbos.dev/quickstart) to install this open-source library and connect it to a Postgres database.
+Then, annotate workflows and steps in your program to make it durable!
+That's all you need to do&mdash;DBOS is entirely contained in this open-source library, there's no additional infrastructure for you to configure or manage.
 
-./gradlew clean build
+## When Should I Use DBOS?
 
-## run tests
+You should consider using DBOS if your application needs to **reliably handle failures**.
+For example, you might be building a payments service that must reliably process transactions even if servers crash mid-operation, or a long-running data pipeline that needs to resume seamlessly from checkpoints rather than restart from the beginning when interrupted.
 
-./gradlew clean test
+Handling failures is costly and complicated, requiring complex state management and recovery logic as well as heavyweight tools like external orchestration services.
+DBOS makes it simpler: annotate your code to checkpoint it in Postgres and automatically recover from any failure.
+DBOS also provides powerful Postgres-backed primitives that makes it easier to write and operate reliable code, including durable queues, notifications, scheduling, event processing, and programmatic workflow management.
 
-## publish to local maven repository
+## Features
 
-./gradlew publishToMavenLocal
+<details open><summary><strong>💾 Durable Workflows</strong></summary>
 
-## import transact into your application
+####
 
-Add to your build.gradle.kts
+DBOS workflows make your program **durable** by checkpointing its state in Postgres.
+If your program ever fails, when it restarts all your workflows will automatically resume from the last completed step.
+
+You add durable workflows to your existing Java program by annotating ordinary functions as workflows and steps:
+
+```java
+
+public interface SimpleWorkflowService {
+
+    void setSimpleWorkflowService(SimpleWorkflowService e);
+    String exampleWorkflow(String input) ;
+    void stepOne() ;
+    void stepTwo() ;
+}
+
+public class SimpleWorkflowServiceImpl implements SimpleWorkflowService {
+
+    public void setSimpleWorkflowService(SimpleWorkflowService simpleWorkflow) {
+        this.simpleWorkflowService = simpleWorkflow;
+    }
+    
+    @Workflow(name = "exampleWorkflow")
+    public String exampleWorkflow(String input) {
+        return input + input;
+    }
+    @Step(name = "stepOne")
+    public void stepOne() {
+        logger.info("Executed stepOne") ;
+    }
+    @Step(name = "stepTwo")
+    public void stepTwo() {
+        logger.info("Executed stepTwo") ;
+    }
+    
+}
+
+public class Demo {
+    
+    public static void main(String[] args) {
+        
+        DBOSConfig dbosConfig = new DBOSConfig.Builder()
+                .name("demo")
+                .dbHost("localhost")
+                .dbPort(5432)
+                .dbUser("postgres")
+                .sysDbName("demo_dbos_sys")
+                .build() ;
+
+        DBOS.initialize(dbosConfig);
+        DBOS dbos = DBOS.getInstance();
+        dbos.launch();
+        
+        SimpleWorkflowService syncExample = dbos.<SimpleWorkflowService>Workflow()
+                .interfaceClass(SimpleWorkflowService.class)
+                .implementation(new SimpleWorkflowServiceImpl())
+                .build();
+        syncExample.setSimpleWorkflowService(syncExample);
+
+        String output = syncExample.exampleWorkflow("HelloDBOS") ;
+        System.out.println("Sync result: " + output);
+    }
+}
+
+
+```
+
+Workflows are particularly useful for
+
+- Orchestrating business processes so they seamlessly recover from any failure.
+- Building observable and fault-tolerant data pipelines.
+- Operating an AI agent, or any application that relies on unreliable or non-deterministic APIs.
+
+[Read more ↗️]()
+
+</details>
+
+<details><summary><strong>📒 Durable Queues</strong></summary>
+
+####
+
+DBOS queues help you **durably** run tasks in the background.
+You can enqueue a task (which can be a single step or an entire workflow) from a durable workflow and one of your processes will pick it up for execution.
+DBOS manages the execution of your tasks: it guarantees that tasks complete, and that their callers get their results without needing to resubmit them, even if your application is interrupted.
+
+Queues also provide flow control, so you can limit the concurrency of your tasks on a per-queue or per-process basis.
+You can also set timeouts for tasks, rate limit how often queued tasks are executed, deduplicate tasks, or prioritize tasks.
+
+You can add queues to your workflows in just a couple lines of code.
+They don't require a separate queueing service or message broker&mdash;just Postgres.
+
+```java
+
+
+ public void queuedTasks() {
+     Queue q = new DBOS.QueueBuilder("childQ").build();
+
+     for (int i = 0; i < 3; i++) {
+
+         String wid = "child" + i;
+         DBOSOptions options = new DBOSOptions.Builder(wid).queue(q).build();
+         try (SetDBOSOptions o = new SetDBOSOptions(options)) {
+             simpleService.childWorkflow(wid);
+         }
+     }
+
+     for (int i = 0 ; i < 3 ; i++) {
+         String wid = "child"+i;
+         WorkflowHandle h = DBOS.retrieveWorkflow(wid);
+         System.out.println(h.getResult());
+     }
+ }
+```
+
+[Read more ↗️](https://docs.dbos.dev/python/tutorials/queue-tutorial)
+
+</details>
+
+
+## Getting Started
+
+To get started, follow the [quickstart](https://docs.dbos.dev/quickstart) to install this open-source library and connect it to a Postgres database.
+Then, check out the [programming guide](https://docs.dbos.dev/python/programming-guide) to learn how to build with durable workflows and queues.
+
+## Documentation
+
+[https://docs.dbos.dev](https://docs.dbos.dev)
+
+## Examples
+
+[https://docs.dbos.dev/examples](https://docs.dbos.dev/examples)
+
+## DBOS vs. Other Systems
+
+<details><summary><strong>DBOS vs. Temporal</strong></summary>
+
+####
+
+Both DBOS and Temporal provide durable execution, but DBOS is implemented in a lightweight Postgres-backed library whereas Temporal is implemented in an externally orchestrated server.
+
+You can add DBOS to your program by installing this open-source library, connecting it to Postgres, and annotating workflows and steps.
+By contrast, to add Temporal to your program, you must rearchitect your program to move your workflows and steps (activities) to a Temporal worker, configure a Temporal server to orchestrate those workflows, and access your workflows only through a Temporal client.
+[This blog post](https://www.dbos.dev/blog/durable-execution-coding-comparison) makes the comparison in more detail.
+
+**When to use DBOS:** You need to add durable workflows to your applications with minimal rearchitecting, or you are using Postgres.
+
+**When to use Temporal:** You don't want to add Postgres to your stack, or you need a language DBOS doesn't support yet.
+
+</details>
+
+<details><summary><strong>DBOS vs. Airflow</strong></summary>
+
+####
+
+DBOS and Airflow both provide workflow abstractions.
+Airflow is targeted at data science use cases, providing many out-of-the-box connectors but requiring workflows be written as explicit DAGs and externally orchestrating them from an Airflow cluster.
+Airflow is designed for batch operations and does not provide good performance for streaming or real-time use cases.
+DBOS is general-purpose, but is often used for data pipelines, allowing developers to write workflows as code and requiring no infrastructure except Postgres.
+
+**When to use DBOS:** You need the flexibility of writing workflows as code, or you need higher performance than Airflow is capable of (particularly for streaming or real-time use cases).
+
+**When to use Airflow:** You need Airflow's ecosystem of connectors.
+
+</details>
+
+<details><summary><strong>DBOS vs. Celery/BullMQ</strong></summary>
+
+####
+
+DBOS provides a similar queue abstraction to dedicated queueing systems like Celery or BullMQ: you can declare queues, submit tasks to them, and control their flow with concurrency limits, rate limits, timeouts, prioritization, etc.
+However, DBOS queues are **durable and Postgres-backed** and integrate with durable workflows.
+For example, in DBOS you can write a durable workflow that enqueues a thousand tasks and waits for their results.
+DBOS checkpoints the workflow and each of its tasks in Postgres, guaranteeing that even if failures or interruptions occur, the tasks will complete and the workflow will collect their results.
+By contrast, Celery/BullMQ are Redis-backed and don't provide workflows, so they provide fewer guarantees but better performance.
+
+**When to use DBOS:** You need the reliability of enqueueing tasks from durable workflows.
+
+**When to use Celery/BullMQ**: You don't need durability, or you need very high throughput beyond what your Postgres server can support.
+</details>
+
+## Community
+
+If you want to ask questions or hang out with the community, join us on [Discord](https://discord.gg/fMwQjeW5zg)!
+If you see a bug or have a feature request, don't hesitate to open an issue here on GitHub.
+If you're interested in contributing, check out our [contributions guide](./CONTRIBUTING.md).
 
-implementation("dev.dbos:transact:1.0-SNAPSHOT")      
-implementation("ch.qos.logback:logback-classic:1.5.6")   
 
-Annotations @Workflow, @Transaction, @Step need to be on implementation class methods. 
 
 
 

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

@@ -326,5 +326,19 @@ public void setEvent(String key, Object value) {
     public Object getEvent(String workflowId, String key, float timeOut) {
         return notificationService.getEvent(workflowId, key, timeOut) ;
     }
+
+    /**
+     *
+     * Durable sleep. When you are in a workflow, use this instead of Thread.sleep.
+     * On restart or during recovery the original expected wakeup time is honoured as
+     * opposed to sleeping all over again.
+     *
+     * @param seconds in seconds
+     */
+
+    public void sleep(float seconds) {
+
+        this.dbosExecutor.sleep(seconds) ;
+    }
 }
 

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

@@ -160,7 +160,7 @@ public Object recv(String workflowUuid, int functionId, int timeoutFunctionId,
             if (!hasExistingNotification) {
                 // Wait for the notification
                 // Support OAOO sleep
-                double actualTimeout = sleep(workflowUuid, timeoutFunctionId, timeoutSeconds, true);
+                double actualTimeout = stepsDAO.sleep(workflowUuid, timeoutFunctionId, timeoutSeconds, true);
                 long timeoutMs = (long) (actualTimeout * 1000);
                 lockPair.condition.await(timeoutMs, TimeUnit.MILLISECONDS);
 
@@ -226,61 +226,7 @@ public Object recv(String workflowUuid, int functionId, int timeoutFunctionId,
             }
         }
     }
-
-    // TODO : can be moved elsewhere when we implement DBOS.sleep
-    public double sleep(String workflowUuid, int functionId, double seconds, boolean skipSleep) throws SQLException {
-        String functionName = "DBOS.sleep";
-
-        StepResult recordedOutput = null ;
-
-        try (Connection c = dataSource.getConnection()) {
-            recordedOutput = stepsDAO.checkStepExecutionTxn(workflowUuid, functionId, functionName, c) ;
-        }
-
-        Double endTime;
-        if (recordedOutput != null) {
-            logger.debug("Replaying sleep, id: {}, seconds: {}", functionId, seconds);
-            String output = recordedOutput.getOutput();
-            if (output == null) {
-                throw new AssertionError("no recorded end time");
-            }
-            endTime = (Double)JSONUtil.deserializeToArray(output)[0] ;
-        } else {
-            logger.debug("Running sleep, id: {}, seconds: {}", functionId, seconds);
-            endTime = System.currentTimeMillis() / 1000.0 + seconds;
-
-            try {
-
-                StepResult output = new StepResult();
-                output.setWorkflowId(workflowUuid);
-                output.setFunctionId(functionId);
-                output.setFunctionName(functionName);
-                output.setOutput(JSONUtil.serialize(endTime));
-                output.setError(null);
-
-                try(Connection conn = dataSource.getConnection()) {
-                    stepsDAO.recordStepResultTxn(output, conn);
-                }
-            } catch (DBOSWorkflowConflictException e) {
-                // Ignore conflict - operation already recorded
-                logger.error("Error recording sleep error ",e);
-            }
-        }
-
-        double duration = Math.max(0, endTime - (System.currentTimeMillis() / 1000.0));
-
-        if (!skipSleep) {
-            try {
-                Thread.sleep((long) (duration * 1000));
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                throw new RuntimeException("Sleep interrupted", e);
-            }
-        }
-
-        return duration;
-    }
-
+    
     public void setEvent(String workflowId, int functionId, String key, Object message) throws SQLException {
         String functionName = "DBOS.setEvent";
 
@@ -403,7 +349,7 @@ public Object getEvent(String targetUuid, String key, double timeoutSeconds, Get
                 double actualTimeout = timeoutSeconds;
                 if (callerCtx != null) {
                     // Support OAOO sleep for workflows
-                    actualTimeout = sleep(
+                    actualTimeout = stepsDAO.sleep(
                             callerCtx.getWorkflowId(),
                             callerCtx.getTimeoutFunctionId(),
                             timeoutSeconds,