Merge pull request #239 from lovromazgon/polling-interval

Make polling interval configurable

Author: cschleiden

backend/backend.go

@@ -44,7 +44,7 @@ type Backend interface {
 	// If the given instance does not exist, it will return an error
 	SignalWorkflow(ctx context.Context, instanceID string, event *history.Event) error
 
-	// GetWorkflowInstance returns a pending workflow task or nil if there are no pending worflow executions
+	// GetWorkflowTask returns a pending workflow task or nil if there are no pending workflow executions
 	GetWorkflowTask(ctx context.Context) (*task.Workflow, error)
 
 	// ExtendWorkflowTask extends the lock of a workflow task

internal/worker/activity.go

@@ -3,7 +3,7 @@ package worker
 import (
 	"context"
 	"errors"
-	"log"
+	"log/slog"
 	"sync"
 	"time"
 
@@ -30,7 +30,8 @@ type ActivityWorker struct {
 	wg        sync.WaitGroup
 	pollersWg sync.WaitGroup
 
-	clock clock.Clock
+	clock  clock.Clock
+	logger *slog.Logger
 }
 
 func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clock clock.Clock, options *Options) *ActivityWorker {
@@ -42,7 +43,8 @@ func NewActivityWorker(backend backend.Backend, registry *workflow.Registry, clo
 		activityTaskQueue:    make(chan *task.Activity),
 		activityTaskExecutor: activity.NewExecutor(backend.Logger(), backend.Tracer(), backend.Converter(), backend.ContextPropagators(), registry),
 
-		clock: clock,
+		clock:  clock,
+		logger: backend.Logger(),
 	}
 }
 
@@ -53,7 +55,7 @@ func (aw *ActivityWorker) Start(ctx context.Context) error {
 		go aw.runPoll(ctx)
 	}
 
-	go aw.runDispatcher(context.Background())
+	go aw.runDispatcher()
 
 	return nil
 }
