Various fixes (#43)

- print scheduled workflows schedules
- fix bug in test
- wait for workflows to complete in shutdown
- fix GetResult() error handling
- Handle errors in admin server
- move global variables in singleton
- fix `processConfig` now that app name and db url are mandatory

Author: maxdml

dbos/admin_server.go

@@ -64,9 +64,10 @@ func newAdminServer(port int) *adminServer {
 		}
 
 		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(http.StatusOK)
 		if err := json.NewEncoder(w).Encode(workflowIDs); err != nil {
 			getLogger().Error("Error encoding response", "error", err)
+			http.Error(w, fmt.Sprintf("Failed to encode response: %v", err), http.StatusInternalServerError)
+			return
 		}
 	})
 
@@ -92,9 +93,10 @@ func newAdminServer(port int) *adminServer {
 		}
 
 		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(http.StatusOK)
 		if err := json.NewEncoder(w).Encode(queueMetadataArray); err != nil {
 			getLogger().Error("Error encoding queue metadata response", "error", err)
+			http.Error(w, fmt.Sprintf("Failed to encode response: %v", err), http.StatusInternalServerError)
+			return
 		}
 	})
 

dbos/dbos.go

@@ -8,17 +8,14 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
-	"net/url"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/robfig/cron/v3"
 )
 
 var (
-	_APP_VERSION               string
-	_EXECUTOR_ID               string
-	_APP_ID                    string
 	_DEFAULT_ADMIN_SERVER_PORT = 3001
 )
 
@@ -40,10 +37,7 @@ type Config struct {
 	AdminServer bool
 }
 
