feat(eventbus): B.3a — system-event-bus (13/13 ACs, 100%)#423
Merged
Conversation
3 tasks
remyluslosius
force-pushed
the
feat/slice-b-b3a-event-bus
branch
from
26b8d26 to
c528325
Compare
Opens Slice B.3 (orchestration plumbing). Typed in-process pub/sub for
OpenWatch-internal events, carrying Bucket B events per the
Kensa/OpenWatch boundary doc § 4. Foundation for B.3b's alert router.
Spec
New: app/specs/system/event-bus.spec.yaml (status: approved).
13 ACs across 9 constraints.
internal/eventbus package
doc.go Architectural choices: in-process via Go channels only,
typed Event interface + closed EventKind enum, per-
subscriber goroutine (slow subscriber doesn't block
others), Shutdown drains + closes.
types.go EventKind closed enum (HeartbeatPulse, DriftDetected).
Event interface (Kind, Timestamp).
HeartbeatPulse struct (HostID, Reachable, OccurredAt,
PriorReachable, ResponseTimeMS).
DriftDetected struct (HostID, ScanID, DriftType,
ScoreDelta + per-severity transition counts).
DefaultBufferSize = 1024 (spec C-04).
SubscribeOptions with BufferSize + Kinds filter.
bus.go Bus struct holding mu (RWMutex), subscribers map,
closed atomic.Bool, metrics.
Publish: non-blocking via select-with-default; per-
subscriber drop+count on full channel.
Subscribe: returns a Subscription registered for the
given kinds.
Shutdown: closes every subscriber channel; subsequent
Publish is a no-op.
Subscription with Events() <-chan, Delivered/Dropped
counts, Unsubscribe (close-once via sync.Once).
Metrics: PublishedCount, DeliveredCount, DroppedCount,
NoSubscribersCount; Snapshot returns typed struct.
ACs covered (13 of 13)
AC-01 One subscriber receives the event within 100ms
AC-02 Zero subscribers → drop silently + NoSubscribersCount++
AC-03 Three subscribers all receive
AC-04 1000 concurrent publishes × 10 subscribers race-clean
AC-05 Shutdown closes channels; post-Shutdown Publish is a no-op
AC-06 BufferSize=1 + 5 publishes without read → 5 dropped
AC-07 EventKind enum has exactly 2 values (AllEventKinds)
AC-08 Heartbeat-only subscriber does NOT receive DriftDetected
AC-09 All counters round-trip through publish/drop scenarios
AC-10 1000 publishes × 10 subscribers under 100ms wall-clock
AC-11 Slow subscriber doesn't block fast subscriber (100 events
fully received by fast despite slow being starved)
AC-12 Source-inspection: no kafka/nats/rabbitmq/redis-pubsub/
GCP-pubsub/SNS/SQS imports
AC-13 Unsubscribe closes channel; subsequent Publish doesn't
deliver and doesn't panic
Local validation
go build ./internal/eventbus/ clean
go vet ./internal/eventbus/ clean
go test -race ./internal/eventbus/ 13 sub-tests pass
specter coverage system-event-bus 13/13 = 100%
Architectural choices worth flagging
- Non-blocking Publish via select with default. A subscriber whose
buffer is full has the event dropped + DroppedCount increments;
other subscribers still receive. This is the load-shedding policy.
- Per-subscriber state in Subscription (delivered/dropped atomic
counters) so consumers can self-monitor without the bus knowing
who they are.
- Closing a subscriber channel exactly once via sync.Once on the
Subscription, with a closed-flag guard on the Bus to avoid the
double-close when Shutdown and Unsubscribe race.
- The Event interface is the wire format; payloads are typed Go
structs that subscribers type-assert. This preserves type safety
on both sides without erasure to []byte or map[string]any.
Slice B.3 status
B.3a event bus this PR — 13/13 ACs
B.3b alert router pending (subscribes to this bus)
remyluslosius
force-pushed
the
feat/slice-b-b3a-event-bus
branch
from
c528325 to
2495b15
Compare
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
Slice B.3a —
system-event-busimplementation. Opens Slice B.3 (orchestration plumbing). Typed in-process pub/sub for OpenWatch-internal events; foundation for B.3b's alert router.-raceWhat landed
app/specs/system/event-bus.spec.yamlinternal/eventbus/Architectural choices locked
Eventinterface — Kind() + Timestamp() — and closedEventKindenum (currentlyHeartbeatPulse+DriftDetected). Subscribers filter by Kind at registration; no wildcardPublish— select-with-default pattern; subscriber whose buffer is full has the event dropped +DroppedCountincrements. Other subscribers still receiveShutdowndrains + closes — subsequent Publish is a no-op without panicSubscriptionso consumers can self-monitor without the bus knowing who they areACs satisfied
Local validation
go build ./internal/eventbus/— cleango vet ./internal/eventbus/— cleango test -race ./internal/eventbus/— 13 sub-tests passspecter coverage— system-event-bus 13/13 = 100%Slice B.3 status
Relationship to other PRs
HeartbeatPulseto this bus on state transitionsDriftDetectedto this bus on non-stable outcomesSlice B running total
Slice B totals so far: 88 ACs across 6 specs, all at 100%.