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 <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user