-// processConfig merges configuration from two sources in order of precedence:
-// 1. programmatic configuration
-// 2. environment variables
-// Finally, it applies default values if needed.
+// processConfig enforces mandatory fields and applies defaults.
 func processConfig(inputConfig *Config) (*Config, error) {
 	// First check required fields
 	if len(inputConfig.DatabaseURL) == 0 {
@@ -53,33 +47,17 @@ func processConfig(inputConfig *Config) (*Config, error) {
 		return nil, fmt.Errorf("missing required config field: appName")
 	}
 
-	dbosConfig := &Config{}
-
-	// Start with environment variables (lowest precedence)
-	if dbURL := os.Getenv("DBOS_SYSTEM_DATABASE_URL"); dbURL != "" {
-		dbosConfig.DatabaseURL = dbURL
-	}
-
-	// Override with programmatic configuration (highest precedence)
-	if len(inputConfig.DatabaseURL) > 0 {
-		dbosConfig.DatabaseURL = inputConfig.DatabaseURL
-	}
-	if len(inputConfig.AppName) > 0 {
-		dbosConfig.AppName = inputConfig.AppName
+	dbosConfig := &Config{
+		DatabaseURL: inputConfig.DatabaseURL,
+		AppName:     inputConfig.AppName,
+		Logger:      inputConfig.Logger,
+		AdminServer: inputConfig.AdminServer,
 	}
-	// Copy over parameters that can only be set programmatically
-	dbosConfig.Logger = inputConfig.Logger
-	dbosConfig.AdminServer = inputConfig.AdminServer
 
 	// Load defaults
 	if dbosConfig.Logger == nil {
 		dbosConfig.Logger = slog.New(slog.NewTextHandler(os.Stderr, nil))
 	}
-	if len(dbosConfig.DatabaseURL) == 0 {
-		dbosConfig.Logger.Info("Using default database URL: postgres://postgres:${PGPASSWORD}@localhost:5432/dbos?sslmode=disable")
-		password := url.QueryEscape(os.Getenv("PGPASSWORD"))
-		dbosConfig.DatabaseURL = fmt.Sprintf("postgres://postgres:%s@localhost:5432/dbos?sslmode=disable", password)
-	}
 
 	return dbosConfig, nil
 }
@@ -93,6 +71,10 @@ type executor struct {
 	queueRunnerDone       chan struct{}
 	adminServer           *adminServer
 	config                *Config
+	applicationVersion    string
+	applicationID         string
+	executorID            string
+	workflowsWg           *sync.WaitGroup
 }
 
 func Initialize(inputConfig Config) error {
@@ -101,11 +83,16 @@ func Initialize(inputConfig Config) error {
 		return newInitializationError("DBOS already initialized")
 	}
 
+	initExecutor := &executor{
+		workflowsWg: &sync.WaitGroup{},
+	}
+
 	// Load & process the configuration
 	config, err := processConfig(&inputConfig)
 	if err != nil {
 		return newInitializationError(err.Error())
 	}
+	initExecutor.config = config
 
 	// Set global logger
 	logger = config.Logger
@@ -115,32 +102,30 @@ func Initialize(inputConfig Config) error {
 	gob.Register(t)
 
 	// Initialize global variables with environment variables, providing defaults if not set
-	_APP_VERSION = os.Getenv("DBOS__APPVERSION")
-	if _APP_VERSION == "" {
-		_APP_VERSION = computeApplicationVersion()
+	initExecutor.applicationVersion = os.Getenv("DBOS__APPVERSION")
+	if initExecutor.applicationVersion == "" {
+		initExecutor.applicationVersion = computeApplicationVersion()
 		logger.Info("DBOS__APPVERSION not set, using computed hash")
 	}
 
-	_EXECUTOR_ID = os.Getenv("DBOS__VMID")
-	if _EXECUTOR_ID == "" {
-		_EXECUTOR_ID = "local"
-		logger.Info("DBOS__VMID not set, using default", "executor_id", _EXECUTOR_ID)
+	initExecutor.executorID = os.Getenv("DBOS__VMID")
+	if initExecutor.executorID == "" {
+		initExecutor.executorID = "local"
+		logger.Info("DBOS__VMID not set, using default", "executor_id", initExecutor.executorID)
 	}
 
-	_APP_ID = os.Getenv("DBOS__APPID")
+	initExecutor.applicationID = os.Getenv("DBOS__APPID")
 
 	// Create the system database
 	systemDB, err := NewSystemDatabase(config.DatabaseURL)
 	if err != nil {
 		return newInitializationError(fmt.Sprintf("failed to create system database: %v", err))
 	}
+	initExecutor.systemDB = systemDB
 	logger.Info("System database initialized")
 
 	// Set the global dbos instance
-	dbos = &executor{
-		config:   config,
-		systemDB: systemDB,
-	}
+	dbos = initExecutor
 
 	return nil
 }
@@ -184,15 +169,15 @@ func Launch() error {
 	}
 
 	// Run a round of recovery on the local executor
-	recoveryHandles, err := recoverPendingWorkflows(context.Background(), []string{_EXECUTOR_ID}) // XXX maybe use the queue runner context here to allow Shutdown to cancel it?
+	recoveryHandles, err := recoverPendingWorkflows(context.Background(), []string{dbos.executorID}) // XXX maybe use the queue runner context here to allow Shutdown to cancel it?
 	if err != nil {
 		return newInitializationError(fmt.Sprintf("failed to recover pending workflows during launch: %v", err))
 	}
 	if len(recoveryHandles) > 0 {
 		logger.Info("Recovered pending workflows", "count", len(recoveryHandles))
 	}
 
-	logger.Info("DBOS initialized", "app_version", _APP_VERSION, "executor_id", _EXECUTOR_ID)
+	logger.Info("DBOS initialized", "app_version", dbos.applicationVersion, "executor_id", dbos.executorID)
 	return nil
 }
 
@@ -203,6 +188,7 @@ func Shutdown() {
 	}
 
 	// XXX is there a way to ensure all workflows goroutine are done before closing?
+	dbos.workflowsWg.Wait()
 
 	// Cancel the context to stop the queue runner
 	if dbos.queueRunnerCancelFunc != nil {

dbos/queue.go

@@ -167,7 +167,7 @@ func queueRunner(ctx context.Context) {
 
 				_, err := registeredWorkflow.wrappedFunction(ctx, input, WithWorkflowID(workflow.id))
 				if err != nil {
-					getLogger().Error("Error recovering workflow", "error", err)
+					getLogger().Error("Error running queued workflow", "error", err)
 				}
 			}
 		}

dbos/queues_test.go

@@ -507,8 +507,8 @@ func TestWorkerConcurrency(t *testing.T) {
 
 	// Stop the queue runner before changing executor ID to avoid race conditions
 	stopQueueRunner()
-	// Change the EXECUTOR_ID global variable to a different value
-	_EXECUTOR_ID = "worker-2"
+	// Change the executor ID to a different value
+	dbos.executorID = "worker-2"
 	// Restart the queue runner
 	restartQueueRunner()
 
@@ -541,7 +541,7 @@ func TestWorkerConcurrency(t *testing.T) {
 	// Stop the queue runner before changing executor ID to avoid race conditions
 	stopQueueRunner()
 	// Change the executor again and wait for the third workflow to start
-	_EXECUTOR_ID = "local"
+	dbos.executorID = "local"
 	// Restart the queue runner
 	restartQueueRunner()
 	startEvents[2].Wait()
@@ -573,7 +573,7 @@ func TestWorkerConcurrency(t *testing.T) {
 	// Stop the queue runner before changing executor ID to avoid race conditions
 	stopQueueRunner()
 	// change executor again and wait for the fourth workflow to start
-	_EXECUTOR_ID = "worker-2"
+	dbos.executorID = "worker-2"
 	// Restart the queue runner
 	restartQueueRunner()
 	startEvents[3].Wait()
@@ -597,7 +597,7 @@ func TestWorkerConcurrency(t *testing.T) {
 		t.Fatal("expected queue entries to be cleaned up after global concurrency test")
 	}
 
-	_EXECUTOR_ID = "local" // Reset executor ID for future tests
+	dbos.executorID = "local" // Reset executor ID for future tests
 }
 
 var (

dbos/recovery.go

@@ -11,7 +11,7 @@ func recoverPendingWorkflows(ctx context.Context, executorIDs []string) ([]Workf
 	pendingWorkflows, err := dbos.systemDB.ListWorkflows(ctx, listWorkflowsDBInput{
 		status:             []WorkflowStatusType{WorkflowStatusPending},
 		executorIDs:        executorIDs,
-		applicationVersion: _APP_VERSION,
+		applicationVersion: dbos.applicationVersion,
 	})
 	if err != nil {
 		return nil, err

dbos/system_database.go

@@ -23,26 +23,39 @@ import (
 /*******************************/
 
 type SystemDatabase interface {
+	// SysDB management
 	Launch(ctx context.Context)
 	Shutdown()
 	ResetSystemDB(ctx context.Context) error
+
+	// Workflows
 	InsertWorkflowStatus(ctx context.Context, input insertWorkflowStatusDBInput) (*insertWorkflowResult, error)
-	RecordOperationResult(ctx context.Context, input recordOperationResultDBInput) error
-	RecordChildWorkflow(ctx context.Context, input recordChildWorkflowDBInput) error
-	CheckChildWorkflow(ctx context.Context, workflowUUID string, functionID int) (*string, error)
 	ListWorkflows(ctx context.Context, input listWorkflowsDBInput) ([]WorkflowStatus, error)
 	UpdateWorkflowOutcome(ctx context.Context, input updateWorkflowOutcomeDBInput) error
 	AwaitWorkflowResult(ctx context.Context, workflowID string) (any, error)
-	DequeueWorkflows(ctx context.Context, queue WorkflowQueue) ([]dequeuedWorkflow, error)
-	ClearQueueAssignment(ctx context.Context, workflowID string) (bool, error)
-	CheckOperationExecution(ctx context.Context, input checkOperationExecutionDBInput) (*recordedResult, error)
+
+	// Child workflows
+	RecordChildWorkflow(ctx context.Context, input recordChildWorkflowDBInput) error
+	CheckChildWorkflow(ctx context.Context, workflowUUID string, functionID int) (*string, error)
 	RecordChildGetResult(ctx context.Context, input recordChildGetResultDBInput) error
+
+	// Steps
+	RecordOperationResult(ctx context.Context, input recordOperationResultDBInput) error
+	CheckOperationExecution(ctx context.Context, input checkOperationExecutionDBInput) (*recordedResult, error)
 	GetWorkflowSteps(ctx context.Context, workflowID string) ([]StepInfo, error)
+
+	// Communication (special steps)
 	Send(ctx context.Context, input workflowSendInputInternal) error
 	Recv(ctx context.Context, input WorkflowRecvInput) (any, error)
 	SetEvent(ctx context.Context, input workflowSetEventInputInternal) error
 	GetEvent(ctx context.Context, input WorkflowGetEventInput) (any, error)
+
+	// Timers (special steps)
 	Sleep(ctx context.Context, duration time.Duration) (time.Duration, error)
+
+	// Queues
+	DequeueWorkflows(ctx context.Context, queue WorkflowQueue) ([]dequeuedWorkflow, error)
+	ClearQueueAssignment(ctx context.Context, workflowID string) (bool, error)
 }
 
 type systemDatabase struct {
@@ -1625,7 +1638,7 @@ func (s *systemDatabase) DequeueWorkflows(ctx context.Context, queue WorkflowQue
 			pendingWorkflowsDict[executorIDRow] = taskCount
 		}
 
-		localPendingWorkflows := pendingWorkflowsDict[_EXECUTOR_ID]
+		localPendingWorkflows := pendingWorkflowsDict[dbos.executorID]
 
 		// Check worker concurrency limit
 		if queue.workerConcurrency != nil {
@@ -1692,7 +1705,7 @@ func (s *systemDatabase) DequeueWorkflows(ctx context.Context, queue WorkflowQue
 	}
 
 	// Execute the query to get workflow IDs
-	rows, err := tx.Query(ctx, query, queue.name, WorkflowStatusEnqueued, _APP_VERSION)
+	rows, err := tx.Query(ctx, query, queue.name, WorkflowStatusEnqueued, dbos.applicationVersion)
 	if err != nil {
 		return nil, fmt.Errorf("failed to query enqueued workflows: %w", err)
 	}
@@ -1742,8 +1755,8 @@ func (s *systemDatabase) DequeueWorkflows(ctx context.Context, queue WorkflowQue
 		var inputString *string
 		err := tx.QueryRow(ctx, updateQuery,
 			WorkflowStatusPending,
-			_APP_VERSION,
-			_EXECUTOR_ID,
+			dbos.applicationVersion,
+			dbos.executorID,
 			startTimeMs,
 			id).Scan(&retWorkflow.name, &inputString)
 

dbos/utils_test.go

@@ -15,11 +15,11 @@ import (
 func getDatabaseURL(t *testing.T) string {
 	databaseURL := os.Getenv("DBOS_SYSTEM_DATABASE_URL")
 	if databaseURL == "" {
-		if os.Getenv("PGPASSWORD") == "" {
-			t.Skip("PGPASSWORD not set, cannot construct database URL")
+		password := os.Getenv("PGPASSWORD")
+		if password == "" {
+			password = "dbos"
 		}
-		password := url.QueryEscape(os.Getenv("PGPASSWORD"))
-		databaseURL = fmt.Sprintf("postgres://postgres:%s@localhost:5432/dbos?sslmode=disable", password)
+		databaseURL = fmt.Sprintf("postgres://postgres:%s@localhost:5432/dbos?sslmode=disable", url.QueryEscape(password))
 	}
 	return databaseURL
 }