Enable other cache implementations

backend/test/e2e.go

@@ -13,9 +13,9 @@ import (
 	"github.com/cschleiden/go-workflows/client"
 	"github.com/cschleiden/go-workflows/core"
 	"github.com/cschleiden/go-workflows/internal/sync"
-	internalwf "github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/worker"
 	"github.com/cschleiden/go-workflows/workflow"
+	"github.com/cschleiden/go-workflows/workflow/executor"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 )
@@ -834,10 +834,10 @@ func EndToEndBackendTest(t *testing.T, setup func(options ...backend.BackendOpti
 type noopWorkflowExecutorCache struct {
 }
 
-var _ internalwf.ExecutorCache = (*noopWorkflowExecutorCache)(nil)
+var _ executor.Cache = (*noopWorkflowExecutorCache)(nil)
 
 // Get implements workflow.ExecutorCache
-func (*noopWorkflowExecutorCache) Get(ctx context.Context, instance *core.WorkflowInstance) (internalwf.WorkflowExecutor, bool, error) {
+func (*noopWorkflowExecutorCache) Get(ctx context.Context, instance *core.WorkflowInstance) (executor.WorkflowExecutor, bool, error) {
 	return nil, false, nil
 }
 
@@ -851,7 +851,7 @@ func (*noopWorkflowExecutorCache) StartEviction(ctx context.Context) {
 }
 
 // Store implements workflow.ExecutorCache
-func (*noopWorkflowExecutorCache) Store(ctx context.Context, instance *core.WorkflowInstance, workflow internalwf.WorkflowExecutor) error {
+func (*noopWorkflowExecutorCache) Store(ctx context.Context, instance *core.WorkflowInstance, workflow executor.WorkflowExecutor) error {
 	return nil
 }
 

go.mod

@@ -17,6 +17,7 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.16.0
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.16.0
 	go.opentelemetry.io/otel/trace v1.16.0
+	go.uber.org/goleak v1.3.0
 	golang.org/x/tools v0.12.0
 	modernc.org/sqlite v1.27.0
 )
@@ -57,7 +58,6 @@ require (
 	go.opentelemetry.io/proto/otlp v0.19.0 // indirect
 	go.tmz.dev/musttag v0.7.2 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
-	go.uber.org/goleak v1.3.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
 	golang.org/x/exp v0.0.0-20230510235704-dd950f8aeaea // indirect

go.sum

@@ -672,8 +672,6 @@ go.tmz.dev/musttag v0.7.2 h1:1J6S9ipDbalBSODNT5jCep8dhZyMr4ttnjQagmGYR5s=
 go.tmz.dev/musttag v0.7.2/go.mod h1:m6q5NiiSKMnQYokefa2xGoyoXnrswCbJ0AWYzf4Zs28=
 go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
-go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
-go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=

internal/worker/workflow.go

@@ -15,15 +15,15 @@ import (
 	"github.com/cschleiden/go-workflows/internal/log"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
 	im "github.com/cschleiden/go-workflows/internal/metrics"
-	"github.com/cschleiden/go-workflows/internal/workflow"
-	"github.com/cschleiden/go-workflows/internal/workflow/cache"
 	"github.com/cschleiden/go-workflows/registry"
+	"github.com/cschleiden/go-workflows/workflow/executor"
+	"github.com/cschleiden/go-workflows/workflow/executor/cache"
 )
 
 type WorkflowWorkerOptions struct {
 	WorkerOptions
 
-	WorkflowExecutorCache     workflow.ExecutorCache
+	WorkflowExecutorCache     executor.Cache
 	WorkflowExecutorCacheSize int
 	WorkflowExecutorCacheTTL  time.Duration
 }
@@ -32,7 +32,7 @@ func NewWorkflowWorker(
 	b backend.Backend,
 	registry *registry.Registry,
 	options WorkflowWorkerOptions,
-) *Worker[backend.WorkflowTask, workflow.ExecutionResult] {
+) *Worker[backend.WorkflowTask, executor.ExecutionResult] {
 	if options.WorkflowExecutorCache == nil {
 		options.WorkflowExecutorCache = cache.NewWorkflowExecutorLRUCache(b.Metrics(), options.WorkflowExecutorCacheSize, options.WorkflowExecutorCacheTTL)
 	}
@@ -44,13 +44,13 @@ func NewWorkflowWorker(
 		logger:   b.Logger(),
 	}
 
-	return NewWorker[backend.WorkflowTask, workflow.ExecutionResult](b, tw, &options.WorkerOptions)
+	return NewWorker[backend.WorkflowTask, executor.ExecutionResult](b, tw, &options.WorkerOptions)
 }
 
 type WorkflowTaskWorker struct {
 	backend  backend.Backend
 	registry *registry.Registry
-	cache    workflow.ExecutorCache
+	cache    executor.Cache
 	logger   *slog.Logger
 }
 
@@ -63,7 +63,7 @@ func (wtw *WorkflowTaskWorker) Start(ctx context.Context) error {
 }
 
 // Complete implements TaskWorker.
-func (wtw *WorkflowTaskWorker) Complete(ctx context.Context, result *workflow.ExecutionResult, t *backend.WorkflowTask) error {
+func (wtw *WorkflowTaskWorker) Complete(ctx context.Context, result *executor.ExecutionResult, t *backend.WorkflowTask) error {
 	logger := wtw.logger.With(
 		slog.String(log.TaskIDKey, t.ID),
 		slog.String(log.InstanceIDKey, t.WorkflowInstance.InstanceID),
@@ -99,7 +99,7 @@ func (wtw *WorkflowTaskWorker) Complete(ctx context.Context, result *workflow.Ex
 	return nil
 }
 
-func (wtw *WorkflowTaskWorker) Execute(ctx context.Context, t *backend.WorkflowTask) (*workflow.ExecutionResult, error) {
+func (wtw *WorkflowTaskWorker) Execute(ctx context.Context, t *backend.WorkflowTask) (*executor.ExecutionResult, error) {
 	// Record how long this task was in the queue
 	firstEvent := t.NewEvents[0]
 	var scheduledAt time.Time
@@ -154,15 +154,15 @@ func (wtw *WorkflowTaskWorker) Get(ctx context.Context) (*backend.WorkflowTask,
 	return t, nil
 }
 
-func (wtw *WorkflowTaskWorker) getExecutor(ctx context.Context, t *backend.WorkflowTask) (workflow.WorkflowExecutor, error) {
+func (wtw *WorkflowTaskWorker) getExecutor(ctx context.Context, t *backend.WorkflowTask) (executor.WorkflowExecutor, error) {
 	// Try to get a cached executor
-	executor, ok, err := wtw.cache.Get(ctx, t.WorkflowInstance)
+	e, ok, err := wtw.cache.Get(ctx, t.WorkflowInstance)
 	if err != nil {
 		wtw.logger.ErrorContext(ctx, "could not get cached workflow task executor", "error", err)
 	}
 
 	if !ok {
-		executor, err = workflow.NewExecutor(
+		e, err = executor.NewExecutor(
 			wtw.logger.With(
 				slog.String(log.InstanceIDKey, t.WorkflowInstance.InstanceID),
 				slog.String(log.ExecutionIDKey, t.WorkflowInstance.ExecutionID),
@@ -182,9 +182,9 @@ func (wtw *WorkflowTaskWorker) getExecutor(ctx context.Context, t *backend.Workf
 	}
 
 	// Cache executor instance for future continuation tasks, or refresh last access time
-	if err := wtw.cache.Store(ctx, t.WorkflowInstance, executor); err != nil {
+	if err := wtw.cache.Store(ctx, t.WorkflowInstance, e); err != nil {
 		wtw.logger.ErrorContext(ctx, "error while caching workflow task executor:", "error", err)
 	}
 
-	return executor, nil
+	return e, nil
 }

tester/tester.go

@@ -24,10 +24,10 @@ import (
 	"github.com/cschleiden/go-workflows/internal/fn"
 	"github.com/cschleiden/go-workflows/internal/log"
 	"github.com/cschleiden/go-workflows/internal/signals"
-	wf "github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/internal/workflowerrors"
 	"github.com/cschleiden/go-workflows/registry"
 	"github.com/cschleiden/go-workflows/workflow"
+	"github.com/cschleiden/go-workflows/workflow/executor"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/mock"
 	"go.opentelemetry.io/otel/trace"
@@ -346,7 +346,7 @@ func (wt *workflowTester[TResult]) Execute(ctx context.Context, args ...any) {
 			tw.pendingEvents = tw.pendingEvents[:0]
 
 			// Execute task
-			e, err := wf.NewExecutor(wt.logger, wt.tracer, wt.registry, wt.converter, wt.propagators, &testHistoryProvider{tw.history}, tw.instance, tw.metadata, wt.clock)
+			e, err := executor.NewExecutor(wt.logger, wt.tracer, wt.registry, wt.converter, wt.propagators, &testHistoryProvider{tw.history}, tw.instance, tw.metadata, wt.clock)
 			if err != nil {
 				panic(fmt.Errorf("could not create workflow executor: %v", err))
 			}

worker/options.go

@@ -3,7 +3,7 @@ package worker
 import (
 	"time"
 
-	"github.com/cschleiden/go-workflows/internal/workflow"
+	"github.com/cschleiden/go-workflows/workflow/executor"
 )
 
 type Options struct {
@@ -47,7 +47,7 @@ type Options struct {
 
 	// WorkflowExecutorCache is the cache to use for workflow executors. If nil, a default cache implementation
 	// will be used.
-	WorkflowExecutorCache workflow.ExecutorCache
+	WorkflowExecutorCache executor.Cache
 }
 
 var DefaultOptions = Options{

worker/worker.go

@@ -11,9 +11,9 @@ import (
 	"github.com/cschleiden/go-workflows/client"
 	"github.com/cschleiden/go-workflows/internal/signals"
 	internal "github.com/cschleiden/go-workflows/internal/worker"
-	workflowinternal "github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/registry"
 	"github.com/cschleiden/go-workflows/workflow"
+	"github.com/cschleiden/go-workflows/workflow/executor"
 )
 
 type Worker struct {
@@ -24,7 +24,7 @@ type Worker struct {
 
 	registry *registry.Registry
 
-	workflowWorker *internal.Worker[backend.WorkflowTask, workflowinternal.ExecutionResult]
+	workflowWorker *internal.Worker[backend.WorkflowTask, executor.ExecutionResult]
 	activityWorker *internal.Worker[backend.ActivityTask, history.Event]
 }
 

internal/workflow/cache.go → workflow/executor/cache.go

@@ -1,12 +1,12 @@
-package workflow
+package executor
 
 import (
 	"context"
 
 	"github.com/cschleiden/go-workflows/core"
 )
 
-type ExecutorCache interface {
+type Cache 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)

internal/workflow/cache/cache.go → workflow/executor/cache/cache.go

@@ -7,22 +7,22 @@ import (
 	"github.com/cschleiden/go-workflows/backend/metrics"
 	"github.com/cschleiden/go-workflows/core"
 	"github.com/cschleiden/go-workflows/internal/metrickeys"
-	"github.com/cschleiden/go-workflows/internal/workflow"
+	"github.com/cschleiden/go-workflows/workflow/executor"
 	"github.com/jellydator/ttlcache/v3"
 )
 
 type lruCache struct {
 	mc metrics.Client
-	c  *ttlcache.Cache[string, workflow.WorkflowExecutor]
+	c  *ttlcache.Cache[string, executor.WorkflowExecutor]
 }
 
 func NewWorkflowExecutorLRUCache(mc metrics.Client, size int, expiration time.Duration) *lruCache {
 	c := ttlcache.New(
-		ttlcache.WithCapacity[string, workflow.WorkflowExecutor](uint64(size)),
-		ttlcache.WithTTL[string, workflow.WorkflowExecutor](expiration),
+		ttlcache.WithCapacity[string, executor.WorkflowExecutor](uint64(size)),
+		ttlcache.WithTTL[string, executor.WorkflowExecutor](expiration),
 	)
 
-	c.OnEviction(func(ctx context.Context, er ttlcache.EvictionReason, i *ttlcache.Item[string, workflow.WorkflowExecutor]) {
+	c.OnEviction(func(ctx context.Context, er ttlcache.EvictionReason, i *ttlcache.Item[string, executor.WorkflowExecutor]) {
 		// Close the executor to allow it to clean up resources.
 		i.Value().Close()
 
@@ -43,7 +43,7 @@ func NewWorkflowExecutorLRUCache(mc metrics.Client, size int, expiration time.Du
 	}
 }
 
-func (lc *lruCache) Get(ctx context.Context, instance *core.WorkflowInstance) (workflow.WorkflowExecutor, bool, error) {
+func (lc *lruCache) Get(ctx context.Context, instance *core.WorkflowInstance) (executor.WorkflowExecutor, bool, error) {
 	e := lc.c.Get(getKey(instance))
 	if e != nil {
 		return e.Value(), true, nil
@@ -52,7 +52,7 @@ func (lc *lruCache) Get(ctx context.Context, instance *core.WorkflowInstance) (w
 	return nil, false, nil
 }
 
-func (lc *lruCache) Store(ctx context.Context, instance *core.WorkflowInstance, executor workflow.WorkflowExecutor) error {
+func (lc *lruCache) Store(ctx context.Context, instance *core.WorkflowInstance, executor executor.WorkflowExecutor) error {
 	lc.c.Set(getKey(instance), executor, ttlcache.DefaultTTL)
 
 	lc.mc.Gauge(metrickeys.WorkflowInstanceCacheSize, metrics.Tags{}, int64(lc.c.Len()))

internal/workflow/cache/cache_test.go → workflow/executor/cache/cache_test.go

@@ -14,9 +14,9 @@ import (
 	"github.com/cschleiden/go-workflows/backend/metadata"
 	"github.com/cschleiden/go-workflows/core"
 	"github.com/cschleiden/go-workflows/internal/metrics"
-	wf "github.com/cschleiden/go-workflows/internal/workflow"
 	"github.com/cschleiden/go-workflows/registry"
 	"github.com/cschleiden/go-workflows/workflow"
+	"github.com/cschleiden/go-workflows/workflow/executor"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -28,14 +28,14 @@ func Test_Cache_StoreAndGet(t *testing.T) {
 	require.NoError(t, r.RegisterWorkflow(workflowWithActivity))
 
 	i := core.NewWorkflowInstance("instanceID", "executionID")
-	e, err := wf.NewExecutor(
+	e, err := executor.NewExecutor(
 		slog.Default(), trace.NewNoopTracerProvider().Tracer(backend.TracerName), r, converter.DefaultConverter,
 		[]workflow.ContextPropagator{}, &testHistoryProvider{}, i, &metadata.WorkflowMetadata{}, clock.New(),
 	)
 	require.NoError(t, err)
 
 	i2 := core.NewWorkflowInstance("instanceID2", "executionID2")
-	e2, err := wf.NewExecutor(
+	e2, err := executor.NewExecutor(
 		slog.Default(), trace.NewNoopTracerProvider().Tracer(backend.TracerName), r, converter.DefaultConverter,
 		[]workflow.ContextPropagator{}, &testHistoryProvider{}, i, &metadata.WorkflowMetadata{}, clock.New(),
 	)
@@ -68,7 +68,7 @@ func Test_Cache_AutoEviction(t *testing.T) {
 	i := core.NewWorkflowInstance("instanceID", "executionID")
 	r := registry.New()
 	require.NoError(t, r.RegisterWorkflow(workflowWithActivity))
-	e, err := wf.NewExecutor(
+	e, err := executor.NewExecutor(
 		slog.Default(), trace.NewNoopTracerProvider().Tracer(backend.TracerName), r,
 		converter.DefaultConverter, []workflow.ContextPropagator{}, &testHistoryProvider{}, i,
 		&metadata.WorkflowMetadata{}, clock.New(),
@@ -98,7 +98,7 @@ func Test_Cache_Evict(t *testing.T) {
 	i := core.NewWorkflowInstance("instanceID", "executionID")
 	r := registry.New()
 	require.NoError(t, r.RegisterWorkflow(workflowWithActivity))
-	e, err := wf.NewExecutor(
+	e, err := executor.NewExecutor(
 		slog.Default(), trace.NewNoopTracerProvider().Tracer(backend.TracerName), r,
 		converter.DefaultConverter, []workflow.ContextPropagator{}, &testHistoryProvider{}, i,
 		&metadata.WorkflowMetadata{}, clock.New(),

internal/workflow/executor.go → workflow/executor/executor.go

@@ -1,4 +1,4 @@
-package workflow
+package executor
 
 import (
 	"context"
@@ -30,10 +30,19 @@ import (
 )
 
 type ExecutionResult struct {
-	State          core.WorkflowInstanceState
-	Executed       []*history.Event
+	// New state of the workflow instance
+	State core.WorkflowInstanceState
+
+	// Events executed during the tastk execution
+	Executed []*history.Event
+
+	// Activities that were scheduled
 	ActivityEvents []*history.Event
-	TimerEvents    []*history.Event
+
+	// Timers that were scheduled
+	TimerEvents []*history.Event
+
+	// Events for other workflow instances
 	WorkflowEvents []history.WorkflowEvent
 }
 
@@ -382,7 +391,7 @@ func (e *executor) handleWorkflowExecutionStarted(a *history.ExecutionStartedAtt
 		return fmt.Errorf("workflow %s not found", a.Name)
 	}
 
-	e.workflow = NewWorkflow(reflect.ValueOf(wfFn))
+	e.workflow = newWorkflow(reflect.ValueOf(wfFn))
 
 	return e.workflow.Execute(e.workflowCtx, a.Inputs)
 }

internal/workflow/executor_test.go → workflow/executor/executor_test.go

@@ -1,4 +1,4 @@
-package workflow
+package executor
 
 import (
 	"context"

internal/workflow/workflow.go → workflow/executor/workflow.go

@@ -1,4 +1,4 @@
-package workflow
+package executor
 
 import (
 	"errors"
@@ -19,7 +19,7 @@ type workflow struct {
 	err    error
 }
 
-func NewWorkflow(workflowFn reflect.Value) *workflow {
+func newWorkflow(workflowFn reflect.Value) *workflow {
 	s := sync.NewScheduler()
 
 	return &workflow{

internal/workflow/workflow_test.go → workflow/executor/workflow_test.go

@@ -1,4 +1,4 @@
-package workflow
+package executor
 
 import (
 	"reflect"
@@ -25,7 +25,7 @@ func Test_Workflow_WrapsPanic(t *testing.T) {
 	ctx := sync.Background()
 	ctx = contextvalue.WithConverter(ctx, converter.DefaultConverter)
 
-	wf := NewWorkflow(reflect.ValueOf(w))
+	wf := newWorkflow(reflect.ValueOf(w))
 	err := wf.Execute(ctx, nil)
 	require.NoError(t, err)