Maxence MaireauxandGitHub
6504cd9692
* fix: resolve P0/P1 reliability and security issues - P0: Process messages synchronously to prevent data loss on crash The Watermill handler now blocks until webhook delivery completes. Messages are only acknowledged after processing, not before. Errors on invalid payloads are returned (nack) instead of swallowed. Per-config failures use continue instead of return, so one bad endpoint no longer blocks delivery to other matching configs. - P1: Replace panic with error on unknown filter key FindManyConfigs now returns an error instead of panicking when called with an unrecognized filter key. A single bad API request no longer crashes the process. - P1: Remove environment and secret logging Removed os.Environ() dumps from server and worker startup. Replaced config %+v logging with safe ID/endpoint-only logging. Removed event-payload from OpenTelemetry span attributes. - Docs: Add architecture, message-processing, and security docs. Fix CTE typo in retry-mechanism.md. No breaking changes to public API, Store interface, or CLI flags. * chore: add CODEOWNERS assigning @formancehq/backend as default reviewers * fix: remove dead code flagged by ineffassign linter * fix: address CodeRabbit review feedback - Return error (nack) when InsertOneAttempt fails for non-success attempts, preventing silent loss of retry records - Rewrite TestProcessMessages_SynchronousProcessing to be deterministic: HTTP handler blocks on a channel, test proves handler is blocked until the HTTP response completes - Add TestProcessMessages_NackOnInsertFailureForRetry for the new nack path - Expand secret logging test patterns to cover %v and more variable names - Rename TestConfigStringerDoesNotLeakSecret to TestDocument_PercentPlusVLeaksSecrets - Fix retry-mechanism.md claim query priority documentation: DISTINCT ON does per-webhook ordering, not global priority * fix: propagate errors to span, avoid err shadowing in defer - Rename loop-scoped err to attemptErr/insertErr so the defer closure captures the correct error for span.RecordError() - Add explicit span.RecordError() on MakeAttempt and InsertOneAttempt failures that previously went unrecorded in traces - Assign err before return on nack path so the defer sees it - Update Go version in architecture doc to 1.25+ (matches go.mod) Addresses review comments from paul-nicolas on PR #142. * fix: handle transport errors in MakeAttempt as retryable attempts - MakeAttempt now returns a retryable Attempt (instead of a bare error) when httpClient.Do fails due to transport/timeout, preventing event loss since the attempt record is persisted for the retry loop - Replace base64-looking test secret with plain dummy to avoid triggering secret scanners in CI Addresses CodeRabbit review comments on PR #142.
2.6 KiB
Message Processing
Overview
The worker consumes events from a message broker (Kafka or NATS) via Watermill and delivers webhooks to matching configured endpoints.
Processing Model
Messages are processed synchronously within the Watermill handler. The handler only returns after all webhook deliveries for that message are complete. This ensures:
- No data loss — The message is acknowledged by the broker only after processing finishes. If the worker crashes mid-processing, the broker redelivers the message.
- Backpressure — If delivery is slow, the consumer naturally slows down. No unbounded queue builds up in memory.
- Error propagation — If the message cannot be parsed, the handler returns an error, allowing Watermill to nack/retry/dead-letter as configured.
Why not a worker pool?
A previous implementation dispatched messages to a pond worker pool and returned nil immediately. This created a data loss window: the message was acknowledged before processing finished. If the worker crashed between ack and completion, messages were silently lost.
The current design relies on Watermill's built-in concurrency model. Watermill's router supports concurrent message handling natively — configure it via RouterConfig.Handler.MaxConcurrentMessages if higher throughput is needed.
Flow
Broker (Kafka/NATS)
│
▼
Watermill Router
│
▼
processMessages handler (synchronous)
│
├─ Unmarshal event
├─ Normalize event type (lowercase, app prefix)
├─ Query matching active configs
├─ For each config:
│ ├─ MakeAttempt (HTTP POST with signature)
│ ├─ Insert attempt record
│ └─ Log result
└─ Return nil (ack) or error (nack)
Event Format
Events are expected as publish.EventMessage JSON objects with at minimum:
type— The event type (matched against configevent_types)app— Optional app prefix (combined asapp.type)
The full event payload is forwarded as the webhook body.
Error Handling
| Error | Behavior |
|---|---|
| Unmarshal failure | Return error → broker nack/retry |
| No matching configs | Return nil → message acknowledged (no work needed) |
| HTTP call failure on one config | Log error, continue to next config |
| Database insert failure | Log error, continue to next config |
Individual config failures do not block other configs from receiving the same event. Only message-level errors (bad payload, config query failure) cause a nack.