feat(routing): implement resource operation routing (topic 8)

[gabriel-farache]

Summary

• Implement resource operation routing — route CloudEvent create/delete/cancel requests to the correct service provider based on service type, with health-aware dispatch, retry with exponential backoff + jitter, deny-list cancel semantics, and dispatched-set tracking
• Harden routing with bug fixes and security improvements — fix cancel ack-on-error (BC-1), poison pill ack-drop (BC-2), cancel error propagation with deny-list safety (BC-3), HTTP 408 non-retryable (BC-4), dispatched-set pre-marking with defer cleanup (BC-5), cancel for never-seen resourceId (BC-6), bounded dispatched set (SEC-1), DenyList maxSize guard (SEC-2), sanitized CE error details (SEC-3), payload validation (SEC-4), atomic AddIfAbsent to fix TOCTOU race, and context-aware retry sleep
• Refactor for maintainability — decompose HandleRequest into focused helpers, extract types/interfaces to types.go, create generic PublishCE helper in cloudevent package, extract CE constants to shared package, remove legacy messaging handler paths

Prerequisites

#15 to be merged

Spec conformance

All requirements in Section 4.8 (REQ-RTE-030 through REQ-RTE-180) are satisfied. Integration tests cover IT-RTE-010 through IT-RTE-130 plus edge cases (TC-7).

Review coverage

Implementation was reviewed by 6 specialized agents across 4 models (Opus 4.6, Sonnet 5, Grok 4.5, Codex 5.3) covering correctness, security, code structure, test coverage, concurrency, and cross-package integration. All Must Fix and Should Fix findings were resolved.

Test plan

• make fmt — no formatting changes
• make lint — 0 issues
• make build — compiles cleanly
• make test — all suites pass (routing: 26 specs, 18 integration + 8 unit)
• make test-race — no race conditions detected
• Routing integration tests verify CE payload structure for all 16 response paths
• Edge case tests: delete bypasses deny list, cancel before any request
• Cancel rejected for in-flight provisioning

Made with Cursor

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (1) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Health key mismatch ✓ Resolved 🐞 Bug ≡ Correctness

Description

Router.resolveProvider calls HealthTracker.GetState(providerName), but HealthTracker state is stored
under StoredProvider.ID, so providers typically appear Unavailable and requests won’t be
forwarded/queued correctly.

Code

internal/routing/router.go[R199-202]

+	state, found := r.healthTracker.GetState(providerName)
+	if !found {
+		return sp, v1alpha1.Unavailable, true
+	}

Evidence

The router uses the registry’s providerName as the health key, but HealthTracker is defined and used
elsewhere as providerID-keyed state (set with StoredProvider.ID). Routing tests currently set state
by name, masking the mismatch.

internal/routing/router.go[186-204]
internal/provider/health_state.go[16-47]
internal/provider/service/service.go[252-258]
internal/provider/service/service.go[359-364]
internal/routing/routing_integration_test.go[302-314]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`routing.Router.resolveProvider` checks provider health using the provider *name* (`providerName` from the registry), but the health subsystem stores and queries health state by provider *ID* (`StoredProvider.ID`). This causes the router to miss health state and treat normal providers as `Unavailable`.

### Issue Context
- Registry maps `serviceType -> providerName`.
- Store record contains both `Name` and `ID`.
- HealthTracker API is explicitly keyed by `providerID`.

### Fix Focus Areas
- internal/routing/router.go[186-203]

### What to change
- After loading `sp` from the store, call `r.healthTracker.GetState(sp.ID)` (not `providerName`).
- Update routing tests/helpers that currently set health state under provider name to set it under the stored provider ID, so tests match production behavior.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Duplicate dispatch on redelivery 🐞 Bug ☼ Reliability

Description

forwardWithRetry adds the resourceId to dispatchedSet but does not short-circuit when it already
exists, so JetStream redeliveries can call CreateResource/DeleteResource multiple times and
duplicate side effects.

Code

internal/routing/router.go[R211-214]

+	newlyAdded := r.dispatchedSet.AddIfAbsent(payload.ResourceID)
+
+	// TOCTOU mitigation: re-check denyList after claiming the dispatchedSet
+	// slot. A concurrent HandleCancel may have added to denyList between

Evidence

