Getevent recv durable sleep (#111)

- Support durable sleep for GetEvent and Recv (this essentially records
a sleep step when sleep is needed)
- Be less verbose in the DBOS logs

Author: maxdml

CONTRIBUTING.md

@@ -1,4 +1,4 @@
-# Contributing to DBOS Transact Python
+# Contributing to DBOS Transact Golang
 
 Thank you for considering contributing to DBOS Transact. We welcome contributions from everyone, including bug fixes, feature enhancements, documentation improvements, or any other form of contribution.
 

dbos/dbos.go

@@ -330,7 +330,7 @@ func NewDBOSContext(inputConfig Config) (DBOSContext, error) {
 		return nil, newInitializationError(fmt.Sprintf("failed to create system database: %v", err))
 	}
 	initExecutor.systemDB = systemDB
-	initExecutor.logger.Info("System database initialized")
+	initExecutor.logger.Debug("System database initialized")
 
 	// Initialize the queue runner and register DBOS internal queue
 	initExecutor.queueRunner = newQueueRunner(initExecutor.logger)
@@ -355,7 +355,7 @@ func NewDBOSContext(inputConfig Config) (DBOSContext, error) {
 			return nil, newInitializationError(fmt.Sprintf("failed to initialize conductor: %v", err))
 		}
 		initExecutor.conductor = conductor
-		initExecutor.logger.Info("Conductor initialized")
+		initExecutor.logger.Debug("Conductor initialized")
 	}
 
 	return initExecutor, nil
@@ -383,7 +383,7 @@ func (c *dbosContext) Launch() error {
 			c.logger.Error("Failed to start admin server", "error", err)
 			return newInitializationError(fmt.Sprintf("failed to start admin server: %v", err))
 		}
-		c.logger.Info("Admin server started", "port", c.config.AdminServerPort)
+		c.logger.Debug("Admin server started", "port", c.config.AdminServerPort)
 		c.adminServer = adminServer
 	}
 