@@ -72,25 +74,28 @@ func (aw *ActivityWorker) WaitForCompletion() error {
 func (aw *ActivityWorker) runPoll(ctx context.Context) {
 	defer aw.pollersWg.Done()
 
+	ticker := time.NewTicker(aw.options.ActivityPollingInterval)
+	defer ticker.Stop()
 	for {
+		task, err := aw.poll(ctx, 30*time.Second)
+		if err != nil {
+			aw.logger.ErrorContext(ctx, "error while polling for activity task", "error", err)
+		}
+		if task != nil {
+			aw.wg.Add(1)
+			aw.activityTaskQueue <- task
+			continue // check for new tasks right away
+		}
+
 		select {
 		case <-ctx.Done():
 			return
-		default:
-			task, err := aw.poll(ctx, 30*time.Second)
-			if err != nil {
-				log.Println("error while polling for activity task:", err)
-				continue
-			}
-
-			if task != nil {
-				aw.activityTaskQueue <- task
-			}
+		case <-ticker.C:
 		}
 	}
 }
 
-func (aw *ActivityWorker) runDispatcher(ctx context.Context) {
+func (aw *ActivityWorker) runDispatcher() {
 	var sem chan struct{}
 	if aw.options.MaxParallelActivityTasks > 0 {
 		sem = make(chan struct{}, aw.options.MaxParallelActivityTasks)
@@ -103,7 +108,6 @@ func (aw *ActivityWorker) runDispatcher(ctx context.Context) {
 
 		task := task
 
-		aw.wg.Add(1)
 		go func() {
 			defer aw.wg.Done()
 
@@ -194,7 +198,7 @@ func (aw *ActivityWorker) poll(ctx context.Context, timeout time.Duration) (*tas
 
 	task, err := aw.backend.GetActivityTask(ctx)
 	if err != nil {
-		if errors.Is(err, context.Canceled) {
+		if errors.Is(err, context.DeadlineExceeded) {
 			return nil, nil
 		}
 	}

internal/worker/options.go

@@ -33,6 +33,16 @@ type Options struct {
 	// WorkflowHeartbeatInterval is the interval between heartbeat attempts on workflow tasks, when enabled.
 	WorkflowHeartbeatInterval time.Duration
 
+	// WorkflowPollingInterval is the interval between polling for new workflow tasks.
+	// Note that if you use a backend that can wait for tasks to be available (e.g. redis) this field has no effect.
+	// Defaults to 200ms.
+	WorkflowPollingInterval time.Duration
+
+	// ActivityPollingInterval is the interval between polling for new activity tasks.
+	// Note that if you use a backend that can wait for tasks to be available (e.g. redis) this field has no effect.
+	// Defaults to 200ms.
+	ActivityPollingInterval time.Duration
+
 	// WorkflowExecutorCache is the max size of the workflow executor cache. Defaults to 128
 	WorkflowExecutorCacheSize int
 
@@ -51,6 +61,8 @@ var DefaultOptions = Options{
 	MaxParallelActivityTasks:  0,
 	ActivityHeartbeatInterval: 25 * time.Second,
 	WorkflowHeartbeatInterval: 25 * time.Second,
+	WorkflowPollingInterval:   200 * time.Millisecond,
+	ActivityPollingInterval:   200 * time.Millisecond,
 
 	WorkflowExecutorCacheSize: 128,
 	WorkflowExecutorCacheTTL:  time.Second * 10,

internal/worker/workflow.go

@@ -84,28 +84,29 @@ func (ww *WorkflowWorker) WaitForCompletion() error {
 func (ww *WorkflowWorker) runPoll(ctx context.Context) {
 	defer ww.pollersWg.Done()
 
+	ticker := time.NewTicker(ww.options.WorkflowPollingInterval)
+	defer ticker.Stop()
 	for {
+		task, err := ww.poll(ctx, 30*time.Second)
+		if err != nil {
+			ww.logger.ErrorContext(ctx, "error while polling for workflow task", "error", err)
+		}
+		if task != nil {
+			ww.wg.Add(1)
+			ww.workflowTaskQueue <- task
+			continue // check for new tasks right away
+		}
+
 		select {
 		case <-ctx.Done():
 			return
-
-		default:
-			task, err := ww.poll(ctx, 30*time.Second)
-			if err != nil {
-				ww.logger.Error("error while polling for workflow task", "error", err)
-				continue
-			}
-
-			if task != nil {
-				ww.wg.Add(1)
-				ww.workflowTaskQueue <- task
-			}
+		case <-ticker.C:
 		}
 	}
 }
 
 func (ww *WorkflowWorker) runDispatcher() {
-	var sem chan (struct{})
+	var sem chan struct{}
 
 	if ww.options.MaxParallelWorkflowTasks > 0 {
 		sem = make(chan struct{}, ww.options.MaxParallelWorkflowTasks)
@@ -177,7 +178,7 @@ func (ww *WorkflowWorker) handle(ctx context.Context, t *task.Workflow) {
 
 	if err := ww.backend.CompleteWorkflowTask(
 		ctx, t, t.WorkflowInstance, state, result.Executed, result.ActivityEvents, result.TimerEvents, result.WorkflowEvents); err != nil {
-		ww.logger.Error("could not complete workflow task", "error", err)
+		ww.logger.ErrorContext(ctx, "could not complete workflow task", "error", err)
 		panic("could not complete workflow task")
 	}
 }
@@ -210,7 +211,7 @@ func (ww *WorkflowWorker) getExecutor(ctx context.Context, t *task.Workflow) (wo
 	// Try to get a cached executor
 	executor, ok, err := ww.cache.Get(ctx, t.WorkflowInstance)
 	if err != nil {
-		ww.logger.Error("could not get cached workflow task executor", "error", err)
+		ww.logger.ErrorContext(ctx, "could not get cached workflow task executor", "error", err)
 	}
 
 	if !ok {
@@ -224,7 +225,7 @@ func (ww *WorkflowWorker) getExecutor(ctx context.Context, t *task.Workflow) (wo
 
 	// Cache executor instance for future continuation tasks, or refresh last access time
 	if err := ww.cache.Store(ctx, t.WorkflowInstance, executor); err != nil {
-		ww.logger.Error("error while caching workflow task executor:", "error", err)
+		ww.logger.ErrorContext(ctx, "error while caching workflow task executor:", "error", err)
 	}
 
 	return executor, nil
@@ -240,7 +241,7 @@ func (ww *WorkflowWorker) heartbeatTask(ctx context.Context, task *task.Workflow
 			return
 		case <-t.C:
 			if err := ww.backend.ExtendWorkflowTask(ctx, task.ID, task.WorkflowInstance); err != nil {
-				ww.logger.Error("could not heartbeat workflow task", "error", err)
+				ww.logger.ErrorContext(ctx, "could not heartbeat workflow task", "error", err)
 				panic("could not heartbeat workflow task")
 			}
 		}
@@ -257,7 +258,7 @@ func (ww *WorkflowWorker) poll(ctx context.Context, timeout time.Duration) (*tas
 
 	task, err := ww.backend.GetWorkflowTask(ctx)
 	if err != nil {
-		if errors.Is(err, context.Canceled) {
+		if errors.Is(err, context.DeadlineExceeded) {
 			return nil, nil
 		}