The router records dispatched-set membership but continues forwarding regardless of whether the
resourceId was already present. Separately, the JetStream handlers ignore Ack errors, making it
possible for a message to be redelivered after successful processing.

internal/routing/router.go[206-223]
internal/messaging/handlers.go[12-26]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`forwardWithRetry` uses `dispatchedSet.AddIfAbsent(resourceId)` but ignores the `false` result and proceeds to forward anyway. If the same message is delivered again (redelivery/duplicate delivery), the router can re-run the SP operation.

### Issue Context
- JetStream consumers can redeliver messages (including when `Ack()` fails).
- `messaging.Client` currently ignores `Ack()` errors, making successful-processing-without-ack a plausible path.

### Fix Focus Areas
- internal/routing/router.go[206-223]
- internal/messaging/handlers.go[12-26]

### What to change
- In `forwardWithRetry`, if `newlyAdded == false`, treat it as a duplicate delivery for an already-dispatched `resourceId` and return without calling the SP forwarder (optionally republish the corresponding ack/error CE if required by your delivery semantics).
- Consider handling `msg.Ack()` errors in `messaging.Client` (at least log them, and/or convert them into a failure path that triggers a retry) to reduce duplicate deliveries after side effects have occurred.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. Empty provider ID collision ✓ Resolved 🐞 Bug ☼ Reliability ⭐ New

Description

Router.resolveProvider now looks up provider health using sp.ID without validating it is non-empty;
if multiple persisted provider records have ID=="", their health state shares the same empty-string
key and routing decisions can be made using the wrong provider’s health. This is most likely to
surface in upgrade/corrupt-store scenarios (not normal registration) and can lead to incorrect
forward/queue decisions.

Code

internal/routing/router.go[R199-200]

