RateLimiter.Period is now a time.Duration (#124)

Author: maxdml

dbos/queue.go

@@ -22,8 +22,8 @@ const (
 // RateLimiter configures rate limiting for workflow queue execution.
 // Rate limits prevent overwhelming external services and provide backpressure.
 type RateLimiter struct {
-	Limit  int     // Maximum number of workflows to start within the period
-	Period float64 // Time period in seconds for the rate limit
+	Limit  int           // Maximum number of workflows to start within the period
+	Period time.Duration // Time period for the rate limit
 }
 
 // WorkflowQueue defines a named queue for workflow execution.
@@ -90,7 +90,7 @@ func WithMaxTasksPerIteration(maxTasks int) QueueOption {
 //	    dbos.WithWorkerConcurrency(5),
 //	    dbos.WithRateLimiter(&dbos.RateLimiter{
 //	        Limit:  100,
-//	        Period: 60.0, // 100 workflows per minute
+//	        Period: 60 * time.Second, // 100 workflows per minute
 //	    }),
 //	    dbos.WithPriorityEnabled(),
 //	)

dbos/queues_test.go

@@ -833,7 +833,7 @@ func rateLimiterTestWorkflow(ctx DBOSContext, _ string) (time.Time, error) {
 func TestQueueRateLimiter(t *testing.T) {
 	dbosCtx := setupDBOS(t, true, true)
 
-	rateLimiterQueue := NewWorkflowQueue(dbosCtx, "test-rate-limiter-queue", WithRateLimiter(&RateLimiter{Limit: 5, Period: 1.8}))
+	rateLimiterQueue := NewWorkflowQueue(dbosCtx, "test-rate-limiter-queue", WithRateLimiter(&RateLimiter{Limit: 5, Period: time.Duration(1800 * time.Millisecond)}))
 
 	// Create workflow with dbosContext
 	RegisterWorkflow(dbosCtx, rateLimiterTestWorkflow)
@@ -842,7 +842,7 @@ func TestQueueRateLimiter(t *testing.T) {
 	require.NoError(t, err, "failed to launch DBOS instance")
 
 	limit := 5
-	period := 1.8
+	periodSeconds := 1.8
 	numWaves := 3
 
 	var handles []WorkflowHandle[time.Time]
@@ -889,7 +889,7 @@ func TestQueueRateLimiter(t *testing.T) {
 	// Group workflows into waves based on their start time
 	for _, workflowTime := range sortedTimes {
 		timeSinceBase := workflowTime.Sub(baseTime).Seconds()
-		waveIndex := int(timeSinceBase / period)
+		waveIndex := int(timeSinceBase / periodSeconds)
 		waveMap[waveIndex] = append(waveMap[waveIndex], workflowTime)
 	}
 	// Verify each wave has fewer than the limit

dbos/system_database.go

@@ -2093,10 +2093,8 @@ func (s *sysDB) dequeueWorkflows(ctx context.Context, input dequeueWorkflowsInpu
 	// First check the rate limiter
 	var numRecentQueries int
 	if input.queue.RateLimit != nil {
-		limiterPeriod := time.Duration(input.queue.RateLimit.Period * float64(time.Second))
-
 		// Calculate the cutoff time: current time minus limiter period
-		cutoffTimeMs := time.Now().Add(-limiterPeriod).UnixMilli()
+		cutoffTimeMs := time.Now().Add(-input.queue.RateLimit.Period).UnixMilli()
 
 		// Count workflows that have started in the limiter period
 		limiterQuery := `