optimize poller behavior, remove BlockingBackend

Author: lovromazgon

backend/redis/redis.go

@@ -12,7 +12,6 @@ import (
 	"github.com/cschleiden/go-workflows/internal/core"
 	"github.com/cschleiden/go-workflows/internal/history"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
-	"github.com/cschleiden/go-workflows/internal/worker"
 	"github.com/cschleiden/go-workflows/metrics"
 	"github.com/redis/go-redis/v9"
 	"go.opentelemetry.io/otel/trace"
@@ -110,7 +109,3 @@ func (rb *redisBackend) ContextPropagators() []contextpropagation.ContextPropaga
 func (rb *redisBackend) Close() error {
 	return rb.rdb.Close()
 }
-
-var _ worker.BlockingBackend = (*redisBackend)(nil)
-
-func (rb *redisBackend) BlockOnGetTask() { /* satisfy interface */ }

internal/worker/activity.go

@@ -74,38 +74,22 @@ func (aw *ActivityWorker) WaitForCompletion() error {
 func (aw *ActivityWorker) runPoll(ctx context.Context) {
 	defer aw.pollersWg.Done()
 
-	poll := func() {
+	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)
-			return
 		}
-
 		if task != nil {
 			aw.activityTaskQueue <- task
+			continue // check for new tasks right away
 		}
-	}
 
-	if _, ok := aw.backend.(BlockingBackend); ok {
-		// Backend blocks on calls to GetActivityTask, poll continuously
-		for {
-			if ctx.Err() != nil {
-				return
-			}
-			poll()
-		}
-	}
-
-	// Backend is polling tasks and returns on calls to GetActivityTask, we need
-	// to throttle the calls.
-	ticker := time.NewTicker(aw.options.ActivityPollingInterval)
-	defer ticker.Stop()
-	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			poll()
 		}
 	}
 }
@@ -214,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/backend.go

@@ -84,39 +84,23 @@ func (ww *WorkflowWorker) WaitForCompletion() error {
 func (ww *WorkflowWorker) runPoll(ctx context.Context) {
 	defer ww.pollersWg.Done()
 
-	poll := func() {
+	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)
-			return
 		}
-
 		if task != nil {
 			ww.wg.Add(1)
 			ww.workflowTaskQueue <- task
+			continue // check for new tasks right away
 		}
-	}
 
-	if _, ok := ww.backend.(BlockingBackend); ok {
-		// Backend blocks on calls to GetActivityTask, poll continuously
-		for {
-			if ctx.Err() != nil {
-				return
-			}
-			poll()
-		}
-	}
-
-	// Backend is polling tasks and returns on calls to GetActivityTask, we need
-	// to throttle the calls.
-	ticker := time.NewTicker(ww.options.WorkflowPollingInterval)
-	defer ticker.Stop()
-	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			poll()
 		}
 	}
 }
@@ -274,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
 		}