+	state, found := r.healthTracker.GetState(sp.ID)
+	if !found {

Evidence

The router now uses sp.ID directly as the health-tracker lookup key, but StoredProvider.ID is
just a string field with no structural guarantee it’s non-empty; persisted providers are also loaded
and tracked using their stored ID as-is. If more than one stored provider has an empty ID, they
collide on the same health key (""), making routing depend on whichever health update last wrote
that key.

internal/routing/router.go[186-204]
internal/provider/store/store.go[19-32]
internal/provider/service/service.go[234-249]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`resolveProvider()` queries the health tracker with `sp.ID` but does not guard against `sp.ID == ""`. If the store contains malformed/legacy provider records with empty IDs, multiple providers can collide on the empty-string key in `HealthTracker`, causing health state contamination and incorrect routing.

### Issue Context
- `HealthTracker` is keyed by `providerID string`.
- Persisted providers are loaded and tracked using whatever `StoredProvider.ID` contains.

### Fix Focus Areas
- Add a defensive guard in `resolveProvider()` to treat `sp.ID == ""` as invalid (log + return ok=false or treat as Unavailable).
- Consider adding validation/quarantine during persisted provider loading so empty IDs never enter health tracking.

- internal/routing/router.go[188-203]
- internal/provider/service/service.go[234-248]
- internal/provider/store/store.go[19-32]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

4. Ack warning lacks message ID ✓ Resolved 🐞 Bug ◔ Observability ⭐ New

Description

The new Ack() failure warnings only log the error, so when ack failures occur (and redelivery may
occur) operators cannot correlate the warning to a specific CloudEvent/message. This reduces the
usefulness of these logs for diagnosing duplicate side effects.

Code

internal/messaging/handlers.go[R27-29]

+	if err := msg.Ack(); err != nil {
+		c.logger.Warn("failed to ack main message, may be redelivered", "error", err)
+	}

Evidence

The added Ack-failure logs only emit the error field, providing no message/CloudEvent identity;
meanwhile DD-190 references ack-error logging as an operational visibility mechanism, which is less
actionable without correlation fields.

internal/messaging/handlers.go[12-29]
.ai/decisions/environment-agent.decisions.md[254-267]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
On `msg.Ack()` failure, the warning log includes only the error. Without a per-message correlation key (e.g., CloudEvent ID), it’s hard to identify which message is affected during investigations.

### Issue Context
DD-190 explicitly mentions ack-error logging as an operational signal, so improving correlation value makes that signal more actionable.

### Fix Focus Areas
- In the `Ack()` error branches, extract a lightweight identifier from `msg.Data()` (e.g., unmarshal just enough to get CloudEvent `id` / `type`, and optionally `resourceId` if present) and include it in the warning fields.
- Apply consistently to both cancel and main handlers.

- internal/messaging/handlers.go[12-29]
- .ai/decisions/environment-agent.decisions.md[254-267]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

Tip of the day

💡 Did you know, you can group findings by type and pick your Finding display, from Minimal to Full

More tips ↗ | Customize Qodo ↗ | Qodo docs ↗

Previous review results

Review updated until commit 72d0095

Results up to commit 2a27d70 ⚖️ Balanced

🐞 Bugs (1) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Health key mismatch ✓ Resolved 🐞 Bug ≡ Correctness

Description

Router.resolveProvider calls HealthTracker.GetState(providerName), but HealthTracker state is stored
under StoredProvider.ID, so providers typically appear Unavailable and requests won’t be
forwarded/queued correctly.

Code

internal/routing/router.go[R199-202]

+	state, found := r.healthTracker.GetState(providerName)
+	if !found {
+		return sp, v1alpha1.Unavailable, true
+	}

Evidence

The router uses the registry’s providerName as the health key, but HealthTracker is defined and used
elsewhere as providerID-keyed state (set with StoredProvider.ID). Routing tests currently set state
by name, masking the mismatch.

internal/routing/router.go[186-204]
internal/provider/health_state.go[16-47]
internal/provider/service/service.go[252-258]
internal/provider/service/service.go[359-364]
internal/routing/routing_integration_test.go[302-314]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`routing.Router.resolveProvider` checks provider health using the provider *name* (`providerName` from the registry), but the health subsystem stores and queries health state by provider *ID* (`StoredProvider.ID`). This causes the router to miss health state and treat normal providers as `Unavailable`.

### Issue Context
- Registry maps `serviceType -> providerName`.
- Store record contains both `Name` and `ID`.
- HealthTracker API is explicitly keyed by `providerID`.

### Fix Focus Areas
- internal/routing/router.go[186-203]

### What to change
- After loading `sp` from the store, call `r.healthTracker.GetState(sp.ID)` (not `providerName`).
- Update routing tests/helpers that currently set health state under provider name to set it under the stored provider ID, so tests match production behavior.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Duplicate dispatch on redelivery 🐞 Bug ☼ Reliability

Description

forwardWithRetry adds the resourceId to dispatchedSet but does not short-circuit when it already
exists, so JetStream redeliveries can call CreateResource/DeleteResource multiple times and
duplicate side effects.

Code

internal/routing/router.go[R211-214]

+	newlyAdded := r.dispatchedSet.AddIfAbsent(payload.ResourceID)
+
+	// TOCTOU mitigation: re-check denyList after claiming the dispatchedSet
+	// slot. A concurrent HandleCancel may have added to denyList between

Evidence

The router records dispatched-set membership but continues forwarding regardless of whether the
resourceId was already present. Separately, the JetStream handlers ignore Ack errors, making it
possible for a message to be redelivered after successful processing.

internal/routing/router.go[206-223]
internal/messaging/handlers.go[12-26]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`forwardWithRetry` uses `dispatchedSet.AddIfAbsent(resourceId)` but ignores the `false` result and proceeds to forward anyway. If the same message is delivered again (redelivery/duplicate delivery), the router can re-run the SP operation.

### Issue Context
- JetStream consumers can redeliver messages (including when `Ack()` fails).
- `messaging.Client` currently ignores `Ack()` errors, making successful-processing-without-ack a plausible path.

### Fix Focus Areas
- internal/routing/router.go[206-223]
- internal/messaging/handlers.go[12-26]

### What to change
- In `forwardWithRetry`, if `newlyAdded == false`, treat it as a duplicate delivery for an already-dispatched `resourceId` and return without calling the SP forwarder (optionally republish the corresponding ack/error CE if required by your delivery semantics).
- Consider handling `msg.Ack()` errors in `messaging.Client` (at least log them, and/or convert them into a failure path that triggers a retry) to reduce duplicate deliveries after side effects have occurred.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

[gabriel-farache]

/agentic_review

[qodo-code-review]

Code review by qodo was updated up to the latest commit 5235ea8

[gabriel-farache]

#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 PR can still be used as history buildup