Merge pull request #16 from dbos-inc/manoj/timeout

Workflow timeouts
changed compile to jdk17

Author: manojdbos

README.md

@@ -148,6 +148,111 @@ They don't require a separate queueing service or message broker—just Post
 
 </details>
 
+<details><summary><strong>📒 Asynchronous execution </strong></summary>
+
+####
+
+DBOS can run your workflows asynchronously without you needing to make any changes to the interface or implementation. 
+
+This is ideal for long-running workflows whose result might not be available immediately. 
+You code can return at a later point and check the status for completion and/or retrive the result.
+
+
+```java
+
+
+ public void runAsyncWorkflow() {
+
+     SimpleWorkflowService syncExample = dbos.<SimpleWorkflowService>Workflow()
+             .interfaceClass(SimpleWorkflowService.class)
+             .implementation(new SimpleWorkflowServiceImpl())
+             .build();
+     syncExample.setSimpleWorkflowService(syncExample);
+
+     String workflowId = "wf-124";
+     options = new DBOSOptions.Builder(workflowId).async().build();
+     try (SetDBOSOptions id = new SetDBOSOptions(options)) {
+         syncExample.exampleWorkflow("HelloDBOS");
+     }
+
+     WorkflowHandle<String> handle = DBOS.retrieveWorkflow(workflowId); ;
+     result = handle.getResult();
+     
+ }
+```
+
+[Read more ↗️](https://docs.dbos.dev/python/tutorials/queue-tutorial)
+
+</details>
+
+
+<details><summary><strong>📅 Durable Scheduling</strong></summary>
+
+####
+
+Schedule workflows using cron syntax, or use durable sleep to pause workflows for as long as you like (even days or weeks) before executing.
+
+You can schedule a workflow using a single annotation:
+
+```python
+@DBOS.scheduled('* * * * *') # crontab syntax to run once every minute
+@DBOS.workflow()
+def example_scheduled_workflow(scheduled_time: datetime, actual_time: datetime):
+    DBOS.logger.info("I am a workflow scheduled to run once a minute.")
+```
+
+You can add a durable sleep to any workflow with a single line of code.
+It stores its wakeup time in Postgres so the workflow sleeps through any interruption or restart, then always resumes on schedule.
+
+```java
+
+public class SchedulerImpl {
+    
+    @Workflow(name = "every5Second")
+    @Scheduled(cron = "0/5 * * * * ?")
+    public void every5Second(Instant schedule , Instant actual) {
+        log.info("Executed workflow  "+  schedule.toString() + "   " + actual.toString()) ;
+    }
+}
+
+// In your main
+// dbos.scheduleWorkflow(new SchedulerImpl());
+```
+
+[Read more ↗️](https://docs.dbos.dev/python/tutorials/scheduled-workflows)
+
+</details>
+
+<details><summary><strong>📫 Durable Notifications</strong></summary>
+
+####
+
+Pause your workflow executions until a notification is received, or emit events from your workflow to send progress updates to external clients.
+All notifications are stored in Postgres, so they can be sent and received with exactly-once semantics.
+Set durable timeouts when waiting for events, so you can wait for as long as you like (even days or weeks) through interruptions or restarts, then resume once a notification arrives or the timeout is reached.
+
+For example, build a reliable billing workflow that durably waits for a notification from a payments service, processing it exactly-once:
+
+```java
+@workflow(name = "billing")
+public void billingWorkflow() {
+    // Calculate the charge, then submit the bill to a payments service
+    String payment_status = (String) dbos.recv(PAYMENT_STATUS, timeout = payment_service_timeout);
+    if (payment_status.equals("paid")) {
+        // handle paid
+    } else {
+        // handle not paid
+    }
+}
+
+@workflow(name = "payment") 
+public void payment() {
+    dbos.send(targetWorkflowId, PAYMENT_STATUS, "paid") ;
+}
+      
+```
+</details>
+
 
 ## Getting Started
 

build.gradle.kts

@@ -8,12 +8,22 @@ plugins {
 group = "dev.dbos"
 version = "1.0-SNAPSHOT"
 
+java {
+    toolchain {
+        languageVersion = JavaLanguageVersion.of(17)
+    }
+}
 
 tasks.withType<JavaCompile> {
-    sourceCompatibility = "11"
-    targetCompatibility = "11"
+    options.release.set(11) // Targets Java 11 bytecode (RECOMMENDED)
+    // (Alternative: sourceCompatibility = "11"; targetCompatibility = "11")
 }
 
+// tasks.withType<JavaCompile> {
+//    sourceCompatibility = "11"
+//    targetCompatibility = "11"
+// }
+
 repositories {
     mavenCentral()
 }

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

@@ -22,8 +22,8 @@ public class DBOSContext {
 
     // workflow timeouts
     private Queue queue;
-    private int workflowTimeoutMs ;
-    private int workflowDeadlineEpochMs ;
+    private long workflowTimeoutMs ;
+    private long workflowDeadlineEpochMs ;
 
     // Queues
     private String deduplicationId ;
@@ -45,14 +45,16 @@ public DBOSContext(String workflowId, int functionId) {
         this.inWorkflow = false;
     }
 
-    public DBOSContext(String workflowId, int functionId, String parentWorkflowId, int parentFunctionId,boolean inWorkflow, boolean async, Queue q) {
+    public DBOSContext(String workflowId, int functionId, String parentWorkflowId, int parentFunctionId,
+                       boolean inWorkflow, boolean async, Queue q, long timeout) {
         this.workflowId = workflowId;
         this.functionId = functionId ;
         this.inWorkflow = inWorkflow;
         this.parentWorkflowId = parentWorkflowId;
         this.parentFunctionId = parentFunctionId;
         this.async = async;
         this.queue = q;
+        this.workflowTimeoutMs = timeout;
 
     }
 
@@ -62,26 +64,34 @@ public DBOSContext(DBOSOptions options, int functionId) {
         this.inWorkflow = false;
         this.async = options.isAsync() ;
         this.queue = options.getQueue();
+        this.workflowTimeoutMs = options.getTimeout()*1000;
     }
 
-    private DBOSContext(String childWorkflowId, String parentWorkflowId, int parentFunctionId, boolean async, Queue queue) {
+    private DBOSContext(String childWorkflowId, String parentWorkflowId, int parentFunctionId, boolean async, Queue queue, long workflowTimeout) {
         this.workflowId = childWorkflowId;
         this.parentWorkflowId = parentWorkflowId;
         this.parentFunctionId = parentFunctionId;
         this.inWorkflow = true;
         this.async = async ;
         this.queue = queue;
+        this.workflowTimeoutMs = workflowTimeout;
     }
 
-    private DBOSContext(DBOSOptions options, String parentWorkflowId, int parentFunctionId) {
+    private DBOSContext(DBOSOptions options, String parentWorkflowId, int parentFunctionId, long parentTimeout) {
         this.workflowId = options.getWorkflowId();
         this.parentWorkflowId = parentWorkflowId;
         this.parentFunctionId = parentFunctionId;
         this.inWorkflow = true;
         this.async = options.isAsync();
         this.queue = options.getQueue();
+        if (options.getTimeout() > 0) {
+            this.workflowTimeoutMs = options.getTimeout()*1000;
+        } else {
+            this.workflowTimeoutMs = parentTimeout ;
+        }
     }
 
+
     public String getWorkflowId() {
         return workflowId;
     }
@@ -119,15 +129,15 @@ public void setStepId(String stepId) {
     }
 
     public DBOSContext copy() {
-        return new DBOSContext(workflowId, functionId, parentWorkflowId, parentFunctionId, inWorkflow, async, queue);
+        return new DBOSContext(workflowId, functionId, parentWorkflowId, parentFunctionId, inWorkflow, async, queue, workflowTimeoutMs);
     }
 
     public DBOSContext createChild(String childWorkflowId) {
-        return new DBOSContext(childWorkflowId, workflowId, this.getAndIncrementFunctionId(), this.async, this.getQueue());
+        return new DBOSContext(childWorkflowId, workflowId, this.getAndIncrementFunctionId(), this.async, this.getQueue(), this.workflowTimeoutMs);
     }
 
     public DBOSContext createChild(DBOSOptions options) {
-        return new DBOSContext(options, workflowId, this.getAndIncrementFunctionId());
+        return new DBOSContext(options, workflowId, this.getAndIncrementFunctionId(), this.workflowTimeoutMs);
     }
 
     public boolean hasParent() {
@@ -149,5 +159,9 @@ public boolean isAsync() {
     public Queue getQueue() {
         return queue;
     }
+
+    public long getWorkflowTimeoutMs() {
+        return workflowTimeoutMs ;
+    }
 }
 

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

@@ -9,7 +9,7 @@ public class DBOSOptions {
     private final boolean async;
     private final Queue queue;
     private final String workflowId;
-    private final float timeoutSeconds ;
+    private final long timeoutSeconds ;
 
     private DBOSOptions(Builder builder) {
         this.async = builder.async;
@@ -32,15 +32,15 @@ public String getWorkflowId() {
     }
 
 
-    public float getTimeout() {
+    public long getTimeout() {
         return timeoutSeconds;
     }
 
     public static class Builder {
         private boolean async = false;
         private Queue queue = null;
         private String workflowId = null;
-        private float timeoutSeconds = 0f;
+        private long timeoutSeconds = 0;
 
         public Builder(String workflowId) {
             this.workflowId = workflowId;
@@ -57,7 +57,7 @@ public Builder queue(Queue queue) {
         }
 
 
-        public Builder timeout(float timeout) {
+        public Builder timeout(long timeout) {
             this.timeoutSeconds = timeout;
             return this;
         }

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

@@ -292,6 +292,16 @@ public double sleep(String workflowId, int functionId, double seconds, boolean s
 
     }
 
+    public void cancelWorkflow(String workflowId) {
+        try {
+            workflowDAO.cancelWorkflow(workflowId);
+        } catch (SQLException sq) {
+            logger.error("Sql Exception", sq);
+            throw new DBOSException(UNEXPECTED.getCode(), sq.getMessage());
+        }
+
+    }
+
     private void createDataSource(String dbName) {
         HikariConfig hikariConfig = new HikariConfig();
 

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

@@ -659,4 +659,45 @@ public void recordChildWorkflow(String parentId,
 
     }
 
+    public void cancelWorkflow(String workflowId) throws SQLException {
+
+        try (Connection conn = dataSource.getConnection()) {
+
+            // Check the status of the workflow. If it is complete, do nothing.
+            String checkStatusSql = " SELECT status FROM %s.workflow_status WHERE workflow_uuid = ? " ;
+            checkStatusSql = String.format(checkStatusSql, Constants.DB_SCHEMA);
+
+            String currentStatus = null;
+            try (PreparedStatement stmt = conn.prepareStatement(checkStatusSql)) {
+                stmt.setString(1, workflowId);
+                try (ResultSet rs = stmt.executeQuery()) {
+                    if (rs.next()) {
+                        currentStatus = rs.getString("status");
+                    }
+                }
+            }
+
+            // If workflow doesn't exist or is already complete, do nothing
+            if (currentStatus == null ||
+                WorkflowState.SUCCESS.name().equals(currentStatus) ||
+                WorkflowState.ERROR.name().equals(currentStatus)) {
+                logger.info("Returning without updating status") ;
+                return;
+            }
+
+            // Set the workflow's status to CANCELLED and remove it from any queue it is on
+            String updateSql = "UPDATE %s.workflow_status SET status = ?, " +
+                " queue_name = NULL, deduplication_id = NULL, started_at_epoch_ms = NULL " +
+                " WHERE workflow_uuid = ? " ;
+            updateSql = String.format(updateSql, Constants.DB_SCHEMA) ;
+
+            try (PreparedStatement stmt = conn.prepareStatement(updateSql)) {
+                stmt.setString(1, WorkflowState.CANCELLED.name());
+                stmt.setString(2, workflowId);
+                stmt.executeUpdate();
+            }
+
+        }
+    }
+
 }

src/main/java/dev/dbos/transact/exceptions/AwaitedWorkflowCancelledException.java

@@ -1,8 +1,14 @@
 package dev.dbos.transact.exceptions;
 
 public class AwaitedWorkflowCancelledException extends DBOSException {
+    private String workflowId ;
     public AwaitedWorkflowCancelledException(String workflowId) {
         super(ErrorCode.WORKFLOW_CONFLICT.getCode(),
                 String.format("Awaited workflow %s was cancelled.", workflowId));
+        this.workflowId = workflowId;
+    }
+
+    public String getWorkflowId() {
+        return workflowId;
     }
 }