fix global concurrency

Author: maxdml

dbos/queue.go

@@ -14,9 +14,13 @@ import (
 )
 
 var (
-	workflowQueueRegistry    = make(map[string]WorkflowQueue)
-	DBOS_INTERNAL_QUEUE_NAME = "_dbos_internal_queue"
-	_                        = NewWorkflowQueue(DBOS_INTERNAL_QUEUE_NAME)
+	workflowQueueRegistry = make(map[string]WorkflowQueue)
+	_                     = NewWorkflowQueue(_DBOS_INTERNAL_QUEUE_NAME)
+)
+
+const (
+	_DBOS_INTERNAL_QUEUE_NAME        = "_dbos_internal_queue"
+	_DEFAULT_MAX_TASKS_PER_ITERATION = 100
 )
 
 // RateLimiter represents a rate limiting configuration
@@ -31,7 +35,7 @@ type WorkflowQueue struct {
 	GlobalConcurrency    *int
 	PriorityEnabled      bool
 	Limiter              *RateLimiter
-	MaxTasksPerIteration uint
+	MaxTasksPerIteration int
 }
 
 // QueueOption is a functional option for configuring a workflow queue
@@ -61,7 +65,7 @@ func WithRateLimiter(limiter *RateLimiter) QueueOption {
 	}
 }
 
-func WithMaxTasksPerIteration(maxTasks uint) QueueOption {
+func WithMaxTasksPerIteration(maxTasks int) QueueOption {
 	return func(q *WorkflowQueue) {
 		q.MaxTasksPerIteration = maxTasks
 	}
@@ -84,7 +88,7 @@ func NewWorkflowQueue(name string, options ...QueueOption) WorkflowQueue {
 		GlobalConcurrency:    nil,
 		PriorityEnabled:      false,
 		Limiter:              nil,
-		MaxTasksPerIteration: 100, // Default max tasks per iteration
+		MaxTasksPerIteration: _DEFAULT_MAX_TASKS_PER_ITERATION,
 	}
 
 	// Apply functional options

dbos/queues_test.go

@@ -405,6 +405,7 @@ func TestGlobalConcurrency(t *testing.T) {
 
 	// Wait for the first workflow to start
 	workflowEvent1.Wait()
+	time.Sleep(2 * time.Second) // Wait for a few seconds to let the queue runner loop
 
 	// Ensure the second workflow has not started yet
 	if workflowEvent2.IsSet {

dbos/system_database.go

@@ -1321,7 +1321,7 @@ func (s *systemDatabase) DequeueWorkflows(ctx context.Context, queue WorkflowQue
 				getLogger().Warn("Local pending workflows on queue exceeds worker concurrency limit", "local_pending", localPendingWorkflows, "queue_name", queue.Name, "concurrency_limit", workerConcurrency)
 			}
 			availableWorkerTasks := max(workerConcurrency-localPendingWorkflows, 0)
-			maxTasks = uint(availableWorkerTasks)
+			maxTasks = availableWorkerTasks
 		}
 
 		// Check global concurrency limit
@@ -1336,8 +1336,8 @@ func (s *systemDatabase) DequeueWorkflows(ctx context.Context, queue WorkflowQue
 				getLogger().Warn("Total pending workflows on queue exceeds global concurrency limit", "total_pending", globalPendingWorkflows, "queue_name", queue.Name, "concurrency_limit", concurrency)
 			}
 			availableTasks := max(concurrency-globalPendingWorkflows, 0)
-			if uint(availableTasks) < maxTasks {
-				maxTasks = uint(availableTasks)
+			if availableTasks < maxTasks {
+				maxTasks = availableTasks
 			}
 		}
 	}
@@ -1375,7 +1375,7 @@ func (s *systemDatabase) DequeueWorkflows(ctx context.Context, queue WorkflowQue
 	}
 
 	// Add limit if maxTasks is finite
-	if maxTasks > 0 {
+	if maxTasks >= 0 {
 		query += fmt.Sprintf(" LIMIT %d", int(maxTasks))
 	}
 

dbos/utils_test.go

@@ -153,7 +153,7 @@ func queueEntriesAreCleanedUp() bool {
 					AND status IN ('ENQUEUED', 'PENDING')`
 
 		var count int
-		err = tx.QueryRow(context.Background(), query, DBOS_INTERNAL_QUEUE_NAME).Scan(&count)
+		err = tx.QueryRow(context.Background(), query, _DBOS_INTERNAL_QUEUE_NAME).Scan(&count)
 		tx.Rollback(context.Background()) // Clean up transaction
 
 		if err != nil {

dbos/workflow.go

@@ -301,7 +301,7 @@ func WithWorkflow[P any, R any](fn WorkflowFunc[P, R], opts ...WorkflowRegistrat
 				scheduledTime = entry.Next
 			}
 			wfID := fmt.Sprintf("sched-%s-%s", fqn, scheduledTime) // XXX we can rethink the format
-			wrappedFunction(context.Background(), any(scheduledTime).(P), WithWorkflowID(wfID), WithQueue(DBOS_INTERNAL_QUEUE_NAME))
+			wrappedFunction(context.Background(), any(scheduledTime).(P), WithWorkflowID(wfID), WithQueue(_DBOS_INTERNAL_QUEUE_NAME))
 		})
 		if err != nil {
 			panic(fmt.Sprintf("failed to register scheduled workflow: %v", err))