Bag o stuff

.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
 	}