refactor(sdk/event): subject-routed Bus, Envelope, MemoryBus with backpressure & route cache#31
Merged
Merged
Conversation
Pure rename. Behaviour and exported field shapes are unchanged.
Renamed (in sdk/event/bus.go):
MemoryBus -> LegacyMemoryBus
EventBus -> LegacyEventBus
Subscription -> LegacySubscription
NoopBus -> LegacyNoopBus
MemoryBusOption -> LegacyMemoryBusOption
SubscribeOption -> LegacySubscribeOption
NewMemoryBus -> NewLegacyMemoryBus
WithBufferSize -> LegacyWithBufferSize
WithDropCallback -> LegacyWithDropCallback
Kept as-is (will be removed wholesale in v0.2.0):
Event, EventType, EventFilter, all EventXxx constants
Every renamed symbol carries a // Deprecated: note. The new subject-routed
Bus / Envelope / MemoryBus types will land in subsequent commits and take
over these short names.
In-tree call sites updated:
sdk/graph/executor/{executor,runner,parallel}.go (+ tests)
sdk/kanban/{board,events,kanban,helpers,board_test,kanban_test,helpers_test}.go
Verified: cd sdk && go test ./... -count=1 # all green
cd sdkx && go build ./... # ok
cd plugin && go build ./... # ok
cd examples/voice-pipeline && go build ./... # ok
Made-with: Cursor
…Observer and backpressure
New types (live alongside the deprecated Legacy* family from the previous
commit; no behaviour change to existing call sites):
Envelope cross-process-friendly carrier; Payload is json.RawMessage
so in-memory and remote bus implementations behave
identically (bytes in / bytes out)
NewEnvelope/MustEnvelope/Decode helpers; well-known header keys
(HeaderRunID/NodeID/ActorID/GraphID/Tenant) and typed accessors
Subject dot-delimited routing key, e.g. graph.run.r1.start
Pattern NATS-style matcher; '*' matches one segment, '>'
matches one or more trailing segments
Subject.Validate / Pattern.Validate / Pattern.Matches
Bus / Subscription new primary interface
SubOption WithBufferSize, WithBackpressure, WithPredicate
BackpressurePolicy DropNewest (default), DropOldest, Block
Observer OnPublish / OnDeliver / OnDrop, invoked
outside any bus lock; replaces the legacy
WithDropCallback hook
DropReason BufferFull / Closed
SubscriptionID opaque per-bus identifier
MemoryBus in-process implementation
NewMemoryBus(opts...) WithObserver option
NoopBus / NewNoopBus
Concurrency fixes baked in (vs LegacyMemoryBus):
* Observer callbacks are deferred into a local slice and fired after the
bus RLock is released — matches the documented "no callbacks under
bus lock" contract.
* MemoryBus.Close runs a multi-phase shutdown:
1. write-lock, mark closed, snapshot subscribers
2. signalClose() each sub (closes sub.done so Block-backpressure
publishers exit instead of staying parked)
3. inflight.Wait() (drains every in-flight Publish)
4. for each sub: senders.Wait() then closeChanFromBus()
This fixes the legacy "send on closed channel" race (P8) without
introducing the inverse "Close hangs forever on a parked Block
publisher" deadlock.
* memSub.Close (user-driven) runs the same two-phase shutdown scoped
to one subscription: removeSub → close(done) → senders.Wait() →
close(ch). signalOnce / chanOnce keep the bus-driven and user-driven
paths from racing each other.
* sub.senders sync.WaitGroup tracks Block-path publishers that have
released the bus RLock. Non-blocking DropNewest / DropOldest sends
complete while still holding RLock and therefore do not need to
register on senders.
Tests (all under -race -count=10):
subject_test.go Subject/Pattern validation + matching
including all wildcard edge cases
envelope_test.go struct/raw/nil payload, headers helpers,
JSON roundtrip, MustEnvelope panic path
memory_test.go pattern routing, predicate, every
backpressure policy, observer counts,
ctx cancel, Close idempotency, plus the
two regression tests:
TestMemoryBus_Close_NoSendOnClosedChannelPanic
(race-detector verifies P8 fix)
TestMemoryBus_Close_UnblocksParkedPublisher
(verifies the deadlock fix above)
Both regression tests were proven to actually catch the bugs they target
by temporarily disabling the corresponding fix and observing the failure.
Made-with: Cursor
Switches every Publish call site in the graph executor from the
deprecated LegacyEventBus / Event API to the new Bus / Envelope API
landed in the previous commit.
Subject convention used by this package (sdk-internal; no relation to
any flowcraft business schema):
graph.run.<runID>.start
graph.run.<runID>.end
graph.run.<runID>.parallel.fork
graph.run.<runID>.parallel.join
graph.run.<runID>.node.<nodeID>.start
graph.run.<runID>.node.<nodeID>.complete
graph.run.<runID>.node.<nodeID>.error
graph.run.<runID>.node.<nodeID>.skipped
graph.run.<runID>.node.<nodeID>.stream.delta
Identifier safety: runID and nodeID are passed through sanitiseID, which
substitutes '_' for any '.', '*', '>' bytes so user-supplied IDs cannot
fragment the Subject or accidentally turn into a wildcard. Empty IDs
become "_" to keep the segment count stable.
Header propagation on every emitted Envelope (replaces the flat fields
on the legacy Event struct):
HeaderRunID cfg.runID
HeaderGraphID g.Name()
HeaderActorID actorKeyFrom(ctx)
HeaderNodeID se.NodeID / nodeID (node-scoped events only)
Helpers exposed for downstream subscribers:
PatternRun(runID) graph.run.<runID>.>
PatternAllRuns() graph.run.>
PatternRunNodes(runID) graph.run.<runID>.node.>
Behaviour changes:
* EventBus type on Runner / WithEventBus / WithRunnerEventBus is now
event.Bus (was event.LegacyEventBus). Source-level breaking change
for any caller that named the type explicitly; constructor usage
(NewLegacyMemoryBus → NewMemoryBus) needs adapting.
* Default bus is event.NoopBus{} (was event.LegacyNoopBus{}); behaves
identically (drops everything).
* Node error envelope payload is now {"error": "<msg>"} instead of the
bare string err.Error(). Required because Envelope.Payload is
json.RawMessage and json-encoding a bare string produces a quoted
JSON string literal — the structured map is what most subscribers
actually want and is easier to extend later (e.g. adding error code
fields). Documented for the MIGRATION note.
Tests:
subjects_test.go new — verifies the convention strings,
Pattern.Validate / Matches behaviour,
and the sanitiseID escaping (including
the dot-in-runID edge case).
executor_test.go TestLocalExecutor_EventBus_Integration
rewritten to subscribe via PatternRun
and assert subject + headers.
runner_test.go TestRunner_WithEventBus rewritten the
same way; non-blocking default-case
drain replaced with a timeout loop
because graph.end is published from a
different goroutine in the new code path.
All sdk packages pass `go test -race -count=1`.
Made-with: Cursor
Switches every Publish call site in the kanban package from the
deprecated LegacyEventBus / Event API to the new Bus / Envelope API
landed in commit 2 of this PR.
Subject convention used by this package (sdk-internal; no relation to
any flowcraft business schema):
kanban.card.<cardID>.task.submitted
kanban.card.<cardID>.task.claimed
kanban.card.<cardID>.task.completed
kanban.card.<cardID>.task.failed
kanban.card.<cardID>.callback.start
kanban.card.<cardID>.callback.done
kanban.cron.<scheduleID>.rule.created
kanban.cron.<scheduleID>.rule.fired
kanban.cron.<scheduleID>.rule.disabled
Choice of partition key:
* card_id for task / callback events: every payload already carries
CardID as its first field, and consumers overwhelmingly want
"tell me everything that happened to this card";
* schedule_id for cron events: cron rules outlive any single card
they spawn, and the natural subscription scope is the rule itself.
Identifier safety: cardID and scheduleID are passed through sanitiseID,
which substitutes '_' for any '.', '*', '>' bytes so user-supplied IDs
cannot fragment the Subject or accidentally turn into a wildcard. Empty
IDs become "_" to keep the segment count stable.
Header propagation on every emitted Envelope (replaces the flat fields
on the legacy Event struct):
HeaderKanbanKind EventKind constant ("kanban.task.submitted", ...)
HeaderCardID cardID (task / callback events only)
HeaderScheduleID scheduleID (cron events only)
Source board.ScopeID()
Subscribers using a broad pattern (kanban.>) can now route on the
kanban_kind header instead of re-parsing the subject.
Helpers exposed for downstream subscribers:
PatternCard(cardID) kanban.card.<cardID>.>
PatternAllCards() kanban.card.>
PatternCronRule(schedID) kanban.cron.<schedID>.>
PatternAllCron() kanban.cron.>
PatternAll() kanban.>
Behaviour changes:
* Board.Bus() / Kanban.Bus() now return event.Bus (was LegacyEventBus).
Source-level breaking change for any caller that named the type
explicitly; constructor usage (NewLegacyMemoryBus -> NewMemoryBus)
needs adapting.
* WithEventBus(event.Bus) Option is still a no-op; signature updated to
keep source compatibility with the new Bus type. Will be removed in
v0.2.0 together with the deprecated bridge.
* The internal eventEnvelope() helper is gone; Publish call sites now
go through publishCardEvent / publishCronEvent, which build an
Envelope, set the well-known headers, and dispatch through Bus.
stampVersion() retains the previous "auto-fill payload Version=N"
behaviour and is exercised directly by TestStampVersion_*.
* CallbackStartPayload / CallbackDonePayload are now also stamped by
stampVersion (oversight in the previous code that the new test
coverage caught).
Tests:
subjects_test.go new asserts every subject string, validates
each Pattern, exercises sanitiseID
(including the dot-in-cardID edge case).
events_test.go rewritten — kindOf(env) replaces the old
string(ev.Type) check; payload assertions go
through env.Decode(&p); a new
TestBus_SubjectAndHeadersForTaskSubmitted pins
the subject + headers contract; a new
TestBus_PatternCardScopesToSingleCard verifies
two concurrent submits land on distinct subjects.
helpers_test.go drainEvents/subscribeBus updated to event.Subscription
and PatternAll(); subscribeBus takes a generous
buffer (1024) so backpressure never trims a test.
board_test.go Bus_Publishes/FanOuts updated to construct
Envelopes via event.NewEnvelope.
kanban_test.go TestKanban_WithEventBus_DeprecatedNoOp updated
to event.NewMemoryBus + Pattern(">").
scheduler_test.go Cron rule tests updated to env.Decode + kindOf.
board_concurrency_test.go Subscribe call updated to Pattern(">").
All sdk packages pass `go test -race -count=1`.
Made-with: Cursor
Raises sdk/event package coverage from 84.8% to 93.3%. Pure test
addition — no production code touched.
What was missing before:
observer.go noopObserver / String() helpers were entirely
uncovered (0% on every method) because no test ever
exercised the no-observer fallback shape directly.
memory.go NoopBus + noopSubscription had 0% across Publish,
Subscribe, Close, ID, C — the documented "drop
everything" Bus implementation was unverified.
MemoryBus.Publish was at 80.6%; the missing branches
were the auto-fill of envelope ID/Time, the post-Close
fast-fail path, and the DropOldest retry-after-evict
path.
*memSub.ID was 0% because no existing test reached
for sub.ID() directly.
envelope.go NewEnvelope was at 68.8% — the []byte payload branch,
the typed-nil ([]byte and json.RawMessage) branches,
the json.Marshal failure branch, and Decode's error-
wrapping branch had no coverage.
subject.go Pattern.Matches was at 83.3% — the documented "if a
malformed '>' slips through Validate, treat it as a
literal" safety net branch was not exercised.
New test files:
observer_test.go BackpressurePolicy / DropReason / noopObserver
noop_test.go NoopBus end-to-end + Bus interface assertion
coverage_extra_test.go MemoryBus.Publish edge paths,
Subscription.ID stability + uniqueness,
Pattern.Matches malformed-trail fallback
Plus six new cases appended to envelope_test.go for the byte-slice,
nil-typed, marshal-failure, decode-failure, and zero-Headers paths.
Remaining gaps (all unreachable from tests by design):
fireObserver 87.5% — the `if b.observer == nil` early return is
dead code: NewMemoryBus seeds observer with
noopObserver{} so the field is never nil. Left in
place as a defensive guard.
Publish 87.1% — the second `if b.closed` check between the
two RLocks is a narrow concurrency window that can
only fire if Bus.Close races with a Publish goroutine
between the inflight registration and the scan
RLock; reproducing this deterministically would
require injecting a sync point into the production
code path.
noopObserver method bodies still report 0% in cover output despite
being called from TestNoopObserver_Methods. This is a known go
cover quirk for single-statement empty methods that the compiler
inlines; the methods are exercised, just not credited.
All sdk packages still pass `go test -race -count=1`.
Made-with: Cursor
File layout:
- bus.go now holds only the Bus / Subscription / SubOption contract
plus NoopBus. Memory.go keeps strictly the in-process implementation.
- All v0.1.x deprecated symbols (Event / EventType / EventFilter and
every Legacy* type) move to a dedicated deprecated.go so v0.2.0
removal is a single `git rm sdk/event/deprecated*.go`. The matching
bus_test.go is renamed to deprecated_test.go for the same reason.
- Per-symbol Migration godoc snippets replace the single MIGRATION.md
referenced by the original plan.
Route cache (memory.go):
- memSub now stores its pattern pre-split (patternSegs); the Publish
hot path no longer pays strings.Split per (subscription, publish).
- matchSegs is the pure segment matcher shared by Pattern.Matches and
the bus, so the public API behaviour is unchanged.
- MemoryBus memoises subject->[]*memSub via a routeCache guarded by a
secondary RWMutex (lock order: b.mu -> routeCacheMu). Hot subjects
skip the O(N) match scan; the cache is wholesale invalidated on
Subscribe / removeSub / Close so it can never serve stale entries.
- WithRouteCacheSize(n) lets callers size or disable the cache.
- Predicate evaluation stays per-publish since predicates may inspect
the envelope and cannot be safely memoised by Subject alone.
Benchmarks (M2 Max, 1000 subs, hot subject):
before: 9102 ns/op 32 B/op 1 alloc/op
after: 55 ns/op 0 B/op 0 allocs/op (~165x)
Tests:
- memory_route_cache_test.go covers cache hit, Subscribe / Unsubscribe
invalidation, predicate-not-cached, LRU cap eviction, disabled cache,
and a -race goroutine churn scenario.
- memory_bench_test.go adds the cached/uncached benchmarks above.
- Observer godoc gains an explicit ordering note: OnDeliver / OnDrop
order within one Publish is implementation-defined; the cache makes
that order stable per Subject, but observers must not depend on it.
Drive-by: `go fmt` reformatted a godoc block in sdk/kanban/events.go.
Made-with: Cursor
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Reworks
sdk/eventfrom the legacyEvent{Type,…}+EventFilterAPI into a subject-routed publish/subscribe bus that is cross-process ready, fixes long-standing concurrency bugs, and adds a route cache that makes hot-subject dispatch O(1) regardless of subscriber count.The legacy API is kept side-by-side under a
Legacy*prefix (LegacyMemoryBus,LegacyEventBus,Event,EventType,EventFilter, …) withDeprecated:godoc and per-symbolMigration:snippets. Removal is scheduled for v0.2.0 and is physically a singlegit rm sdk/event/deprecated*.go.Downstream
sdk/graph/executorandsdk/kanbanare migrated to the new API in this PR.internal/eventlogbridge and thecontracts/eventgencodegen are deliberately deferred to follow-up PRs (see scope notes below).What's new
New surface (
sdk/event)Bus— minimal interface:Publish(ctx, Envelope) → error,Subscribe(ctx, Pattern, …SubOption) → (Subscription, error),Close() → error. ReturnsErrBusClosedafter shutdown;Closeis idempotent.Envelope— JSON-friendly carrier withjson.RawMessagepayload (zero-copy across in-process and remote implementations), well-known headers (HeaderRunID,HeaderNodeID,HeaderActorID,HeaderGraphID,HeaderTenant), and typed accessors.Subject/Pattern— dot-delimited routing keys with NATS-style wildcards (*single segment,>trailing).Validate()enforces well-formedness.Observer— three-call lifecycle (OnPublish/OnDeliver/OnDrop) replacing the singleWithDropCallback. Invoked outside every bus lock; observers may not call back into the bus.BackpressurePolicy—DropNewest(default) /DropOldest/Block, configured per subscription viaWithBackpressure.NoopBus— the documented zero implementation, kept next to the contract for nil-check ergonomics.MemoryBusimplementation highlightssend-on-closed-channelpanic: signaldonefirst → draininflightand per-subsenderswait-groups → close subscriber channels.senderswait-group before releasing the bus RLock, soClosecan wait deterministically.Subject → []*memSub, guarded by a secondary RWMutex with strict lock orderingb.mu → routeCacheMu. Cache is wholesale invalidated on any subscriber churn, so it can never serve stale entries. Predicates (WithPredicate) are still evaluated per publish because they may inspect the envelope.WithRouteCacheSize(n)lets callers size or disable the cache.Benchmarks (M2 Max, hot subject, no matching subscriber)
Dispatch cost is now decoupled from subscriber count for hot subjects.
File layout
Downstream migration (in this PR)
sdk/graph/executorpublishes viaevent.Bus; subjects formed by helpers insubjects.go(e.g.subjGraphStart,subjNodeStreamDelta).sdk/kanbanpublishes viaevent.Bus;EventKindconstants now hold dot-separated kind values, surfaced both inSubjectand theHeaderKanbanKindenvelope header.Compatibility
Deprecatedwarnings and per-symbolMigration:examples in godoc.LegacyEventBusand the newBusare not assignable to each other; field types must be migrated explicitly.EventFilter.ActorIDis broadcast" semantics is removed in the new API; subscribers should usePattern+WithPredicateinstead.sdk/kanbanEventKindvalues changed to the new dot-form (e.g.kanban.task.submitted); they were not exported across module boundaries before.Tests
sdk/event: subject validation, pattern matching (including malformed>totality), three backpressure policies, observer lock-free contract,Closerace regression, route cache (hit, Subscribe/Unsubscribe invalidation, predicate-not-cached, LRU cap eviction, disabled cache), and a-racegoroutine-churn stress test.sdk/graph/executorandsdk/kanbantest suites updated to assert the new envelope shape.make vet+make test(with-race) green forsdkandsdkxmodules. (examples/voice-pipelinehas a pre-existing unrelatedgo mod tidyissue onmain.)Scope
Included (commits in this PR, oldest → newest):
refactor(sdk/event): rename legacy bus types with Legacy* prefixfeat(sdk/event): introduce Envelope/Subject/Pattern + MemoryBus with Observer and backpressurerefactor(sdk/graph/executor): migrate to new event.Bus APIrefactor(sdk/kanban): migrate to new event.Bus APItest(sdk/event): backfill coverage for low-coverage pathsrefactor(sdk/event): split Bus contract from MemoryBus, add route cacheDeferred to follow-up PRs (kept out to keep this PR reviewable):
cmd/eventgencodegen +contracts/events/*.yamlsubject:field —sdk/event/subjectsto be machine-generated.internal/eventlogbridge rewrite to consume the new bus / Pattern.internal/platform,internal/bootstrap,internal/gatewaysubscription migrations.MIGRATION.mdaggregator + CI guard for deprecated-symbol references.Test plan
cd sdk && go test -race ./event/... -count=10cd sdk && go test -race ./...(all modules)cd sdkx && go test -race ./...cd sdk && go test -bench=BenchmarkPublish -benchmem ./event/...(numbers above)make vetgreenMade with Cursor