document monoprocess backend

Author: lovromazgon

backend/monoprocess/monoprocess.go

@@ -23,9 +23,15 @@ type monoprocessBackend struct {
 	logger *slog.Logger
 }
 
+// NewMonoprocessBackend wraps an existing backend and improves its responsiveness
+// in case the backend and worker are running in the same process. This backend
+// uses channels to notify the worker every time there is a new task ready to be
+// worked on. Note that only one worker will be notified.
+// IMPORTANT: Only use this backend if the backend and worker are running in the
+// same process.
 func NewMonoprocessBackend(b backend.Backend, signalBufferSize int, signalTimeout time.Duration) *monoprocessBackend {
 	if signalTimeout <= 0 {
-		signalTimeout = time.Second
+		signalTimeout = time.Second // default
 	}
 	mb := &monoprocessBackend{
 		Backend:        b,
@@ -41,12 +47,12 @@ func (b *monoprocessBackend) GetWorkflowTask(ctx context.Context) (*task.Workflo
 	if w, err := b.Backend.GetWorkflowTask(ctx); w != nil || err != nil {
 		return w, err
 	}
-	b.logger.Debug("worker waiting for workflow task signal")
+	b.logger.DebugContext(ctx, "worker waiting for workflow task signal")
 	select {
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	case <-b.workflowSignal:
-		b.logger.Debug("worker got a workflow task signal")
+		b.logger.DebugContext(ctx, "worker got a workflow task signal")
 		return b.GetWorkflowTask(ctx)
 	}
 }
@@ -55,12 +61,12 @@ func (b *monoprocessBackend) GetActivityTask(ctx context.Context) (*task.Activit
 	if a, err := b.Backend.GetActivityTask(ctx); a != nil || err != nil {
 		return a, err
 	}
-	b.logger.Debug("worker waiting for activity task signal")
+	b.logger.DebugContext(ctx, "worker waiting for activity task signal")
 	select {
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	case <-b.activitySignal:
-		b.logger.Debug("worker got an activity task signal")
+		b.logger.DebugContext(ctx, "worker got an activity task signal")
 		return b.GetActivityTask(ctx)
 	}
 }
@@ -89,7 +95,7 @@ func (b *monoprocessBackend) CompleteWorkflowTask(
 			continue
 		}
 		if !b.notifyActivityWorker(ctx) {
-			break // no reason to notify more
+			break // no reason to notify more, queue is full
 		}
 	}
 	for _, e := range timerEvents {
@@ -98,7 +104,7 @@ func (b *monoprocessBackend) CompleteWorkflowTask(
 			b.logger.Warn("unknown attributes type in timer event", "type", reflect.TypeOf(e.Attributes).String())
 			continue
 		}
-		b.logger.Debug("scheduling timer to notify workflow worker")
+		b.logger.DebugContext(ctx, "scheduling timer to notify workflow worker")
 		time.AfterFunc(attr.At.Sub(time.Now()), func() { b.notifyWorkflowWorker(ctx) }) // TODO: cancel timer if the event gets cancelled
 	}
 	for _, e := range workflowEvents {
@@ -108,7 +114,7 @@ func (b *monoprocessBackend) CompleteWorkflowTask(
 			continue
 		}
 		if !b.notifyWorkflowWorker(ctx) {
-			break // no reason to notify more
+			break // no reason to notify more, queue is full
 		}
 	}
 	return nil
@@ -139,35 +145,31 @@ func (b *monoprocessBackend) SignalWorkflow(ctx context.Context, instanceID stri
 }
 
 func (b *monoprocessBackend) notifyActivityWorker(ctx context.Context) bool {
+	ctx, cancel := context.WithTimeout(ctx, b.signalTimeout)
+	defer cancel()
 	select {
 	case <-ctx.Done():
 		// we didn't manage to notify the worker that there is a new task, it
 		// will pick it up after the poll timeout
-		b.logger.Debug("failed to signal activity task to worker, context cancelled")
-	case <-time.After(b.signalTimeout):
-		// we didn't manage to notify the worker that there is a new task, it
-		// will pick it up after the poll timeout
-		b.logger.Debug("failed to signal activity task to worker, timeout")
+		b.logger.DebugContext(ctx, "failed to signal activity task to worker", "reason", ctx.Err())
+		return false
 	case b.activitySignal <- struct{}{}:
-		b.logger.Debug("signalled a new activity task to worker")
+		b.logger.DebugContext(ctx, "signalled a new activity task to worker")
 		return true
 	}
-	return false
 }
 
 func (b *monoprocessBackend) notifyWorkflowWorker(ctx context.Context) bool {
+	ctx, cancel := context.WithTimeout(ctx, b.signalTimeout)
+	defer cancel()
 	select {
 	case <-ctx.Done():
 		// we didn't manage to notify the worker that there is a new task, it
 		// will pick it up after the poll timeout
-		b.logger.Debug("failed to signal workflow task to worker, context cancelled")
-	case <-time.After(b.signalTimeout):
-		// we didn't manage to notify the worker that there is a new task, it
-		// will pick it up after the poll timeout
-		b.logger.Debug("failed to signal workflow task to worker, timeout")
+		b.logger.DebugContext(ctx, "failed to signal workflow task to worker", "reason", ctx.Err())
+		return false
 	case b.workflowSignal <- struct{}{}:
-		b.logger.Debug("signalled a new workflow task to worker")
+		b.logger.DebugContext(ctx, "signalled a new workflow task to worker")
 		return true
 	}
-	return false
 }