Merge pull request #313 from cschleiden/start-cache-eviction

Start cache eviction

Author: cschleiden

internal/worker/activity.go

@@ -50,6 +50,10 @@ func (atw *ActivityTaskWorker) Complete(ctx context.Context, event *history.Even
 	return nil
 }
 
+func (atw *ActivityTaskWorker) Start(ctx context.Context) error {
+	return nil
+}
+
 func (atw *ActivityTaskWorker) Execute(ctx context.Context, task *backend.ActivityTask) (*history.Event, error) {
 	a := task.Event.Attributes.(*history.ActivityScheduledAttributes)
 	ametrics := atw.backend.Metrics().WithTags(metrics.Tags{metrickeys.ActivityName: a.Name})

internal/worker/worker.go

@@ -12,6 +12,7 @@ import (
 )
 
 type TaskWorker[Task, Result any] interface {
+	Start(context.Context) error
 	Get(context.Context) (*Task, error)
 	Extend(context.Context, *Task) error
 	Execute(context.Context, *Task) (*Result, error)
@@ -55,6 +56,10 @@ func NewWorker[Task, TaskResult any](
 }
 
 func (w *Worker[Task, TaskResult]) Start(ctx context.Context) error {
+	if err := w.tw.Start(ctx); err != nil {
+		return fmt.Errorf("starting task worker: %w", err)
+	}
+
 	w.pollersWg.Add(w.options.Pollers)
 
 	for i := 0; i < w.options.Pollers; i++ {

internal/worker/workflow.go

@@ -53,6 +53,14 @@ type WorkflowTaskWorker struct {
 	logger   *slog.Logger
 }
 
+func (wtw *WorkflowTaskWorker) Start(ctx context.Context) error {
+	if wtw.cache != nil {
+		go wtw.cache.StartEviction(ctx)
+	}
+
+	return nil
+}
+
 // Complete implements TaskWorker.
 func (wtw *WorkflowTaskWorker) Complete(ctx context.Context, result *workflow.ExecutionResult, t *backend.WorkflowTask) error {
 	state := result.State