feat(operator): reconcile NATS event sinks

[flemzord]

Summary

• add a validated Cluster.spec.sinks.nats API for Ledger v3 NATS JetStream sinks
• reconcile sink configuration at runtime through Ledger's Raft-backed ledgerctl events API, without restarting the StatefulSet
• track operator-owned sink names in status so externally managed sinks are preserved and name conflicts fail visibly
• include the nats build tag in direct/PR Ledger images and the Operator E2E image
• add documentation, a sample manifest, unit coverage, and a permanent Kind/Chainsaw scenario

Declarative behavior

• absent spec.sinks: runtime sinks remain unmanaged
• present spec.sinks (including {}): the Operator maintains the declared set and removes only sinks it previously created
• SinksSynced=True: the Raft-replicated sink configuration matches the CR
• NATS stream provisioning and delivery monitoring remain external; the stream must capture <topic>.>
• this first CRD surface covers NATS only, and rejects credentials embedded in the URL because CR specs are not secret storage

Validation

• nix develop --command bash scripts/agent-check — PASS
• nix develop --command bash -c 'unset GOROOT; cd misc/operator && go test ./...' — PASS
• nix develop --command bash -c 'unset GOROOT; go test -tags nats ./internal/application/events ./cmd/ledgerctl/events' — PASS
• chainsaw test --config e2e/chainsaw-test.yaml --test-dir e2e/tests/event-sinks-nats --kube-context kind-ledger-e2e — PASS

Kind + NATS evidence

The permanent Chainsaw scenario built the local Ledger image with BUILD_TAGS=s3,nats, installed this Operator into Kind, provisioned NATS JetStream, and configured the sink only by updating the Cluster CR.

• CR status converged to appliedSinks: [primary] and SinksSynced=True
• sink-only update did not roll Ledger: pod UID remained 160f13d7-e04f-4260-ada1-647524ac9465 and spec hash remained 3ce71741975bbad7f72580b58b22399ff8c1331f4aa24a52ff5be3ccd5c99f13
• NATS received JSON payloads for CREATED_LEDGER and COMMITTED_TRANSACTION, both identifying operator-sink-e2e
• ledgerctl events list --json reported sink primary, cursor 3, and error: null
• JetStream LEDGER_EVENTS reported messages: 2 across two subjects
• test cleanup completed successfully

Images used for the proof:

• nats:2.12.14-alpine@sha256:7cef1bd3fed6034e95cf6e6bc9c28c5afa6dc58e9fb778dd7924a1ac62569f2d
• natsio/nats-box:0.19.2@sha256:8031d190c7ee24081f3f27cc939fb647a1eeb29ebb5c60fef9b5b6c7a846d6a2

Notes for reviewers

Updating an operator-owned sink is intentionally a two-pass remove/recreate because Ledger currently rejects duplicate sink names. Ledger retains the per-name cursor, so committed events remain eligible for at-least-once delivery while the configuration converges. SinksSynced proves configuration convergence; the E2E scenario separately proves delivery and cursor advancement.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 76.78%. Comparing base (4bb374a) to head (704e274).
⚠️ Report is 3 commits behind head on release/v3.0.

Additional details and impacted files

@@               Coverage Diff                @@
##           release/v3.0    #1844      +/-   ##
================================================
- Coverage         76.79%   76.78%   -0.02%     
================================================
  Files               472      472              
  Lines             50356    50362       +6     
================================================
- Hits              38673    38672       -1     
- Misses             8286     8288       +2     
- Partials           3397     3402       +5     

