Merge pull request #278 from cschleiden/explicitly-remove-workflow-from-cache

Explicitly evict finished workflow instances from cache

Author: cschleiden

backend/test/e2e.go

@@ -724,6 +724,11 @@ func (*noopWorkflowExecutorCache) Get(ctx context.Context, instance *core.Workfl
 	return nil, false, nil
 }
 
+// Evict implements workflow.ExecutorCache
+func (*noopWorkflowExecutorCache) Evict(ctx context.Context, instance *core.WorkflowInstance) error {
+	return nil
+}
+
 // StartEviction implements workflow.ExecutorCache
 func (*noopWorkflowExecutorCache) StartEviction(ctx context.Context) {
 }

internal/worker/workflow.go

@@ -59,6 +59,13 @@ func (wtw *WorkflowTaskWorker) Complete(ctx context.Context, result *workflow.Ex
 				metrickeys.ContinuedAsNew: fmt.Sprint(state == core.WorkflowInstanceStateContinuedAsNew),
 			}, 1)
 		}
+
+		// Workflow is finished, explicitly evict from cache (if one is used)
+		if wtw.cache != nil {
+			if err := wtw.cache.Evict(ctx, t.WorkflowInstance); err != nil {
+				wtw.logger.ErrorContext(ctx, "could not evict workflow executor from cache", "error", err)
+			}
+		}
 	}
 
 	wtw.backend.Metrics().Counter(metrickeys.ActivityTaskScheduled, metrics.Tags{}, int64(len(result.ActivityEvents)))

internal/workflow/cache.go

@@ -8,6 +8,7 @@ import (
 
 type ExecutorCache interface {
 	Store(ctx context.Context, instance *core.WorkflowInstance, workflow WorkflowExecutor) error
+	Evict(ctx context.Context, instance *core.WorkflowInstance) error
 	Get(ctx context.Context, instance *core.WorkflowInstance) (WorkflowExecutor, bool, error)
 	StartEviction(ctx context.Context)
 }

internal/workflow/cache/cache.go

@@ -60,6 +60,14 @@ func (lc *LruCache) Store(ctx context.Context, instance *core.WorkflowInstance,
 	return nil
 }
 
+func (lc *LruCache) Evict(ctx context.Context, instance *core.WorkflowInstance) error {
+	lc.c.Delete(getKey(instance))
+
+	lc.mc.Gauge(metrickeys.WorkflowInstanceCacheSize, metrics.Tags{}, int64(lc.c.Len()))
+
+	return nil
+}
+
 func (lc *LruCache) StartEviction(ctx context.Context) {
 	go lc.c.Start()