feat(messaging): implement NATS messaging integration (topic 7) - #15
Closed
gabriel-farache wants to merge 2 commits into
Closed
feat(messaging): implement NATS messaging integration (topic 7)#15gabriel-farache wants to merge 2 commits into
gabriel-farache wants to merge 2 commits into
Conversation
gabriel-farache
requested review from
LinskId,
NoamNakash,
gpb88,
jordigilh and
tkiss28
July 31, 2026 15:03
PR Summary by QodoImplement NATS JetStream messaging client and wire into agent startup
AI Description
Diagram
High-Level Assessment
Files changed (35)
|
Code Review by Qodo
1. Negative Retry-After backoff
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Comment on lines
+169
to
+173
| func setupMessaging(cfg *config.Config, logger *slog.Logger) (*messaging.Client, string, error) { | ||
| topics := messaging.DeriveTopicNames(cfg.Agent.Name, cfg.Messaging.TopicName) | ||
| if err := messaging.ValidateTopicName(topics.Main); err != nil { | ||
| return nil, "", fmt.Errorf("invalid topic name: %w", err) | ||
| } |
There was a problem hiding this comment.
2. Derived topics unvalidated 🐞 Bug ≡ Correctness
setupMessaging validates only the base topic (topics.Main), but DeriveTopicNames appends ".retry"/".cancel" which can exceed ValidateTopicName’s 255-character limit. This allows a configuration that passes validation to later fail JetStream stream initialization at runtime.
Agent Prompt
## Issue description
Only `topics.Main` is validated, but the client uses derived subjects (`.retry`, `.cancel`) when creating streams. A 255-character base topic passes validation but produces derived subjects >255 characters, causing messaging startup to fail later.
## Issue Context
`DeriveTopicNames()` always appends suffixes; `Client.initStreams()` uses those derived values as JetStream subjects.
## Fix Focus Areas
- cmd/environment-agent/main.go[169-179]
- internal/messaging/topics.go[18-29]
- internal/messaging/topics.go[33-43]
- internal/messaging/client.go[183-201]
## Suggested fix
- Validate *all* derived subjects before constructing/starting the client:
- `ValidateTopicName(topics.Main)`
- `ValidateTopicName(topics.Retry)`
- `ValidateTopicName(topics.Cancel)`
- Alternatively, enforce a stricter max length for the base topic so that derived names remain <=255 (but validating each derived value is clearer).
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Contributor
Author
There was a problem hiding this comment.
Fixed in 6cf52ed and 8555d06 — added ValidateJetStreamSafeName (REQ-MSG-011), which rejects dots and caps the base name so every derived JetStream stream/consumer name (including the longest suffix, -cancel-consumer) stays within NATS's 255-char server-side limit. Wired into setupMessaging in main.go so an unsafe base name fails startup fast instead of only surfacing later as a stream-init failure.
Comment on lines
+124
to
+130
| func (c *Client) doSetup(ctx context.Context, conn *nats.Conn) { | ||
| if c.stopped.Load() { | ||
| return | ||
| } | ||
| c.setupOnce.Do(func() { | ||
| c.setupStreamsAndConsume(ctx, conn) | ||
| }) |
There was a problem hiding this comment.
3. Jetstream setup never retries 🐞 Bug ☼ Reliability
internal/messaging.Client.doSetup guards setup with sync.Once, but setupStreamsAndConsume logs and returns on any initialization error. A transient JetStream/stream/consumer failure permanently disables consumption because later (re)connections can’t rerun setup.
Agent Prompt
## Issue description
`doSetup()` uses `sync.Once`, but `setupStreamsAndConsume()` returns early on errors. If setup fails once (permissions, JetStream temporarily unavailable, stream creation conflict, etc.), the client will never attempt setup again, leaving the agent connected but non-consuming.
## Issue Context
- The connect handler calls `doSetup()`, but the reconnect handler does not.
- Even if reconnect did call it, `sync.Once` would prevent any retry after the first failure.
## Fix Focus Areas
- internal/messaging/client.go[94-107]
- internal/messaging/client.go[124-131]
- internal/messaging/client.go[133-177]
## Suggested fix
- Replace `sync.Once` with an explicit state machine / mutex-protected flag that only flips to “setup complete” after successful initialization.
- On setup failure, schedule retry with backoff (or re-attempt on subsequent reconnect events).
- If you allow retries, ensure you don’t leak duplicate consumers/ConsumeContexts (stop previous ones before starting new ones).
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Contributor
Author
There was a problem hiding this comment.
Fixed in 53d284a — replaced the sync.Once-guarded doSetup with a retryable attemptSetup, so a transient JetStream/stream/consumer failure no longer permanently disables consumption. Same underlying issue as the resolved thread below on client.go:128.
Comment on lines
+9
to
+43
| var validTopicNameRe = regexp.MustCompile(`^[A-Za-z0-9._-]+$`) | ||
|
|
||
| // TopicNames holds the derived main, retry, and cancel topic names. | ||
| type TopicNames struct { | ||
| Main string | ||
| Retry string | ||
| Cancel string | ||
| } | ||
|
|
||
| // DeriveTopicNames derives main/retry/cancel topic names from the agent name | ||
| // and an optional override. If topicNameOverride is non-empty, it takes precedence. | ||
| func DeriveTopicNames(agentName, topicNameOverride string) TopicNames { | ||
| base := agentName | ||
| if topicNameOverride != "" { | ||
| base = topicNameOverride | ||
| } | ||
| return TopicNames{ | ||
| Main: base, | ||
| Retry: base + ".retry", | ||
| Cancel: base + ".cancel", | ||
| } | ||
| } | ||
|
|
||
| // ValidateTopicName validates that the given name conforms to NATS subject token rules. | ||
| func ValidateTopicName(name string) error { | ||
| if name == "" { | ||
| return errors.New("topic name must not be empty") | ||
| } | ||
| if len(name) > 255 { | ||
| return errors.New("topic name exceeds 255 characters") | ||
| } | ||
| if !validTopicNameRe.MatchString(name) { | ||
| return errors.New("topic name contains invalid characters (allowed: alphanumeric, hyphens, dots, underscores)") | ||
| } | ||
| return nil |
There was a problem hiding this comment.
4. Empty subject tokens allowed 🐞 Bug ≡ Correctness
internal/messaging.ValidateTopicName claims to validate “NATS subject token rules” but only checks a character whitelist, so it accepts names with empty tokens like ".foo", "foo." or "foo..bar". These accepted values are then used as NATS/JetStream subjects during stream setup.
Agent Prompt
## Issue description
`ValidateTopicName()` does not actually enforce non-empty dot-separated tokens even though its comment says it validates subject token rules. This allows malformed subjects that are likely to fail later during stream/consumer setup.
## Issue Context
The current validation only checks:
- non-empty string
- length <= 255
- characters in `[A-Za-z0-9._-]`
## Fix Focus Areas
- internal/messaging/topics.go[9-43]
## Suggested fix
- Add token validation, e.g.:
- reject strings starting/ending with `.`
- reject `..` anywhere
- optionally split on `.` and ensure every token is non-empty
- Keep wildcard rejection (already implicit via the regex) as-is.
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Contributor
Author
There was a problem hiding this comment.
Fixed in 781ee13 — ValidateTopicName now rejects leading/trailing dots and .. (empty dot-separated tokens), covered by UT-MSG-047/048.
gabriel-farache
force-pushed
the
feat/topic7-messaging-integration
branch
2 times, most recently
from
July 31, 2026 19:21
7b6ba0c to
df8e816
Compare
Contributor
Author
|
/agentic_review |
Comment on lines
+159
to
+162
| cc, err := cancelCons.Consume(func(msg jetstream.Msg) { | ||
| c.handleCancelMessage(msg) | ||
| _ = msg.Ack() | ||
| }) |
There was a problem hiding this comment.
1. Cancel always acked 🐞 Bug ≡ Correctness
Both cancel-topic and main-topic consumers acknowledge JetStream messages even when CloudEvent parsing/JSON decoding fails or handlers fail, meaning invalid or transiently failing requests are permanently dropped and may proceed without deny-list enforcement. This also allows publishing acknowledgements with an empty resourceId, breaking deny-list matching and preventing retries.
Agent Prompt
## Issue description
JetStream messages for both cancel-topic and main-topic handling are effectively treated as successfully processed even when CloudEvent parsing/JSON decoding fails, `resourceId` is missing/empty, or downstream handlers fail. Because the consumer still `Ack()`s these messages, invalid events and transient errors are not retried, cancels can be dropped without updating the deny-list, and responses can be published with an empty `resourceId`.
## Issue Context
- `handleCancelMessage` can exit early on JSON/CloudEvent parse errors and ignores the return value from `cancelHandler`, but the consumer callback still acknowledges the message.
- `handleMainMessage` ignores the `json.Unmarshal(event.Data(), &payload)` error and then uses `payload.ResourceID` directly (even if empty) to drive deny-list deletion, invoke `mainHandler`, and publish a response, after which the message is Ack()ed.
## Fix Focus Areas
- internal/messaging/client.go[159-162]
- internal/messaging/client.go[232-250]
- internal/messaging/handlers.go[22-41]
- internal/messaging/handlers.go[59-82]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Contributor
Author
There was a problem hiding this comment.
Fixed in 53d284a — both handleMainMessage and handleCancelMessage now Nak/NakWithDelay on parse or handler failure and only Ack after the handler succeeds, so invalid/failed messages are redelivered instead of silently dropped.
|
Code review by qodo was updated up to the latest commit df8e816 |
gabriel-farache
force-pushed
the
feat/topic7-messaging-integration
branch
2 times, most recently
from
August 4, 2026 08:04
18fcc8c to
9103259
Compare
8 tasks
jordigilh
reviewed
Aug 5, 2026
jordigilh
reviewed
Aug 5, 2026
| if cons == nil { | ||
| return 0 | ||
| } | ||
| return int64(cons.CachedInfo().NumPending) |
There was a problem hiding this comment.
NumPending is JetStream's undelivered-backlog count, not the unacknowledged-in-flight count. REQ-DCM-150/AC-DCM-090 define consumerLag as "unacknowledged messages," which maps to NumAckPending, not NumPending — they diverge exactly when the agent has pulled/delivered messages but is slow to process them, which is the backpressure scenario this metric exists to catch. AC-090's integration test only exercises a hand-set stubConsumerLagProvider{lag: 5}, so this mismatch against real JetStream semantics isn't caught by any test. Worth confirming which field DCM's consumer actually wants before this ships.
Contributor
Author
There was a problem hiding this comment.
Fixed in 21e4af3 — ConsumerLag now sums NumPending + NumAckPending (not-yet-delivered + delivered-but-unacked), matching REQ-DCM-150's definition of unacknowledged rather than just the undelivered backlog.
chadcrum
reviewed
Aug 6, 2026
| conn, err := nats.Connect(c.cfg.URL, | ||
| nats.RetryOnFailedConnect(true), | ||
| nats.MaxReconnects(-1), | ||
| nats.ReconnectWait(2*time.Second), |
There was a problem hiding this comment.
Reconnect wait is a fixed 2s (nats.ReconnectWait(2*time.Second) with MaxReconnects(-1)). REQ-MSG-100 wants the same exponential backoff as REQ-DCM-050: min(initial × 2^attempt, max) with full jitter.
DCM already does this via internal/backoff (CalculateBackoff + ApplyJitter); messaging doesn't. Not saying reconnect is broken - it works - but it doesn't match the agreed policy.
Could wire this through nats.CustomReconnectDelay using those same helpers (same initial/max as DCM, or messaging-specific equivalents if that's preferred).
Contributor
Author
There was a problem hiding this comment.
Fixed in 53d284a — replaced nats.ReconnectWait(2*time.Second) with nats.CustomReconnectDelay(c.reconnectDelay), using the same internal/backoff exponential-with-full-jitter helpers as REQ-DCM-050 (REQ-MSG-100 explicitly cross-references it).
machacekondra
approved these changes
Aug 6, 2026
gabriel-farache
force-pushed
the
feat/topic7-messaging-integration
branch
2 times, most recently
from
August 6, 2026 13:56
6b068ad to
a1886fa
Compare
gabriel-farache
marked this pull request as draft
August 7, 2026 08:32
gabriel-farache
force-pushed
the
feat/topic7-messaging-integration
branch
2 times, most recently
from
August 11, 2026 07:32
227e6bc to
d065c08
Compare
gabriel-farache
marked this pull request as ready for review
August 11, 2026 07:47
Add UT-MSG-037–041 (whitelist regex edge cases) to unit test plan and IT-MSG-071–073 (nested payload, delete response, publish error nak) to integration test plan. Update traceability matrices accordingly. Assisted by: Claude Code - claude-4.6-opus Signed-off-by: gabriel-farache <gfarache@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com>
Add CloudEvent builder (internal/cloudevent) with FormatSource/FormatType helpers and NewResponseCE constructor. Implement NATS JetStream messaging client (internal/messaging) with topic subscription, handler dispatch, response publishing, and cancel-request deny-list management. Wire messaging setup into main via setupMessaging helper. Includes whitelist-based topic validation, typed CE payload extraction, compile-time interface assertions, and separated handler file. Config extended with NATS and messaging fields. Full unit and integration test coverage with test suites. Assisted by: Claude Code - claude-4.6-opus Signed-off-by: gabriel-farache <gfarache@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com>
gabriel-farache
force-pushed
the
feat/topic7-messaging-integration
branch
from
August 11, 2026 07:48
d065c08 to
9d99214
Compare
|
Code review by qodo was updated up to the latest commit d065c08 |
gciavarrini
reviewed
Aug 11, 2026
gciavarrini
left a comment
gciavarrini
left a comment
Contributor
There was a problem hiding this comment.
Some lines in the PR description do not match the diff.
FormatType and NewResponseCE are not in internal/cloudevent. DeriveTopicNames builds main, retry, and cancel, not a response topic (responses use the fixed dcm.agents.responses subject).
Prefer a shorter summary that matches what this PR adds
Contributor
Author
|
#19 is covering all of this and as it's the terminal PR, it has some hardening so closing this PR in favour of the terminal one. |
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
5 participants
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
Implements the NATS JetStream messaging subsystem for the environment agent (spec §7):
internal/cloudevent):NewCloudEventconstructor andFormatSourcehelper for building spec-compliant CloudEventsinternal/messaging): NATS JetStream connection lifecycle, topic subscription with handler dispatch, response CE publishing, and cancel-request deny-list managementDeriveTopicNameshelper for main/retry/cancel topic derivation (response CEs use the fixeddcm.agents.responsessubject)setupMessaginghelper replaces stub implementations (messagingStatus,stubConsumerLagProvider) with realmessaging.Client— satisfiesIsConnected()andConsumerLag()interfaces for health and DCM registrarFiles changed (topic 7 only)
internal/cloudevent/builder.go,builder_test.go,suite_test.gointernal/messaging/client.go,handlers.go,topics.go,topics_test.go,suite_test.go,client_integration_test.gointernal/config/config.go,config_test.gocmd/environment-agent/main.go,main_test.goMakefilego.mod,go.sum.ai/test-plans/*.md+1,602 −43 across 18 files.
Dependencies
Test plan
make test-unit— messaging unit tests (topic validation, CE builder) passmake test-integration— messaging integration tests (NATS JetStream subscribe/publish/nak lifecycle) passmake lint— no new lint issuesmessagingStatus,stubConsumerLagProvider) are fully removed frommain.goMade with Cursor