Flag	Coverage Δ
e2e	76.78% <ø> (-0.02%)	⬇️
scenario	76.78% <ø> (-0.02%)	⬇️
unit	76.78% <ø> (-0.02%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[NumaryBot]

🛑 Changes requested — automated review

A failed sink creation can leave a durable ownership reservation that later authorizes deletion of an externally created sink.

[NumaryBot]

NumaryBot posted 1 new inline finding.

Summary: #1844 (comment)

[shipfox-ai]

Arbiter review — PR #1844: feat(operator): reconcile NATS event sinks

Summary

The declarative NATS-sink reconciler is well structured: the ownership model (nil = unmanaged, {} = managed, status.appliedSinks scoping removals) is coherent, default-value normalization does not cause drop/recreate churn, the spec-hash exclusion correctly avoids a StatefulSet roll, shell-out arguments are single-quoted, and the pure diff/reconcile logic has meaningful unit coverage. However, the reconciler couples an out-of-band runtime mutation (Raft add-sink/remove-sink) to a separately persisted ownership record without an atomic or self-healing protocol, and it deliberately never re-adopts an equal existing sink. That combination produces two paths to permanent, human-intervention-only bad state, plus a steady-state gap where declared configuration is not restored after drift. Recommendation: request changes.

Confirmed findings

Major — Ownership can be permanently lost after a successful add-sink

misc/operator/internal/controller/cluster_sinks_reconcile.go:159-164, persisted via cluster_controller.go:88-91 and :425
Ownership is appended to created only after add-sink returns success (:163), and the in-memory status.appliedSinks update is persisted by a deferred updateStatus whose error is only logged (cluster_controller.go:88-91). If the Raft add-sink commits but the exec/RPC response is lost (reconcileEventSinks returns before appending), or if add-sink succeeds but the Kubernetes status write transiently fails, the runtime sink exists with no persisted ownership record. On the next pass diffEventSinks treats an identical existing sink as external and does not adopt it (cluster_sinks.go:211-212, codified by the test "matching external sink is preserved and not adopted"). The operator then permanently treats its own sink as external: removing it from the CR leaves it running, and changing it produces an external-name conflict. A recoverable protocol (adopt an equal, still-declared sink into appliedSinks, and/or verify the postcondition after an ambiguous add-sink) is needed.

Major — Substring "not found" matching in the remove path can relinquish ownership without removing the sink

misc/operator/internal/controller/cluster_sinks_reconcile.go:169-174 (isLedgerNotFound at ledger_crd_controller.go:594)
isLedgerNotFound is an unstructured strings.Contains(..., "not found") check. In the remove loop, any error containing "not found" is suppressed and dropped is appended unconditionally (:174), removing the name from appliedSinks. If pod 0 disappears between the successful events list and the remove-sink call, podExec returns an error such as pods "…-0" not found, which matches — so ownership is relinquished even though the Raft sink was never removed. For a de-declared sink this leaves an unowned sink running while the next pass reports convergence; for an update it becomes a permanent external-name conflict. This also directly contradicts the operator's own documented design principle: the README (scale-down section) states the operator "does not match human-readable error substrings" and instead re-queries state. Verify the sink postcondition (re-list) or propagate structured not-found information instead of matching transport/command text.

Major — Steady state does not requeue, so out-of-band runtime drift is not restored and SinksSynced=True goes stale

misc/operator/internal/controller/cluster_sinks_reconcile.go:89-97
On convergence handleSinkReconcile returns baseResult, which is the zero ctrl.Result{} in the steady state (reconcile_statefulset.go returns ctrl.Result{} when converged), so no RequeueAfter is scheduled. The controller only watches Kubernetes resources (SetupWithManager); a direct ledgerctl events remove-sink or any other Raft sink mutation emits no watched event. An operator-owned sink can therefore be deleted or altered while SinksSynced=True remains stale and events silently stop, with restoration deferred until an unrelated Cluster reconcile or the default informer resync (~10h). The InProgress path already requeues every 5s, so a steady-state requeue (or a runtime-drift event source) plus a regression test that mutates a converged sink out of band and observes recovery is the natural fix.

Minor — A single external-name conflict aborts the entire reconcile, blocking unrelated owned changes

misc/operator/internal/controller/cluster_sinks_reconcile.go:148-149
reconcileEventSinks returns immediately when diff.conflict is non-empty, before executing any toCreate or toDrop. One conflicting, externally-owned sink thus also blocks unrelated, legitimate creations and removals of operator-owned sinks (stale sinks keep delivering events). Failing closed on the conflict is defensible, but blocking unrelated owned work is an avoidable side effect (e.g. apply non-conflicting drops/creates first, then surface the conflict).

Findings considered and dropped

• Conflict surfaced with Reason: "Error" / 5s error-log repetition — observability nitpick; the README does not promise a distinct Conflict reason and no events.md exists, so the premise is only partly supported and there is no correctness impact.
• Unit-test gap on handleSinkReconcile / earlierRequeue — the pure functions are well covered; the missing coverage is low-risk glue, not risky test coverage warranting a change request.
• Default-value normalization churn, shell injection, cursor survival, CRD enum parity (Reviewer A non-issues) — independently confirmed as non-issues.

Reviewed by Claude (claude-opus-4.8) and Codex (gpt-5.6-sol) via Shipfox; verified and synthesized by Claude.

[NumaryBot]

NumaryBot posted 1 new inline finding.

Summary: #1844 (comment)

[NumaryBot]

NumaryBot posted 1 new inline finding.

Summary: #1844 (comment)

[Azorlogh]

Reviewed the sinks logic closely and it holds up. I verified the parts most likely to be silently wrong:

• The events list --json parser matches protojson exactly — I probed protoJSONOpts and confirmed camelCase keys, batchDelayMs as a quoted string (hence protoJSONInt64), eventTypes as enum names, and unset oneof members omitted so listedSinkKind works.
• CRD batchSize max 100000 == domain.MaxSinkBatchSize, so no CRD-valid value can be rejected by the FSM.
• Reserve-ownership-before-create plus verify-after-failed-remove is idempotency-correct, and the deliberate asymmetry between the two is covered by tests.
• ledgerctlCommand single-quotes every arg, so CRD strings can't reach /bin/sh.

One unrelated commit to split out and three smaller points inline.

One more, pre-existing but load-bearing for this PR: cmd/ledgerctl/events/add_sink.go advertises "If a sink with the same name already exists, it is replaced (upsert)" and aliases the command upsert, while processAddEventsSink returns ErrSinkAlreadyExists. That wrong help text is exactly why the two-pass remove/recreate exists here — worth correcting while you're in the area.

[Azorlogh]

This commit (704e274) doesn't belong in a NATS-sinks PR. It re-adds spec.dnsEndpoint, which #1617 deliberately replaced with dnsEndpoints; nothing in the repo consumes it, the PR body never mentions DNS, and the commit message has no body explaining what broke. Please split it out with a stated reason — and if this is only about CRs that already exist in a cluster, updating those CRs beats carrying a compat shim in an unreleased v3.

[Azorlogh]

A name conflict can never self-heal, but it lands here and requeues at sinkRequeueInterval (5s) forever — roughly 17k ledgerctl events list execs per day into pod-0 for one misconfigured Cluster. Worth backing off to sinkDriftCheckInterval when the failure is a conflict rather than a transient exec error.

[Azorlogh]

The same response also carries sinkStatuses[].cursor and .error, which this struct drops. A Ledger image built without the nats tag stores the config fine and then fails at createSink with "unsupported events sink type: nats (not compiled in this build)" — the CR still reports SinksSynced=True and shows nothing. I know the docs scope delivery monitoring out, but the data is already in hand here, so surfacing it is close to free.

[Azorlogh]

The scenario covers create-and-deliver only. Removal and the two-pass update are exercised solely against the fake exec, which is where the ErrSinkAlreadyExists / ErrSinkNotFound assumptions could actually diverge from real Ledger. A final step flipping to sinks: {} and asserting appliedSinks empties plus the sink disappearing from events list would close that.

[gfyrag]

I think this API would be more coherent as a dedicated EventSink CRD rather than embedding the lifecycle under Cluster.spec.sinks.

One Kubernetes resource per sink would provide a natural identity and lifecycle, independent status/conditions, isolated conflicts, simpler GitOps/RBAC, and a cleaner path for future transports and Secret references. It would also remove the subtle absent-vs-empty spec.sinks semantics and the overloaded Cluster.status.appliedSinks bookkeeping.

For example:

apiVersion: ledger.formance.com/v1alpha1
kind: EventSink
metadata:
  name: primary
spec:
  clusterRef:
    name: sink-cluster
  nats:
    url: nats://nats:4222
    topic: ledger.events
  format: json
status:
  conditions: []
  cursor: 42

The dedicated CRD alone would not fully solve the current ownership blocker, however. Ownership should also be represented in Ledger's Raft-replicated SinkConfig with an opaque controller identity derived from the CR UID (or an equivalent durable token), and the reconciler should verify that identity before updating or deleting a runtime sink. A finalizer can then drive cleanup safely. This avoids treating either a Kubernetes status reservation or a matching runtime name as proof that the operator created the sink after an ambiguous/failed add-sink call.

Since Ledger v3 is still unreleased and has no compatibility burden between development revisions, I recommend establishing this API/lifecycle now rather than carrying the name-only ownership protocol forward.