@@ -392,18 +392,18 @@ func (c *dbosContext) Launch() error {
 	go func() {
 		c.queueRunner.run(c)
 	}()
-	c.logger.Info("Queue runner started")
+	c.logger.Debug("Queue runner started")
 
 	// Start the workflow scheduler if it has been initialized
 	if c.workflowScheduler != nil {
 		c.workflowScheduler.Start()
-		c.logger.Info("Workflow scheduler started")
+		c.logger.Debug("Workflow scheduler started")
 	}
 
 	// Start the conductor if it has been initialized
 	if c.conductor != nil {
 		c.conductor.Launch()
-		c.logger.Info("Conductor started")
+		c.logger.Debug("Conductor started")
 	}
 
 	// Run a round of recovery on the local executor
@@ -414,7 +414,7 @@ func (c *dbosContext) Launch() error {
 	if len(recoveryHandles) > 0 {
 		c.logger.Info("Recovered pending workflows", "count", len(recoveryHandles))
 	} else {
-		c.logger.Info("No pending workflows to recover")
+		c.logger.Debug("No pending workflows to recover")
 	}
 
 	c.logger.Info("DBOS initialized", "app_version", c.applicationVersion, "executor_id", c.executorID)
@@ -438,45 +438,45 @@ func (c *dbosContext) Launch() error {
 //
 // Shutdown is a permanent operation and should be called when the application is terminating.
 func (c *dbosContext) Shutdown(timeout time.Duration) {
-	c.logger.Info("Shutting down DBOS context")
+	c.logger.Debug("Shutting down DBOS context")
 
 	// Cancel the context to signal all resources to stop
 	c.ctxCancelFunc(errors.New("DBOS cancellation initiated"))
 
 	// Wait for all workflows to finish
-	c.logger.Info("Waiting for all workflows to finish")
+	c.logger.Debug("Waiting for all workflows to finish")
 	done := make(chan struct{})
 	go func() {
 		c.workflowsWg.Wait()
 		close(done)
 	}()
 	select {
 	case <-done:
-		c.logger.Info("All workflows completed")
+		c.logger.Debug("All workflows completed")
 	case <-time.After(timeout):
 		// For now just log a warning: eventually we might want Cancel to return an error.
 		c.logger.Warn("Timeout waiting for workflows to complete", "timeout", timeout)
 	}
 
 	// Wait for queue runner to finish
 	if c.queueRunner != nil && c.launched.Load() {
-		c.logger.Info("Waiting for queue runner to complete")
+		c.logger.Debug("Waiting for queue runner to complete")
 		select {
 		case <-c.queueRunner.completionChan:
-			c.logger.Info("Queue runner completed")
+			c.logger.Debug("Queue runner completed")
 		case <-time.After(timeout):
 			c.logger.Warn("Timeout waiting for queue runner to complete", "timeout", timeout)
 		}
 	}
 
 	// Stop the workflow scheduler and wait until all scheduled workflows are done
 	if c.workflowScheduler != nil && c.launched.Load() {
-		c.logger.Info("Stopping workflow scheduler")
+		c.logger.Debug("Stopping workflow scheduler")
 		ctx := c.workflowScheduler.Stop()
 
 		select {
 		case <-ctx.Done():
-			c.logger.Info("All scheduled jobs completed")
+			c.logger.Debug("All scheduled jobs completed")
 			c.workflowScheduler = nil
 		case <-time.After(timeout):
 			c.logger.Warn("Timeout waiting for jobs to complete. Moving on", "timeout", timeout)
@@ -485,24 +485,24 @@ func (c *dbosContext) Shutdown(timeout time.Duration) {
 
 	// Shutdown the conductor
 	if c.conductor != nil {
-		c.logger.Info("Shutting down conductor")
+		c.logger.Debug("Shutting down conductor")
 		c.conductor.Shutdown(timeout)
 	}
 
 	// Shutdown the admin server
 	if c.adminServer != nil && c.launched.Load() {
-		c.logger.Info("Shutting down admin server")
+		c.logger.Debug("Shutting down admin server")
 		err := c.adminServer.Shutdown(timeout)
 		if err != nil {
 			c.logger.Error("Failed to shutdown admin server", "error", err)
 		} else {
-			c.logger.Info("Admin server shutdown complete")
+			c.logger.Debug("Admin server shutdown complete")
 		}
 	}
 
 	// Close the system database
 	if c.systemDB != nil {
-		c.logger.Info("Shutting down system database")
+		c.logger.Debug("Shutting down system database")
 		c.systemDB.shutdown(c, timeout)
 	}
 

dbos/queue.go

@@ -264,7 +264,7 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 		// Sleep with jittered interval, but allow early exit on context cancellation
 		select {
 		case <-ctx.Done():
-			qr.logger.Info("Queue runner stopping due to context cancellation", "cause", context.Cause(ctx))
+			qr.logger.Debug("Queue runner stopping due to context cancellation", "cause", context.Cause(ctx))
 			qr.completionChan <- struct{}{}
 			return
 		case <-time.After(sleepDuration):

dbos/system_database.go

@@ -54,7 +54,7 @@ type systemDatabase interface {
 	getEvent(ctx context.Context, input getEventInput) (any, error)
 
 	// Timers (special steps)
-	sleep(ctx context.Context, duration time.Duration) (time.Duration, error)
+	sleep(ctx context.Context, input sleepInput) (time.Duration, error)
 
 	// Queues
 	dequeueWorkflows(ctx context.Context, input dequeueWorkflowsInput) ([]dequeuedWorkflow, error)
@@ -111,7 +111,7 @@ func createDatabaseIfNotExists(ctx context.Context, databaseURL string, logger *
 		if err != nil {
 			return newInitializationError(fmt.Sprintf("failed to create database %s: %v", dbName, err))
 		}
-		logger.Info("Database created", "name", dbName)
+		logger.Debug("Database created", "name", dbName)
 	}
 
 	return nil
@@ -304,7 +304,7 @@ func (s *sysDB) launch(ctx context.Context) {
 }
 
 func (s *sysDB) shutdown(ctx context.Context, timeout time.Duration) {
-	s.logger.Info("DBOS: Closing system database connection pool")
+	s.logger.Debug("DBOS: Closing system database connection pool")
 	if s.pool != nil {
 		s.pool.Close()
 	}
@@ -319,7 +319,7 @@ func (s *sysDB) shutdown(ctx context.Context, timeout time.Duration) {
 
 	if s.launched {
 		// Wait for the notification loop to exit
-		s.logger.Info("DBOS: Waiting for notification listener loop to finish")
+		s.logger.Debug("DBOS: Waiting for notification listener loop to finish")
 		select {
 		case <-s.notificationLoopDone:
 		case <-time.After(timeout):
@@ -1438,10 +1438,17 @@ func (s *sysDB) getWorkflowSteps(ctx context.Context, workflowID string) ([]Step
 	return steps, nil
 }
 
+type sleepInput struct {
+	duration  time.Duration // Duration to sleep
+	skipSleep bool          // If true, the function will not actually sleep (useful for testing)
+}
+
 // Sleep is a special type of step that sleeps for a specified duration
 // A wakeup time is computed and recorded in the database
 // If we sleep is re-executed, it will only sleep for the remaining duration until the wakeup time
-func (s *sysDB) sleep(ctx context.Context, duration time.Duration) (time.Duration, error) {
+// sleep can be called within other special steps (e.g., getEvent, recv) to provide durable sleep
+
+func (s *sysDB) sleep(ctx context.Context, input sleepInput) (time.Duration, error) {
 	functionName := "DBOS.sleep"
 
 	// Get workflow state from context
@@ -1486,9 +1493,9 @@ func (s *sysDB) sleep(ctx context.Context, duration time.Duration) (time.Duratio
 		}
 	} else {
 		// First execution: calculate and record the end time
-		s.logger.Debug("Durable sleep", "stepID", stepID, "duration", duration)
+		s.logger.Debug("Durable sleep", "stepID", stepID, "duration", input.duration)
 
-		endTime = time.Now().Add(duration)
+		endTime = time.Now().Add(input.duration)
 
 		// Record the operation result with the calculated end time
 		recordInput := recordOperationResultDBInput{
@@ -1512,8 +1519,10 @@ func (s *sysDB) sleep(ctx context.Context, duration time.Duration) (time.Duratio
 	// Calculate remaining duration until wake up time
 	remainingDuration := max(0, time.Until(endTime))
 
-	// Actually sleep for the remaining duration
-	time.Sleep(remainingDuration)
+	if !input.skipSleep {
+		// Actually sleep for the remaining duration
+		time.Sleep(remainingDuration)
+	}
 
 	return remainingDuration, nil
 }
@@ -1527,7 +1536,7 @@ func (s *sysDB) notificationListenerLoop(ctx context.Context) {
 		s.notificationLoopDone <- struct{}{}
 	}()
 
-	s.logger.Info("DBOS: Starting notification listener loop")
+	s.logger.Debug("DBOS: Starting notification listener loop")
 	mrr := s.notificationListenerConnection.Exec(ctx, fmt.Sprintf("LISTEN %s; LISTEN %s", _DBOS_NOTIFICATIONS_CHANNEL, _DBOS_WORKFLOW_EVENTS_CHANNEL))
 	results, err := mrr.ReadAll()
 	if err != nil {
@@ -1555,7 +1564,7 @@ func (s *sysDB) notificationListenerLoop(ctx context.Context) {
 		if err != nil {
 			// Context cancellation
 			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
-				s.logger.Info("Notification listener loop exiting due to context cancellation", "cause", context.Cause(ctx), "error", err)
+				s.logger.Debug("Notification listener loop exiting due to context cancellation", "cause", context.Cause(ctx), "error", err)
 				return
 			}
 
@@ -1750,10 +1759,18 @@ func (s *sysDB) recv(ctx context.Context, input recvInput) (any, error) {
 			close(done)
 		}()
 
+		timeout, err := s.sleep(ctx, sleepInput{
+			duration:  input.Timeout,
+			skipSleep: true,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to sleep before recv timeout: %w", err)
+		}
+
 		select {
 		case <-done:
 			s.logger.Debug("Received notification on condition variable", "payload", payload)
-		case <-time.After(input.Timeout):
+		case <-time.After(timeout):
 			s.logger.Warn("Recv() timeout reached", "payload", payload, "timeout", input.Timeout)
 		}
 	}
@@ -1969,11 +1986,21 @@ func (s *sysDB) getEvent(ctx context.Context, input getEventInput) (any, error)
 			close(done)
 		}()
 
+		timeout := input.Timeout
+		if isInWorkflow {
+			timeout, err = s.sleep(ctx, sleepInput{
+				duration:  input.Timeout,
+				skipSleep: true,
+			})
+			if err != nil {
+				return nil, fmt.Errorf("failed to sleep before getEvent timeout: %w", err)
+			}
+		}
+
 		select {
 		case <-done:
 			// Received notification
-		case <-time.After(input.Timeout):
-			// Timeout reached
+		case <-time.After(timeout):
 			s.logger.Warn("GetEvent() timeout reached", "target_workflow_id", input.TargetWorkflowID, "key", input.Key, "timeout", input.Timeout)
 		case <-ctx.Done():
 			return nil, fmt.Errorf("context cancelled while waiting for event: %w", ctx.Err())

dbos/workflow.go

@@ -1241,7 +1241,7 @@ func GetEvent[R any](ctx DBOSContext, targetWorkflowID, key string, timeout time
 }
 
 func (c *dbosContext) Sleep(_ DBOSContext, duration time.Duration) (time.Duration, error) {
-	return c.systemDB.sleep(c, duration)
+	return c.systemDB.sleep(c, sleepInput{duration: duration, skipSleep: false})
 }
 
 // Sleep pauses workflow execution for the specified duration.