Bag o stuff (#56)

Major fix: deadline computation for enqueued workflows. Previously we'd
actually set the deadline when the workflow was enqueued, not dequeued.
This PR fixes it by considering the timeout when setting the final
deadline. (Note: we also revert the change from #53 and let
`resumeWorkflow` clear the deadline. It'll be recomputed from the
timeout in `RunAsWorkflow`.)

Also:
- Add a test `TimeoutOnlySetOnDequeue`, to verify the above fix
- Record a workflow error when the function is called with the wrong
parameter types. This could happen if the client enqueues a workflow
with the wrong input type.
- Prevent the spawning of child workflows within steps and add a test
(`ChildWorkflowCannotBeSpawnedFromStep`)
- Add a GHA to perform vulnerability checks and static analysis
- Set toolchain to 1.25 (and update test GHA accordingly). Note this is
*not* the minimum version required by library consumers.
- Run `go vet` in test GHA
- Add "join discord" badge to README
- Make hardcoded values package level constants
- Fix a small bug where returning the admin server object would be
missing the port
- Handle errors when writing the response in the healthcheck admin
endpoint
- Set `ReadHeaderTimeout` on the admin server to prevent Slowloris
attacks
- Handle sub millisecond timeouts (round to next millisecond)
- Handle negative timeouts (can happen with propagation delays): set
timeout to 1ms
- Use a sync.Map for the workflow registry
- Check the executable is a regular file before opening it to compute
its hash
- Remove unused WorkflowFunctionNotFound -- I don't see a use case for
it
- Use `struct{}{}` instead of `bool` for notification channels
(`struct{}{}` use zero space)
- Use an input struct for `dequeueWorkflows`
- Cleanup handles implementation with a base handler that implement once
shared logic (`GetStatus`, `GetWorkflowID`). other handles compose with
this struct.
- Handle errors when failing to close notification listener connection
- Handle error when closing the connection used to set `LISTEN` channels
in Postgres
- Handle errors (and print a log entry) when failing to run a scheduled
workflow
- Fix possible integer overflows when converting uint to int (for queue
priorities and fork start step)
- Complete DLQ test: resume the DLQ-ed workflow and check it can now run
to completion. Check that the completed workflow can be ran many times
past the retry limit.
- Pass `workflowCustomNametoFQN` to new contexts

Author: maxdml

.github/workflows/security.yml

@@ -0,0 +1,52 @@
+name: Security Checks
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+    types:
+      - ready_for_review
+      - opened
+      - reopened
+      - synchronize
+  workflow_dispatch:
+
+jobs:
+  security:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        fetch-depth: 0
+
+    - name: Setup Go
+      uses: actions/setup-go@v5
+      with:
+        go-version: '1.25.x'
+
+    - name: Cache Go modules
+      uses: actions/cache@v4
+      with:
+        path: |
+          ~/go/pkg/mod
+          ~/.cache/go-build
+        key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+        restore-keys: |
+          ${{ runner.os }}-go-
+
+    - name: Download dependencies
+      run: go mod download
+
+    - name: Install security tools
+      run: |
+        go install golang.org/x/vuln/cmd/govulncheck@latest
+        go install github.com/securego/gosec/v2/cmd/gosec@latest
+
+    - name: Run govulncheck
+      run: govulncheck ./...
+
+    - name: Run gosec
+      run: gosec ./...

.github/workflows/tests.yml

@@ -42,7 +42,7 @@ jobs:
     - name: Setup Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.23.x'
+        go-version: '1.25.x'
 
     - name: Cache Go modules
       uses: actions/cache@v4
@@ -61,7 +61,7 @@ jobs:
       run: go install gotest.tools/gotestsum@latest
 
     - name: Run tests
-      run: gotestsum --format github-action -- -race ./...
+      run: go vet ./... && gotestsum --format github-action -- -race ./...
       working-directory: ./dbos
       env:
         PGPASSWORD: a!b@c$d()e*_,/:;=?@ff[]22

README.md

@@ -3,11 +3,12 @@
 [![Go Reference](https://pkg.go.dev/badge/github.com/dbos-inc/dbos-transact-go.svg)](https://pkg.go.dev/github.com/dbos-inc/dbos-transact-go)
 [![Go Report Card](https://goreportcard.com/badge/github.com/dbos-inc/dbos-transact-go)](https://goreportcard.com/report/github.com/dbos-inc/dbos-transact-go)
 [![GitHub release (latest SemVer)](https://img.shields.io/github/v/release/dbos-inc/dbos-transact-go?sort=semver)](https://github.com/dbos-inc/dbos-transact-go/releases)
+[![Join Discord](https://img.shields.io/badge/Discord-Join%20Chat-5865F2?logo=discord&logoColor=white)](https://discord.com/invite/jsmC6pXGgX)
 
 
 # DBOS Transact: Lightweight Durable Workflow Orchestration with Postgres
 
-#### [Documentation](https://docs.dbos.dev/)   •    [Examples](https://docs.dbos.dev/examples)   •   [Github](https://github.com/dbos-inc)   •   [Discord](https://discord.com/invite/jsmC6pXGgX)
+#### [Documentation](https://docs.dbos.dev/)   •    [Examples](https://docs.dbos.dev/examples)   •   [Github](https://github.com/dbos-inc)
 </div>
 
 #### This Golang version of DBOS Transact is in Alpha!

dbos/admin_server.go

@@ -10,28 +10,37 @@ import (
 )
 
 const (
-	healthCheckPath            = "/dbos-healthz"
-	workflowRecoveryPath       = "/dbos-workflow-recovery"
-	workflowQueuesMetadataPath = "/dbos-workflow-queues-metadata"
+	_HEALTHCHECK_PATH              = "/dbos-healthz"
+	_WORKFLOW_RECOVERY_PATH        = "/dbos-workflow-recovery"
+	_WORKFLOW_QUEUES_METADATA_PATH = "/dbos-workflow-queues-metadata"
+
+	_ADMIN_SERVER_READ_HEADER_TIMEOUT = 5 * time.Second
+	_ADMIN_SERVER_SHUTDOWN_TIMEOUT    = 10 * time.Second
 )
 
 type adminServer struct {
 	server *http.Server
 	logger *slog.Logger
+	port   int
 }
 
 func newAdminServer(ctx *dbosContext, port int) *adminServer {
 	mux := http.NewServeMux()
 
 	// Health endpoint
-	mux.HandleFunc(healthCheckPath, func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(_HEALTHCHECK_PATH, func(w http.ResponseWriter, r *http.Request) {
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
-		w.Write([]byte(`{"status":"healthy"}`))
+		_, err := w.Write([]byte(`{"status":"healthy"}`))
+		if err != nil {
+			ctx.logger.Error("Error writing health check response", "error", err)
+			http.Error(w, "Internal Server Error", http.StatusInternalServerError)
+			return
+		}
 	})
 
 	// Recovery endpoint
-	mux.HandleFunc(workflowRecoveryPath, func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(_WORKFLOW_RECOVERY_PATH, func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodPost {
 			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 			return
@@ -67,7 +76,7 @@ func newAdminServer(ctx *dbosContext, port int) *adminServer {
 	})
 
 	// Queue metadata endpoint
-	mux.HandleFunc(workflowQueuesMetadataPath, func(w http.ResponseWriter, r *http.Request) {
+	mux.HandleFunc(_WORKFLOW_QUEUES_METADATA_PATH, func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodGet {
 			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 			return
@@ -84,18 +93,20 @@ func newAdminServer(ctx *dbosContext, port int) *adminServer {
 	})
 
 	server := &http.Server{
-		Addr:    fmt.Sprintf(":%d", port),
-		Handler: mux,
+		Addr:              fmt.Sprintf(":%d", port),
+		Handler:           mux,
+		ReadHeaderTimeout: _ADMIN_SERVER_READ_HEADER_TIMEOUT,
 	}
 
 	return &adminServer{
 		server: server,
 		logger: ctx.logger,
+		port:   port,
 	}
 }
 
 func (as *adminServer) Start() error {
-	as.logger.Info("Starting admin server", "port", 3001)
+	as.logger.Info("Starting admin server", "port", as.port)
 
 	go func() {
 		if err := as.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
@@ -109,8 +120,8 @@ func (as *adminServer) Start() error {
 func (as *adminServer) Shutdown(ctx context.Context) error {
 	as.logger.Info("Shutting down admin server")
 
-	// XXX consider moving the grace period to DBOSContext.Shutdown()
-	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	// Note: consider moving the grace period to DBOSContext.Shutdown()
+	ctx, cancel := context.WithTimeout(ctx, _ADMIN_SERVER_SHUTDOWN_TIMEOUT)
 	defer cancel()
 
 	if err := as.server.Shutdown(ctx); err != nil {

dbos/admin_server_test.go

@@ -38,7 +38,7 @@ func TestAdminServer(t *testing.T) {
 
 		// Verify admin server is not running
 		client := &http.Client{Timeout: 1 * time.Second}
-		_, err = client.Get("http://localhost:3001" + healthCheckPath)
+		_, err = client.Get("http://localhost:3001" + _HEALTHCHECK_PATH)
 		require.Error(t, err, "Expected request to fail when admin server is not started")
 
 		// Verify the DBOS executor doesn't have an admin server instance
@@ -89,13 +89,13 @@ func TestAdminServer(t *testing.T) {
 			{
 				name:           "Health endpoint responds correctly",
 				method:         "GET",
-				endpoint:       "http://localhost:3001" + healthCheckPath,
+				endpoint:       "http://localhost:3001" + _HEALTHCHECK_PATH,
 				expectedStatus: http.StatusOK,
 			},
 			{
 				name:           "Recovery endpoint responds correctly with valid JSON",
 				method:         "POST",
-				endpoint:       "http://localhost:3001" + workflowRecoveryPath,
+				endpoint:       "http://localhost:3001" + _WORKFLOW_RECOVERY_PATH,
 				body:           bytes.NewBuffer(mustMarshal([]string{"executor1", "executor2"})),
 				contentType:    "application/json",
 				expectedStatus: http.StatusOK,
@@ -109,21 +109,21 @@ func TestAdminServer(t *testing.T) {
 			{
 				name:           "Recovery endpoint rejects invalid methods",
 				method:         "GET",
-				endpoint:       "http://localhost:3001" + workflowRecoveryPath,
+				endpoint:       "http://localhost:3001" + _WORKFLOW_RECOVERY_PATH,
 				expectedStatus: http.StatusMethodNotAllowed,
 			},
 			{
 				name:           "Recovery endpoint rejects invalid JSON",
 				method:         "POST",
-				endpoint:       "http://localhost:3001" + workflowRecoveryPath,
+				endpoint:       "http://localhost:3001" + _WORKFLOW_RECOVERY_PATH,
 				body:           strings.NewReader(`{"invalid": json}`),
 				contentType:    "application/json",
 				expectedStatus: http.StatusBadRequest,
 			},
 			{
 				name:           "Queue metadata endpoint responds correctly",
 				method:         "GET",
-				endpoint:       "http://localhost:3001" + workflowQueuesMetadataPath,
+				endpoint:       "http://localhost:3001" + _WORKFLOW_QUEUES_METADATA_PATH,
 				expectedStatus: http.StatusOK,
 				validateResp: func(t *testing.T, resp *http.Response) {
 					var queueMetadata []WorkflowQueue
@@ -149,7 +149,7 @@ func TestAdminServer(t *testing.T) {
 			{
 				name:           "Queue metadata endpoint rejects invalid methods",
 				method:         "POST",
-				endpoint:       "http://localhost:3001" + workflowQueuesMetadataPath,
+				endpoint:       "http://localhost:3001" + _WORKFLOW_QUEUES_METADATA_PATH,
 				expectedStatus: http.StatusMethodNotAllowed,
 			},
 		}

dbos/client_test.go

@@ -473,10 +473,6 @@ func TestCancelResume(t *testing.T) {
 		// Verify the deadline was reset (should be different from original)
 		assert.False(t, resumeStatus.Deadline.Equal(originalDeadline), "expected deadline to be reset after resume, but it remained the same: %v", originalDeadline)
 
-		// The new deadline should be after resumeStart + workflowTimeout
-		expectedDeadline := resumeStart.Add(workflowTimeout - 100*time.Millisecond) // Allow some leeway for processing time
-		assert.True(t, resumeStatus.Deadline.After(expectedDeadline), "deadline %v is too early (expected around %v)", resumeStatus.Deadline, expectedDeadline)
-
 		// Wait for the workflow to complete
 		_, err = resumeHandle.GetResult()
 		require.Error(t, err, "expected timeout error, but got none")
@@ -491,6 +487,10 @@ func TestCancelResume(t *testing.T) {
 		finalStatus, err := resumeHandle.GetStatus()
 		require.NoError(t, err, "failed to get final workflow status")
 
+		// The new deadline should have been set after resumeStart + workflowTimeout
+		expectedDeadline := resumeStart.Add(workflowTimeout - 100*time.Millisecond) // Allow some leeway for processing time
+		assert.True(t, finalStatus.Deadline.After(expectedDeadline), "deadline %v is too early (expected around %v)", resumeStatus.Deadline, expectedDeadline)
+
 		assert.Equal(t, WorkflowStatusCancelled, finalStatus.Status, "expected final workflow status to be CANCELLED")
 
 		require.True(t, queueEntriesAreCleanedUp(serverCtx), "expected queue entries to be cleaned up after cancel/resume timeout test")

dbos/dbos.go

@@ -15,6 +15,7 @@ import (
 	"io"
 	"log/slog"
 	"os"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -119,10 +120,9 @@ type dbosContext struct {
 	// Wait group for workflow goroutines
 	workflowsWg *sync.WaitGroup
 
-	// Workflow registry
-	workflowRegistry        map[string]workflowRegistryEntry
-	workflowRegMutex        *sync.RWMutex
-	workflowCustomNametoFQN sync.Map // Maps fully qualified workflow names to custom names. Usefor when client enqueues a workflow by name because registry is indexed by FQN.
+	// Workflow registry - read-mostly sync.Map since registration happens only before launch
+	workflowRegistry        *sync.Map // map[string]workflowRegistryEntry
+	workflowCustomNametoFQN *sync.Map // Maps fully qualified workflow names to custom names. Usefor when client enqueues a workflow by name because registry is indexed by FQN.
 
 	// Workflow scheduler
 	workflowScheduler *cron.Cron
@@ -158,15 +158,15 @@ func WithValue(ctx DBOSContext, key, val any) DBOSContext {
 	// Will do nothing if the concrete type is not dbosContext
 	if dbosCtx, ok := ctx.(*dbosContext); ok {
 		return &dbosContext{
-			ctx:                context.WithValue(dbosCtx.ctx, key, val), // Spawn a new child context with the value set
-			logger:             dbosCtx.logger,
-			systemDB:           dbosCtx.systemDB,
-			workflowsWg:        dbosCtx.workflowsWg,
-			workflowRegistry:   dbosCtx.workflowRegistry,
-			workflowRegMutex:   dbosCtx.workflowRegMutex,
-			applicationVersion: dbosCtx.applicationVersion,
-			executorID:         dbosCtx.executorID,
-			applicationID:      dbosCtx.applicationID,
+			ctx:                     context.WithValue(dbosCtx.ctx, key, val), // Spawn a new child context with the value set
+			logger:                  dbosCtx.logger,
+			systemDB:                dbosCtx.systemDB,
+			workflowsWg:             dbosCtx.workflowsWg,
+			workflowRegistry:        dbosCtx.workflowRegistry,
+			workflowCustomNametoFQN: dbosCtx.workflowCustomNametoFQN,
+			applicationVersion:      dbosCtx.applicationVersion,
+			executorID:              dbosCtx.executorID,
+			applicationID:           dbosCtx.applicationID,
 		}
 	}
 	return nil
@@ -181,15 +181,15 @@ func WithoutCancel(ctx DBOSContext) DBOSContext {
 	}
 	if dbosCtx, ok := ctx.(*dbosContext); ok {
 		return &dbosContext{
-			ctx:                context.WithoutCancel(dbosCtx.ctx),
-			logger:             dbosCtx.logger,
-			systemDB:           dbosCtx.systemDB,
-			workflowsWg:        dbosCtx.workflowsWg,
-			workflowRegistry:   dbosCtx.workflowRegistry,
-			workflowRegMutex:   dbosCtx.workflowRegMutex,
-			applicationVersion: dbosCtx.applicationVersion,
-			executorID:         dbosCtx.executorID,
-			applicationID:      dbosCtx.applicationID,
+			ctx:                     context.WithoutCancel(dbosCtx.ctx),
+			logger:                  dbosCtx.logger,
+			systemDB:                dbosCtx.systemDB,
+			workflowsWg:             dbosCtx.workflowsWg,
+			workflowRegistry:        dbosCtx.workflowRegistry,
+			workflowCustomNametoFQN: dbosCtx.workflowCustomNametoFQN,
+			applicationVersion:      dbosCtx.applicationVersion,
+			executorID:              dbosCtx.executorID,
+			applicationID:           dbosCtx.applicationID,
 		}
 	}
 	return nil
@@ -205,15 +205,15 @@ func WithTimeout(ctx DBOSContext, timeout time.Duration) (DBOSContext, context.C
 	if dbosCtx, ok := ctx.(*dbosContext); ok {
 		newCtx, cancelFunc := context.WithTimeoutCause(dbosCtx.ctx, timeout, errors.New("DBOS context timeout"))
 		return &dbosContext{
-			ctx:                newCtx,
-			logger:             dbosCtx.logger,
-			systemDB:           dbosCtx.systemDB,
-			workflowsWg:        dbosCtx.workflowsWg,
-			workflowRegistry:   dbosCtx.workflowRegistry,
-			workflowRegMutex:   dbosCtx.workflowRegMutex,
-			applicationVersion: dbosCtx.applicationVersion,
-			executorID:         dbosCtx.executorID,
-			applicationID:      dbosCtx.applicationID,
+			ctx:                     newCtx,
+			logger:                  dbosCtx.logger,
+			systemDB:                dbosCtx.systemDB,
+			workflowsWg:             dbosCtx.workflowsWg,
+			workflowRegistry:        dbosCtx.workflowRegistry,
+			workflowCustomNametoFQN: dbosCtx.workflowCustomNametoFQN,
+			applicationVersion:      dbosCtx.applicationVersion,
+			executorID:              dbosCtx.executorID,
+			applicationID:           dbosCtx.applicationID,
 		}, cancelFunc
 	}
 	return nil, func() {}
@@ -261,11 +261,11 @@ func (c *dbosContext) GetApplicationID() string {
 func NewDBOSContext(inputConfig Config) (DBOSContext, error) {
 	ctx, cancelFunc := context.WithCancelCause(context.Background())
 	initExecutor := &dbosContext{
-		workflowsWg:      &sync.WaitGroup{},
-		ctx:              ctx,
-		ctxCancelFunc:    cancelFunc,
-		workflowRegistry: make(map[string]workflowRegistryEntry),
-		workflowRegMutex: &sync.RWMutex{},
+		workflowsWg:             &sync.WaitGroup{},
+		ctx:                     ctx,
+		ctxCancelFunc:           cancelFunc,
+		workflowRegistry:        &sync.Map{},
+		workflowCustomNametoFQN: &sync.Map{},
 	}
 
 	// Load and process the configuration
@@ -438,7 +438,20 @@ func getBinaryHash() (string, error) {
 		return "", err
 	}
 
-	file, err := os.Open(execPath)
+	execPath, err = filepath.EvalSymlinks(execPath)
+	if err != nil {
+		return "", fmt.Errorf("resolve self path: %w", err)
+	}
+
+	fi, err := os.Lstat(execPath)
+	if err != nil {
+		return "", err
+	}
+	if !fi.Mode().IsRegular() {
+		return "", fmt.Errorf("executable is not a regular file")
+	}
+
+	file, err := os.Open(execPath) // #nosec G304 -- opening our own executable, not user-supplied
 	if err != nil {
 		return "", err
 	}