From dabc458b965e1e3eeb32e7f177bd68943be3f62c Mon Sep 17 00:00:00 2001 From: Henry Date: Sun, 23 Aug 2026 21:00:06 +0000 Subject: [PATCH] perf(orchestrator): stop re-reading and recompiling workflow steps every trigger (#2946) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .../event_handlers_workflow_step_cache.go | 49 ++ apps/backend/internal/orchestrator/service.go | 3 + .../orchestrator/workflow_step_cache.go | 219 ++++++++ .../orchestrator/workflow_step_cache_test.go | 486 ++++++++++++++++++ .../internal/orchestrator/workflow_store.go | 67 ++- 5 files changed, 808 insertions(+), 16 deletions(-) create mode 100644 apps/backend/internal/orchestrator/event_handlers_workflow_step_cache.go create mode 100644 apps/backend/internal/orchestrator/workflow_step_cache.go create mode 100644 apps/backend/internal/orchestrator/workflow_step_cache_test.go diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow_step_cache.go b/apps/backend/internal/orchestrator/event_handlers_workflow_step_cache.go new file mode 100644 index 0000000000..d162a780fe --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_step_cache.go @@ -0,0 +1,49 @@ +package orchestrator + +import ( + "context" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +// subscribeWorkflowStepCacheEvents keeps workflowStore's compiled-step cache +// fresh by invalidating on the same workflow_step.* events the queue +// reconciler already subscribes to (see subscribeWorkflowQueueEvents). This +// is the fast invalidation path; stepSpecCacheTTL is the correctness floor +// for writer paths (git-backed sync, step reorder, step-delete reference +// clearing) that don't publish these events. +func (s *Service) subscribeWorkflowStepCacheEvents() { + if s.eventBus == nil { + return + } + for _, eventType := range []string{events.WorkflowStepCreated, events.WorkflowStepUpdated, events.WorkflowStepDeleted} { + if _, err := s.eventBus.Subscribe(eventType, s.handleWorkflowStepCacheInvalidation); err != nil { + s.logger.Error("failed to subscribe to workflow step event for cache invalidation", + zap.String("event_type", eventType), zap.Error(err)) + } + } +} + +func (s *Service) handleWorkflowStepCacheInvalidation(_ context.Context, event *bus.Event) error { + if s.workflowStore == nil || s.workflowStore.stepCache == nil || event == nil { + return nil + } + data, ok := event.Data.(map[string]interface{}) + if !ok { + return nil + } + step, ok := data["step"].(map[string]interface{}) + if !ok { + return nil + } + stepID, _ := step["id"].(string) + workflowID, _ := step["workflow_id"].(string) + if stepID == "" && workflowID == "" { + return nil + } + s.workflowStore.stepCache.invalidate(workflowID, stepID) + return nil +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 5a5b63b10d..a4e76657c2 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -2315,6 +2315,9 @@ func (s *Service) Start(ctx context.Context) error { // Reconcile queued tasks when WIP limits or feeder settings change. s.subscribeWorkflowQueueEvents() + // Invalidate the compiled-step cache when workflow steps change. + s.subscribeWorkflowStepCacheEvents() + // Restore durable dynamic policy waits after the route and lifecycle // services are ready. Only un-dispatched pending states are scheduled. s.startDynamicPolicyRecovery(ctx) diff --git a/apps/backend/internal/orchestrator/workflow_step_cache.go b/apps/backend/internal/orchestrator/workflow_step_cache.go new file mode 100644 index 0000000000..6ffd042f11 --- /dev/null +++ b/apps/backend/internal/orchestrator/workflow_step_cache.go @@ -0,0 +1,219 @@ +package orchestrator + +import ( + "strconv" + "strings" + "sync" + "time" + + "golang.org/x/sync/singleflight" + + "github.com/kandev/kandev/internal/workflow/engine" +) + +// stepSpecCacheTTL is the correctness floor for cached compiled steps. Step +// writes reach the cache either through workflow_step.created/updated/deleted +// events (the fast path, invalidating within the same process tick) or through +// this TTL (the floor): writer paths that don't publish those events today — +// git-backed workflow sync, step reorder, ClearStepReferences on step delete — +// leave a stale entry for at most one TTL window instead of indefinitely. Same +// defensive-interval reasoning as gitSnapshotCache's persist interval. +const stepSpecCacheTTL = 5 * time.Second + +// stepSpecCacheMaxEntries bounds the combined size of both cache maps so a +// long-lived backend with many workflows can't grow it without limit. +const stepSpecCacheMaxEntries = 2048 + +// posDirection distinguishes LoadNextStep from LoadPreviousStep entries that +// otherwise share a (workflowID, position) key. +type posDirection string + +const ( + posDirectionNext posDirection = "next" + posDirectionPrev posDirection = "prev" +) + +type stepSpecCacheEntry struct { + spec engine.StepSpec + expiresAt time.Time + storedAt time.Time +} + +// stepSpecCache caches engine.CompileStep output for workflowStore's +// LoadStep/LoadNextStep/LoadPreviousStep. StepSpec.Events is a reference type +// (map[Trigger][]Action), so every cached entry is shared across callers; all +// consumers only read it (range/len), never mutate it — see +// workflow_step_cache_test.go and the callers cited in the task plan. A +// cached StepSpec returned by get must never be mutated by callers. +// +// byStep is keyed by stepID (LoadStep). byPos is keyed by +// workflowID+"\x00"+direction+"\x00"+position (LoadNextStep/LoadPreviousStep), +// since those are looked up by position, not step ID. +type stepSpecCache struct { + mu sync.RWMutex + byStep map[string]stepSpecCacheEntry + byPos map[string]stepSpecCacheEntry + ttl time.Duration + maxSize int + now func() time.Time + + // sfStep/sfPos coalesce concurrent misses on the same key into one + // fetch, so N sessions hitting an expired/cold entry in the same instant + // don't all fall through to the DB — the exact contention this cache + // exists to remove. Separate groups mirror byStep/byPos: a stepID and a + // posKey live in disjoint key spaces already (posKey embeds NUL-joined + // segments a stepID can't contain), but two groups keep that invariant + // structural rather than relied-upon. + sfStep singleflight.Group + sfPos singleflight.Group +} + +func newStepSpecCache() *stepSpecCache { + return &stepSpecCache{ + byStep: make(map[string]stepSpecCacheEntry), + byPos: make(map[string]stepSpecCacheEntry), + ttl: stepSpecCacheTTL, + maxSize: stepSpecCacheMaxEntries, + now: time.Now, + } +} + +func posKey(workflowID string, direction posDirection, position int) string { + return workflowID + "\x00" + string(direction) + "\x00" + strconv.Itoa(position) +} + +func (c *stepSpecCache) getStep(stepID string) (engine.StepSpec, bool) { + return c.get(c.byStep, stepID) +} + +func (c *stepSpecCache) putStep(stepID string, spec engine.StepSpec) { + c.put(c.byStep, stepID, spec) +} + +func (c *stepSpecCache) getPos(workflowID string, direction posDirection, position int) (engine.StepSpec, bool) { + return c.get(c.byPos, posKey(workflowID, direction, position)) +} + +func (c *stepSpecCache) putPos(workflowID string, direction posDirection, position int, spec engine.StepSpec) { + c.put(c.byPos, posKey(workflowID, direction, position), spec) +} + +// getOrLoadStep returns stepID's cached spec when fresh; otherwise it runs +// fetch, coalescing concurrent misses on the same stepID via sfStep so only +// one caller reaches the DB. A fetch error is propagated to every waiter and +// never cached, matching getStep/putStep's existing miss contract. +func (c *stepSpecCache) getOrLoadStep(stepID string, fetch func() (engine.StepSpec, error)) (engine.StepSpec, error) { + if spec, ok := c.getStep(stepID); ok { + return spec, nil + } + v, err, _ := c.sfStep.Do(stepID, func() (interface{}, error) { + if spec, ok := c.getStep(stepID); ok { // another caller may have filled it + return spec, nil + } + spec, ferr := fetch() + if ferr != nil { + return engine.StepSpec{}, ferr + } + c.putStep(stepID, spec) + return spec, nil + }) + if err != nil { + return engine.StepSpec{}, err + } + return v.(engine.StepSpec), nil +} + +// getOrLoadPos is getOrLoadStep for the byPos cache, coalescing concurrent +// misses on the same (workflowID, direction, position) key via sfPos. +func (c *stepSpecCache) getOrLoadPos(workflowID string, direction posDirection, position int, fetch func() (engine.StepSpec, error)) (engine.StepSpec, error) { + if spec, ok := c.getPos(workflowID, direction, position); ok { + return spec, nil + } + key := posKey(workflowID, direction, position) + v, err, _ := c.sfPos.Do(key, func() (interface{}, error) { + if spec, ok := c.getPos(workflowID, direction, position); ok { // another caller may have filled it + return spec, nil + } + spec, ferr := fetch() + if ferr != nil { + return engine.StepSpec{}, ferr + } + c.putPos(workflowID, direction, position, spec) + return spec, nil + }) + if err != nil { + return engine.StepSpec{}, err + } + return v.(engine.StepSpec), nil +} + +// get reports a hit only when the key is present and unexpired. m must be +// c.byStep or c.byPos; the map header is passed by value but the underlying +// table is shared, so reads observe concurrent writes safely under the lock. +func (c *stepSpecCache) get(m map[string]stepSpecCacheEntry, key string) (engine.StepSpec, bool) { + c.mu.RLock() + entry, ok := m[key] + c.mu.RUnlock() + if !ok || c.now().After(entry.expiresAt) { + return engine.StepSpec{}, false + } + return entry.spec, true +} + +func (c *stepSpecCache) put(m map[string]stepSpecCacheEntry, key string, spec engine.StepSpec) { + c.mu.Lock() + defer c.mu.Unlock() + now := c.now() + if _, exists := m[key]; !exists && c.maxSize > 0 && len(c.byStep)+len(c.byPos) >= c.maxSize { + c.evictOldestLocked() + } + m[key] = stepSpecCacheEntry{spec: spec, expiresAt: now.Add(c.ttl), storedAt: now} +} + +// evictOldestLocked drops the single oldest entry (by storedAt) across both +// maps. Caller must hold c.mu. O(n) over the cache; only invoked when the +// cache is full, which is rare in practice — workflows hold a handful of +// steps and the combined cap is generous. +func (c *stepSpecCache) evictOldestLocked() { + var oldestMap map[string]stepSpecCacheEntry + var oldestKey string + var oldestAt time.Time + consider := func(m map[string]stepSpecCacheEntry) { + for k, entry := range m { + if oldestMap == nil || entry.storedAt.Before(oldestAt) { + oldestMap, oldestKey, oldestAt = m, k, entry.storedAt + } + } + } + consider(c.byStep) + consider(c.byPos) + if oldestMap != nil { + delete(oldestMap, oldestKey) + } +} + +// invalidate drops stepID's byStep entry and every cached entry belonging to +// workflowID. A step edit changes that step's own spec (byStep) and can shift +// Next/Prev results for its whole workflow (a position or move_to_step change), +// so the coarser workflow-scoped invalidation is required there — see the +// task plan's finding on ReorderSteps rewriting Position across every step. +func (c *stepSpecCache) invalidate(workflowID, stepID string) { + c.mu.Lock() + defer c.mu.Unlock() + if stepID != "" { + delete(c.byStep, stepID) + } + if workflowID != "" { + for k, entry := range c.byStep { + if entry.spec.WorkflowID == workflowID { + delete(c.byStep, k) + } + } + prefix := workflowID + "\x00" + for k := range c.byPos { + if strings.HasPrefix(k, prefix) { + delete(c.byPos, k) + } + } + } +} diff --git a/apps/backend/internal/orchestrator/workflow_step_cache_test.go b/apps/backend/internal/orchestrator/workflow_step_cache_test.go new file mode 100644 index 0000000000..2d9f601a8c --- /dev/null +++ b/apps/backend/internal/orchestrator/workflow_step_cache_test.go @@ -0,0 +1,486 @@ +package orchestrator + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/workflow/engine" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// blockingStepGetter blocks the first GetStep call on release, closing +// entered exactly once so a concurrent-miss test can deterministically wait +// for the fetch to actually start before releasing it. If singleflight +// coalescing ever regresses and GetStep is invoked a second time while the +// first is still blocked, the second close(entered) panics, failing loudly. +type blockingStepGetter struct { + *countingStepGetter + entered chan struct{} + release chan struct{} +} + +func newBlockingStepGetter() *blockingStepGetter { + return &blockingStepGetter{ + countingStepGetter: newCountingStepGetter(), + entered: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (b *blockingStepGetter) GetStep(ctx context.Context, stepID string) (*wfmodels.WorkflowStep, error) { + close(b.entered) + <-b.release + return b.countingStepGetter.GetStep(ctx, stepID) +} + +// erroringStepGetter always fails GetStep, mirroring the real repository's +// error return for a missing step (unlike mockStepGetter, which returns +// nil, nil and would make LoadStep call engine.CompileStep(nil)). +type erroringStepGetter struct { + *mockStepGetter + err error +} + +func (e *erroringStepGetter) GetStep(_ context.Context, _ string) (*wfmodels.WorkflowStep, error) { + return nil, e.err +} + +// countingStepGetter wraps mockStepGetter and counts calls per method so +// tests can assert the cache actually avoided a repeated getter call. +type countingStepGetter struct { + *mockStepGetter + mu sync.Mutex + getStepCalls int + getNextCalls int + getPreviousCalls int +} + +func newCountingStepGetter() *countingStepGetter { + return &countingStepGetter{mockStepGetter: newMockStepGetter()} +} + +func (c *countingStepGetter) GetStep(ctx context.Context, stepID string) (*wfmodels.WorkflowStep, error) { + c.mu.Lock() + c.getStepCalls++ + c.mu.Unlock() + return c.mockStepGetter.GetStep(ctx, stepID) +} + +func (c *countingStepGetter) GetNextStepByPosition(ctx context.Context, workflowID string, currentPosition int) (*wfmodels.WorkflowStep, error) { + c.mu.Lock() + c.getNextCalls++ + c.mu.Unlock() + return c.mockStepGetter.GetNextStepByPosition(ctx, workflowID, currentPosition) +} + +func (c *countingStepGetter) GetPreviousStepByPosition(ctx context.Context, workflowID string, currentPosition int) (*wfmodels.WorkflowStep, error) { + c.mu.Lock() + c.getPreviousCalls++ + c.mu.Unlock() + return c.mockStepGetter.GetPreviousStepByPosition(ctx, workflowID, currentPosition) +} + +func stepEvent(eventType, stepID, workflowID string) *bus.Event { + return bus.NewEvent(eventType, "test", map[string]interface{}{ + "step": map[string]interface{}{ + "id": stepID, + "workflow_id": workflowID, + }, + }) +} + +// 1. A cache hit must not re-invoke the getter. +func TestWorkflowStore_LoadStep_CachesAcrossCalls(t *testing.T) { + ctx := context.Background() + stepGetter := newCountingStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Name: "Planning", Position: 0} + + store := newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + var last engine.StepSpec + for i := 0; i < 4; i++ { + spec, err := store.LoadStep(ctx, "wf1", "step1") + if err != nil { + t.Fatalf("LoadStep call %d failed: %v", i, err) + } + last = spec + } + + if last.ID != "step1" { + t.Fatalf("expected spec for step1, got %q", last.ID) + } + stepGetter.mu.Lock() + calls := stepGetter.getStepCalls + stepGetter.mu.Unlock() + if calls != 1 { + t.Fatalf("expected exactly 1 getter call across 4 LoadStep calls, got %d", calls) + } +} + +// 2. Expiry after TTL forces a refetch, via an injected clock (no sleeps). +func TestWorkflowStore_LoadStep_ExpiresAfterTTL(t *testing.T) { + ctx := context.Background() + stepGetter := newCountingStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Name: "Planning"} + + store := newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + now := time.Now() + store.stepCache.now = func() time.Time { return now } + + if _, err := store.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("first LoadStep failed: %v", err) + } + // Still within TTL: no refetch. + now = now.Add(stepSpecCacheTTL - time.Millisecond) + if _, err := store.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("second LoadStep failed: %v", err) + } + // Past TTL: must refetch. + now = now.Add(2 * time.Millisecond) + if _, err := store.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("third LoadStep failed: %v", err) + } + + stepGetter.mu.Lock() + calls := stepGetter.getStepCalls + stepGetter.mu.Unlock() + if calls != 2 { + t.Fatalf("expected 2 getter calls (initial + post-expiry refetch), got %d", calls) + } +} + +// 3. workflow_step.updated for a step drops that step's byStep entry. +func TestStepSpecCache_UpdatedEventDropsStepEntry(t *testing.T) { + ctx := context.Background() + stepGetter := newCountingStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Name: "Planning"} + + svc := &Service{logger: testLogger()} + svc.workflowStore = newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + if _, err := svc.workflowStore.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("LoadStep failed: %v", err) + } + + if err := svc.handleWorkflowStepCacheInvalidation(ctx, stepEvent(events.WorkflowStepUpdated, "step1", "wf1")); err != nil { + t.Fatalf("handleWorkflowStepCacheInvalidation failed: %v", err) + } + + if _, err := svc.workflowStore.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("LoadStep after invalidation failed: %v", err) + } + + stepGetter.mu.Lock() + calls := stepGetter.getStepCalls + stepGetter.mu.Unlock() + if calls != 2 { + t.Fatalf("expected a refetch after invalidation (2 calls), got %d", calls) + } +} + +// 4. workflow_step.updated drops the Next/Prev entries of the *same* +// workflow and leaves other workflows' entries untouched. +func TestStepSpecCache_UpdatedEventDropsSameWorkflowPositionEntriesOnly(t *testing.T) { + ctx := context.Background() + stepGetter := newCountingStepGetter() + stepGetter.steps["wf1-step1"] = &wfmodels.WorkflowStep{ID: "wf1-step1", WorkflowID: "wf1", Position: 0} + stepGetter.steps["wf1-step2"] = &wfmodels.WorkflowStep{ID: "wf1-step2", WorkflowID: "wf1", Position: 1} + stepGetter.steps["wf2-step1"] = &wfmodels.WorkflowStep{ID: "wf2-step1", WorkflowID: "wf2", Position: 0} + stepGetter.steps["wf2-step2"] = &wfmodels.WorkflowStep{ID: "wf2-step2", WorkflowID: "wf2", Position: 1} + + svc := &Service{logger: testLogger()} + svc.workflowStore = newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + if _, err := svc.workflowStore.LoadNextStep(ctx, "wf1", 0); err != nil { + t.Fatalf("LoadNextStep(wf1) failed: %v", err) + } + if _, err := svc.workflowStore.LoadNextStep(ctx, "wf2", 0); err != nil { + t.Fatalf("LoadNextStep(wf2) failed: %v", err) + } + + if err := svc.handleWorkflowStepCacheInvalidation(ctx, stepEvent(events.WorkflowStepUpdated, "wf1-step2", "wf1")); err != nil { + t.Fatalf("handleWorkflowStepCacheInvalidation failed: %v", err) + } + + if _, err := svc.workflowStore.LoadNextStep(ctx, "wf1", 0); err != nil { + t.Fatalf("LoadNextStep(wf1) after invalidation failed: %v", err) + } + if _, err := svc.workflowStore.LoadNextStep(ctx, "wf2", 0); err != nil { + t.Fatalf("LoadNextStep(wf2) after invalidation failed: %v", err) + } + + stepGetter.mu.Lock() + nextCalls := stepGetter.getNextCalls + stepGetter.mu.Unlock() + // wf1: 2 (initial + refetch after invalidation). wf2: 1 (still cached). + if nextCalls != 3 { + t.Fatalf("expected 3 GetNextStepByPosition calls (wf1 refetched, wf2 still cached), got %d", nextCalls) + } +} + +// 5. A deleted step can have references cleared from other steps before its +// event is published. The event must therefore drop every direct entry for +// the workflow, not only the deleted step's entry. +func TestStepSpecCache_DeletedEventDropsAllWorkflowStepEntries(t *testing.T) { + cache := newStepSpecCache() + cache.putStep("source", engine.StepSpec{ID: "source", WorkflowID: "wf1"}) + cache.putStep("deleted", engine.StepSpec{ID: "deleted", WorkflowID: "wf1"}) + cache.putStep("other-workflow", engine.StepSpec{ID: "other-workflow", WorkflowID: "wf2"}) + + cache.invalidate("wf1", "deleted") + + if _, ok := cache.getStep("source"); ok { + t.Fatal("expected workflow-wide invalidation to drop the cached source step") + } + if _, ok := cache.getStep("deleted"); ok { + t.Fatal("expected deleted step entry to be removed") + } + if _, ok := cache.getStep("other-workflow"); !ok { + t.Fatal("expected entries for other workflows to remain cached") + } +} + +// 6. Both .deleted and .created invalidate (created covers the demoted- +// start-step fan-out, which publishes .updated for demoted steps but the +// cache must also tolerate a .created arriving for an unrelated step ID). +func TestStepSpecCache_CreatedAndDeletedEventsInvalidate(t *testing.T) { + for _, eventType := range []string{events.WorkflowStepCreated, events.WorkflowStepDeleted} { + t.Run(eventType, func(t *testing.T) { + ctx := context.Background() + stepGetter := newCountingStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1"} + + svc := &Service{logger: testLogger()} + svc.workflowStore = newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + if _, err := svc.workflowStore.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("LoadStep failed: %v", err) + } + if err := svc.handleWorkflowStepCacheInvalidation(ctx, stepEvent(eventType, "step1", "wf1")); err != nil { + t.Fatalf("handleWorkflowStepCacheInvalidation failed: %v", err) + } + if _, err := svc.workflowStore.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("LoadStep after invalidation failed: %v", err) + } + + stepGetter.mu.Lock() + calls := stepGetter.getStepCalls + stepGetter.mu.Unlock() + if calls != 2 { + t.Fatalf("expected a refetch after %s (2 calls), got %d", eventType, calls) + } + }) + } +} + +// 7. A getter error is returned and not cached. +func TestWorkflowStore_LoadStep_ErrorNotCached(t *testing.T) { + ctx := context.Background() + stepGetter := &erroringStepGetter{mockStepGetter: newMockStepGetter(), err: errors.New("boom")} + + store := newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + if _, err := store.LoadStep(ctx, "wf1", "step1"); err == nil { + t.Fatal("expected an error for a missing step") + } + if _, ok := store.stepCache.getStep("step1"); ok { + t.Fatal("an error result must not be cached") + } + if _, err := store.LoadStep(ctx, "wf1", "step1"); err == nil { + t.Fatal("expected the error again on the second call") + } +} + +// 8. LoadNextStep's step == nil case still returns today's "no next step" +// error, uncached. +func TestWorkflowStore_LoadNextStep_NoNextStepNotCached(t *testing.T) { + ctx := context.Background() + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Position: 0} + + store := newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + _, err := store.LoadNextStep(ctx, "wf1", 0) + if err == nil { + t.Fatal("expected an error when there is no next step") + } + if _, ok := store.stepCache.getPos("wf1", posDirectionNext, 0); ok { + t.Fatal("a nil-step result must not be cached") + } +} + +// 9. Eviction at maxEntries: filling the cache past its bound evicts the +// oldest entry rather than growing without limit. +func TestStepSpecCache_EvictsOldestAtMaxEntries(t *testing.T) { + cache := newStepSpecCache() + cache.maxSize = 2 + now := time.Now() + cache.now = func() time.Time { return now } + + cache.putStep("step1", engine.StepSpec{ID: "step1"}) + now = now.Add(time.Millisecond) + cache.putStep("step2", engine.StepSpec{ID: "step2"}) + now = now.Add(time.Millisecond) + cache.putStep("step3", engine.StepSpec{ID: "step3"}) + + if _, ok := cache.getStep("step1"); ok { + t.Fatal("expected the oldest entry (step1) to have been evicted") + } + if _, ok := cache.getStep("step2"); !ok { + t.Fatal("expected step2 to still be cached") + } + if _, ok := cache.getStep("step3"); !ok { + t.Fatal("expected step3 to still be cached") + } +} + +// 10. Concurrent get/put/invalidate must not race (run with -race). +func TestStepSpecCache_ConcurrentAccess(t *testing.T) { + cache := newStepSpecCache() + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(3) + go func(i int) { + defer wg.Done() + cache.putStep("step1", engine.StepSpec{ID: "step1"}) + }(i) + go func(i int) { + defer wg.Done() + cache.getStep("step1") + }(i) + go func(i int) { + defer wg.Done() + cache.invalidate("wf1", "step1") + }(i) + } + wg.Wait() +} + +// 11. LoadPreviousStep caches across calls and is invalidated by a workflow +// step event, exercising the byPos "prev" direction — the rest of this suite +// only exercises "next" via LoadNextStep. +func TestWorkflowStore_LoadPreviousStep_CachesAndInvalidates(t *testing.T) { + ctx := context.Background() + stepGetter := newCountingStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Position: 0} + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ID: "step2", WorkflowID: "wf1", Position: 1} + + svc := &Service{logger: testLogger()} + svc.workflowStore = newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + spec, err := svc.workflowStore.LoadPreviousStep(ctx, "wf1", 1) + if err != nil { + t.Fatalf("first LoadPreviousStep failed: %v", err) + } + if spec.ID != "step1" { + t.Fatalf("expected previous step %q, got %q", "step1", spec.ID) + } + if _, err := svc.workflowStore.LoadPreviousStep(ctx, "wf1", 1); err != nil { + t.Fatalf("second LoadPreviousStep failed: %v", err) + } + + stepGetter.mu.Lock() + calls := stepGetter.getPreviousCalls + stepGetter.mu.Unlock() + if calls != 1 { + t.Fatalf("expected exactly 1 getter call across 2 LoadPreviousStep calls, got %d", calls) + } + + if err := svc.handleWorkflowStepCacheInvalidation(ctx, stepEvent(events.WorkflowStepUpdated, "step1", "wf1")); err != nil { + t.Fatalf("handleWorkflowStepCacheInvalidation failed: %v", err) + } + + if _, err := svc.workflowStore.LoadPreviousStep(ctx, "wf1", 1); err != nil { + t.Fatalf("LoadPreviousStep after invalidation failed: %v", err) + } + + stepGetter.mu.Lock() + calls = stepGetter.getPreviousCalls + stepGetter.mu.Unlock() + if calls != 2 { + t.Fatalf("expected a refetch after invalidation (2 calls), got %d", calls) + } +} + +// 12. subscribeWorkflowStepCacheEvents actually registers the handler on the +// bus and reacts to a delivered event, rather than trusting that +// handleWorkflowStepCacheInvalidation (called directly by tests 3-5) is ever +// wired up. A wrong event-type constant or a missing Subscribe call would +// leave every LoadStep call here served from a stale cache entry. +func TestSubscribeWorkflowStepCacheEvents_DeliversThroughBus(t *testing.T) { + ctx := context.Background() + stepGetter := newCountingStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1"} + + eventBus := bus.NewMemoryEventBus(testLogger()) + svc := &Service{logger: testLogger(), eventBus: eventBus} + svc.workflowStore = newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + svc.subscribeWorkflowStepCacheEvents() + + for _, eventType := range []string{events.WorkflowStepCreated, events.WorkflowStepUpdated, events.WorkflowStepDeleted} { + if _, err := svc.workflowStore.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("LoadStep before %s failed: %v", eventType, err) + } + if err := eventBus.Publish(ctx, eventType, stepEvent(eventType, "step1", "wf1")); err != nil { + t.Fatalf("publish %s failed: %v", eventType, err) + } + } + if _, err := svc.workflowStore.LoadStep(ctx, "wf1", "step1"); err != nil { + t.Fatalf("final LoadStep failed: %v", err) + } + + stepGetter.mu.Lock() + calls := stepGetter.getStepCalls + stepGetter.mu.Unlock() + // One fetch per LoadStep call (4 total): each of the 3 bus-delivered + // events must invalidate the entry the prior LoadStep just populated, or + // this would stay at 1. + if calls != 4 { + t.Fatalf("expected 4 getter calls (each bus-delivered event invalidating the prior load), got %d", calls) + } +} + +// 13. Concurrent misses on the same key coalesce into one getter call via +// stepCache's singleflight group, rather than every waiter independently +// re-reading and recompiling the step. +func TestWorkflowStore_LoadStep_CoalescesConcurrentMisses(t *testing.T) { + ctx := context.Background() + stepGetter := newBlockingStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Name: "Planning"} + + store := newWorkflowStore(nil, stepGetter, nil, noopPublisher, testLogger()) + + const n = 10 + var wg sync.WaitGroup + errCh := make(chan error, n) + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + _, err := store.LoadStep(ctx, "wf1", "step1") + errCh <- err + }() + } + + <-stepGetter.entered // the leader has reached the DB read + close(stepGetter.release) // let it (and only it) complete the fetch + wg.Wait() + close(errCh) + for err := range errCh { + if err != nil { + t.Fatalf("LoadStep returned error: %v", err) + } + } + + stepGetter.mu.Lock() + calls := stepGetter.getStepCalls + stepGetter.mu.Unlock() + if calls != 1 { + t.Fatalf("expected exactly 1 getter call across %d concurrent LoadStep calls, got %d", n, calls) + } +} diff --git a/apps/backend/internal/orchestrator/workflow_store.go b/apps/backend/internal/orchestrator/workflow_store.go index 66fa645f78..c80387a719 100644 --- a/apps/backend/internal/orchestrator/workflow_store.go +++ b/apps/backend/internal/orchestrator/workflow_store.go @@ -101,6 +101,7 @@ type workflowStore struct { stepHistoryRecorder StepHistoryRecorder guardedLifecycle guardedTransitionLifecycle appliedOps sync.Map + stepCache *stepSpecCache } func newWorkflowStore( @@ -145,6 +146,7 @@ func newWorkflowStore( publishStateChanged: stateChanged, logger: log, stepHistoryRecorder: history, + stepCache: newStepSpecCache(), } } @@ -179,34 +181,67 @@ func (s *workflowStore) LoadState(ctx context.Context, taskID, sessionID string) return assembleMachineState(task, session, isPassthrough), nil } +// LoadStep returns stepID's compiled spec, serving it from the process-local +// stepCache when fresh. On a miss, concurrent callers for the same stepID +// coalesce onto one DB read + compile via stepCache's singleflight group. The +// returned StepSpec is shared across callers and must be treated as +// immutable — see stepSpecCache's doc comment. func (s *workflowStore) LoadStep(ctx context.Context, _, stepID string) (engine.StepSpec, error) { - step, err := s.workflowStepGetter.GetStep(ctx, stepID) - if err != nil { - return engine.StepSpec{}, fmt.Errorf("load step %s: %w", stepID, err) + fetch := func() (engine.StepSpec, error) { + step, err := s.workflowStepGetter.GetStep(ctx, stepID) + if err != nil { + return engine.StepSpec{}, fmt.Errorf("load step %s: %w", stepID, err) + } + if step == nil { + return engine.StepSpec{}, fmt.Errorf("step %s not found", stepID) + } + return engine.CompileStep(step), nil } - return engine.CompileStep(step), nil + if s.stepCache == nil { + return fetch() + } + return s.stepCache.getOrLoadStep(stepID, fetch) } +// LoadNextStep returns the compiled spec of the step after currentPosition in +// workflowID, serving it from stepCache when fresh and coalescing concurrent +// misses on the same position. See LoadStep on cache sharing/immutability. func (s *workflowStore) LoadNextStep(ctx context.Context, workflowID string, currentPosition int) (engine.StepSpec, error) { - step, err := s.workflowStepGetter.GetNextStepByPosition(ctx, workflowID, currentPosition) - if err != nil { - return engine.StepSpec{}, fmt.Errorf("load next step after position %d: %w", currentPosition, err) + fetch := func() (engine.StepSpec, error) { + step, err := s.workflowStepGetter.GetNextStepByPosition(ctx, workflowID, currentPosition) + if err != nil { + return engine.StepSpec{}, fmt.Errorf("load next step after position %d: %w", currentPosition, err) + } + if step == nil { + return engine.StepSpec{}, fmt.Errorf("no next step after position %d in workflow %s", currentPosition, workflowID) + } + return engine.CompileStep(step), nil } - if step == nil { - return engine.StepSpec{}, fmt.Errorf("no next step after position %d in workflow %s", currentPosition, workflowID) + if s.stepCache == nil { + return fetch() } - return engine.CompileStep(step), nil + return s.stepCache.getOrLoadPos(workflowID, posDirectionNext, currentPosition, fetch) } +// LoadPreviousStep returns the compiled spec of the step before +// currentPosition in workflowID, serving it from stepCache when fresh and +// coalescing concurrent misses on the same position. See LoadStep on cache +// sharing/immutability. func (s *workflowStore) LoadPreviousStep(ctx context.Context, workflowID string, currentPosition int) (engine.StepSpec, error) { - step, err := s.workflowStepGetter.GetPreviousStepByPosition(ctx, workflowID, currentPosition) - if err != nil { - return engine.StepSpec{}, fmt.Errorf("load previous step before position %d: %w", currentPosition, err) + fetch := func() (engine.StepSpec, error) { + step, err := s.workflowStepGetter.GetPreviousStepByPosition(ctx, workflowID, currentPosition) + if err != nil { + return engine.StepSpec{}, fmt.Errorf("load previous step before position %d: %w", currentPosition, err) + } + if step == nil { + return engine.StepSpec{}, fmt.Errorf("no previous step before position %d in workflow %s", currentPosition, workflowID) + } + return engine.CompileStep(step), nil } - if step == nil { - return engine.StepSpec{}, fmt.Errorf("no previous step before position %d in workflow %s", currentPosition, workflowID) + if s.stepCache == nil { + return fetch() } - return engine.CompileStep(step), nil + return s.stepCache.getOrLoadPos(workflowID, posDirectionPrev, currentPosition, fetch) } func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID, fromStepID, toStepID string, trigger engine.Trigger) error {