Conversation
Implements Task 23 (PR 7) — three typed module factories for the workflow-plugin-eventbus gRPC plugin: - module.go / module_test.go: infra.eventbus — validates provider × deploy_target using the existing conformance matrix, registers ClusterConfig in a global registry, resolves broker URI from EVENTBUS_<NAME>_URI or NATS_URL env vars for runtime step use. - stream.go / stream_test.go: infra.eventbus.stream — validates name + subjects, registers StreamConfig for consume and trigger lookups. - consumer.go / consumer_test.go: infra.eventbus.consumer — validates name + stream_name, registers ConsumerConfig for step.eventbus.consume and trigger.eventbus.subscribe. All three types implement sdk.ModuleInstance with compile-time assertions (var _ sdk.ModuleInstance = (*T)(nil)). Zero map[string]any — all config boundaries use typed proto pointers (proto messages embed sync.Mutex via protoimpl.MessageState). 19 unit tests, all passing. Adds github.com/GoCodeAlone/workflow v0.20.1 as a direct dependency for the sdk.ModuleInstance interface. wfctl audit plugins -strict-contracts: module 3/3 strict, step 3/3 strict, trigger 1/1 strict — no findings on workflow-plugin-eventbus. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Addresses 5 code-reviewer findings on Task 23: Important #1 — Missing sdk.TypedModuleProvider compile-time assertions: Added ClusterModuleFactory, StreamModuleFactory, ConsumerModuleFactory — each a named struct with 'var _ sdk.TypedModuleProvider = (*T)(nil)'. CreateTypedModule unpacks anypb.Any and delegates to the NewXModule constructor. Returns sdk.ErrTypedContractNotHandled for wrong types so the plugin's routing loop can skip. Important #2 — deploy_target not explicitly validated before ValidateProviderTarget: added an explicit empty-string check for config.deploy_target in NewClusterModule, producing a clear error message distinct from the provider × target compatibility error. Important #3 — NATS connection lifecycle: module did not own the connection it declared. Added natsConnCache (sync.Mutex-guarded map[string]*nats.Conn), GetOrDialNATSConn (lazy dial via injectable natsDialFn), RegisterNATSConn / GetNATSConn / closeNATSConn helpers. Stop() now calls closeNATSConn, which closes and evicts the cached connection (nil-guarded, idempotent). natsDialFn is a package-level var so integration tests and unit tests can inject without a real server. Minor #4 — consumer goroutine leak concern: explicit comment in consumerModule.Start documenting that no goroutines are launched — consumption is pull-based, driven by step.eventbus.consume. Minor #5 — factory tests missing: added TypedModuleProvider tests for all three factory types (TypedModuleTypes, WrongType, NilConfig paths). All 28 tests pass. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
BLOCKER-A (module_test.go): Replace bare os.Unsetenv calls in
TestClusterModule_InitNoURIWhenEnvNotSet with LookupEnv + t.Cleanup
restore pattern. Bare Unsetenv permanently cleared NATS_URL for all
subsequent tests in the run; the cleanup now restores any pre-existing
value on exit.
BLOCKER-B (module.go): Eliminate connCacheMu→urlMu nested lock ordering
in GetOrDialNATSConn by extracting the URI lookup between the fast-path
unlock and the slow-path re-lock. Pattern:
1. Lock, check cache, unlock (fast path — no urlMu touch).
2. GetBusURI with no lock held (urlMu only, no nesting).
3. Dial with no lock held.
4. Lock, double-check (another goroutine may have won), insert or
discard the redundant connection, unlock.
Added lock-ordering doc comment to the NATS cache section.
MINOR-4 (all three *_test.go): Add if err != nil { t.Fatalf } after
NewXModule constructors in StopUnregisters and StopEvictsNATSConn tests
so a nil module doesn't panic on the next line.
All 28 tests pass, -race clean.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- steps/publish.go: PublishHandler uses JetStream PublishMsg, returns broker sequence + RFC3339 acked_at timestamp - steps/consume.go: ConsumeHandler binds to existing durable JetStream consumer via BindStream; maps msg.Reply→ack_token, Metadata→sequence and published_at; ErrTimeout is treated as empty batch (not an error) - steps/ack.go: AckHandler publishes empty payload to ack_token (the JetStream reply subject), acknowledging the message - steps/factories.go: TypedStepFactory vars + All() slice for plugin server registration; compile-time TypedStepProvider assertions - module.go: add DefaultBusConn() — single-bus helper for step handlers - consumer.go: add GetConsumerByName() — resolve ConsumerConfig by durable name (cfg.name) for step.eventbus.consume lookup All 9 new step unit tests pass alongside existing 24 module tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…eview CRITICAL: pass nats.Context(ctx) to nc.JetStream() in publish.go and consume.go so that context cancellation/deadline is respected by both js.PublishMsg and sub.Fetch — previously ctx was dead code in all JetStream calls. IMPORTANT-1: sort registered cluster names before selecting default in DefaultBusConn(); eliminates non-deterministic map iteration that would route to a random broker in multi-bus deployments. IMPORTANT-2: cap batch_size at maxBatchSize=1000 in ConsumeHandler; prevents a caller from passing 2^31-1 and blocking fetch indefinitely. MINOR-3: clarify sub.Drain() nolint comment — "ephemeral per-fetch" explains why best-effort is the right posture for a PullSubscribe. MINOR-4: add explicit invariant comment to TestPublishHandler_NoBusRegistered naming the t.Cleanup guarantee that makes the test's assumption safe. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…t + integration test - trigger.go: SubscribeTriggerModuleFactory (TypedModuleProvider) + subscribeTrigger (ModuleInstance + TriggerInstance) with bounded goroutine, clean shutdown via context cancel + done channel, fetchPollInterval backpressure on idle streams. cb=nil path is a no-op Start (external plugin gRPC transport). - trigger_test.go: 7 unit tests covering factory, validation, and nil-callback lifecycle. - cmd/workflow-plugin-eventbus/main.go: eventbusPlugin wires all 4 module factories (cluster, stream, consumer, trigger), 3 step factories (publish, consume, ack), TriggerProvider, and ContractProvider returning all 7 strict-proto descriptors. - integration_test.go: real gRPC subprocess transport — compiles binary, starts via go-plugin, verifies manifest, contract registry (7 contracts, all STRICT_PROTO), infra.eventbus module lifecycle, publish/consume/ack step error paths over wire, trigger module lifecycle, and GetModuleTypes/GetStepTypes/GetTriggerTypes RPCs. No live NATS server required; exercises descriptive error paths. All tests pass (GOWORK=off go test ./... -count=1). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Failure 1 — plugin.go missing:
- Split cmd/workflow-plugin-eventbus/main.go into main.go (entrypoint only:
sdk.Serve(&eventbusPlugin{})) and plugin.go (package main: eventbusPlugin
type, compile-time assertions, moduleFactories, stepFactories, and all
interface method impls: Manifest, TypedModuleTypes, CreateTypedModule,
TypedStepTypes, CreateTypedStep, TriggerTypes, CreateTrigger, ContractRegistry).
Failure 2 — integration_test.go wrong gate + missing NATS scenario:
- TestE2E_EventbusPluginScenario: removed testing.Short() gate — runs always.
Extracted declareModule helper (Create→Init→Start+Cleanup) to reduce repetition.
- TestE2E_EventbusPluginScenario_WithNATS: new test gated on INTEGRATION_NATS=1.
Reads NATS_URL from env (inherited by subprocess). Pre-creates JetStream stream
+ durable consumer directly via nats.go in test process. Declares infra.eventbus,
infra.eventbus.stream, infra.eventbus.consumer modules via gRPC. Publishes 10
messages via step.eventbus.publish (each ExecuteStep call over gRPC wire).
Consumes all 10 via step.eventbus.consume batch_size=10. Acks each via
step.eventbus.ack using ack_token from ConsumeResponse. Asserts len(messages)==10.
All tests pass (GOWORK=off go test ./... -count=1). WithNATS test skips correctly.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… review
CRITICAL — callback map didn't match Message proto contract (trigger.go):
fetchAndFire now builds *eventbusv1.Message the same way ConsumeHandler does:
Payload=m.Data ([]byte, not string), AckToken=m.Reply (not "reply"), plus
Headers (map[string]string from NATS headers), Sequence (stream seq as string),
PublishedAt (RFC3339 UTC). Callback data map keys are the six proto field names:
subject, payload, headers, sequence, published_at, ack_token.
IMPORTANT-1 — zero test coverage for fetchAndFire (trigger_test.go):
Added four new tests using an embedded NATS server (nats-server/v2, already an
indirect dep, now explicit in go.mod):
- TestSubscribeTrigger_FetchAndFire_CallbackData: publishes 1 msg with a custom
header, asserts all six data map fields are present with correct Go types
([]byte for payload, string for ack_token — catches the CRITICAL regression).
- TestSubscribeTrigger_FetchLoop_ExitsOnCancel: Stop returns within 5s after
Start; goroutine exits cleanly on ctx cancel.
- TestSubscribeTrigger_FetchLoop_RetryOnError: publishes after Start; loop keeps
polling until the message arrives (retry path).
- TestSubscribeTrigger_DoubleStart: second Start returns error; first goroutine
exits cleanly on Stop (double-start guard test).
IMPORTANT-2 — double-Start goroutine leak (trigger.go):
Start() now returns an error if t.cancel != nil (already started).
MINOR-3 — time.After allocation in retry backoff (trigger.go):
Replaced time.After with time.NewTimer + backoff.Reset so a single timer is
reused across all retry iterations in fetchLoop.
MINOR-4 — CreateTrigger silent type assertion discard (plugin.go):
Added configString() helper that returns an explicit error when a config key
is present but not a string (e.g. config["name"]=42 → "config[name] must be
a string, got int"), rather than silently returning "" and confusing callers.
All tests pass (GOWORK=off go test ./... -count=1).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… nit) - module.go: export UnregisterNATSConn(instanceName) — evicts the entry from natsConnCache without closing the connection (for tests that own the conn lifetime via nc.Close() + embedded-server shutdown). - trigger_test.go: replace the three duplicate UnregisterBusURI cleanups that followed RegisterNATSConn with UnregisterNATSConn, ensuring natsConnCache is cleaned between tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
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
infra.eventbus/infra.eventbus.stream/infra.eventbus.consumer— three typed module providers (ClusterModuleFactory,StreamModuleFactory,ConsumerModuleFactory) implementingsdk.TypedModuleProvider; NATS connection cache with double-checked locking;DefaultBusConn()with deterministic lexicographic sort; URI resolution fromEVENTBUS_<NAME>_URI/NATS_URLenv varsstep.eventbus.publish/step.eventbus.consume/step.eventbus.ack— three typed step handlers insteps/usingsdk.NewTypedStepFactory;nats.Context(ctx)forwarded to JetStream; batch cap at 1000;GetConsumerByNamefor durable consumer lookup; ack via NATS reply subjecttrigger.eventbus.subscribe—SubscribeTriggerModuleFactory+subscribeTrigger(implements bothModuleInstanceandTriggerInstance); bounded goroutine with context cancel + done-channel shutdown;fetchPollInterval=2sMaxWait cap;time.NewTimerbackoff reuse; double-Start guard; callback data map matchesMessageproto (subject, payload[]byte, headers, sequence, published_at, ack_token)cmd/workflow-plugin-eventbus/main.go+plugin.go—eventbusPluginwires all factories viasdk.Serve();ContractProviderreturns 7 strict-proto descriptors matchingplugin.contracts.jsonintegration_test.go— real gRPC subprocess transport (no mocks):TestE2E_EventbusPluginScenario(always runs, no NATS needed) +TestE2E_EventbusPluginScenario_WithNATS(gated onINTEGRATION_NATS=1, publishes 10 msgs → consumes 10 → acks 10 over wire)trigger_test.go— 11 unit tests incl. 4 using embedded NATS server (nats-server/v2): callback data contract, goroutine exit on cancel, retry path, double-start guardTest plan
GOWORK=off go test ./... -count=1 -race— all packages greenGOWORK=off go test -run TestE2E_EventbusPluginScenario -v .— gRPC subprocess test passes (~10s)GOWORK=off go test -run TestSubscribeTrigger_FetchAndFire_CallbackData -v .— embedded NATS callback contract test passesINTEGRATION_NATS=1 NATS_URL=nats://localhost:4222 GOWORK=off go test -run TestE2E_EventbusPluginScenario_WithNATS -v .— full publish/consume/ack scenario (requires live NATS with JetStream)🤖 Generated with Claude Code