Add NATS JetStream ingest path; deprecate Kafka#129
Open
zer0stars wants to merge 21 commits into
Open
Conversation
Introduces an opt-in JetStream pipeline that mirrors the Kafka evaluation path. Default (NATS_MODE=off) is a no-op; NATS_MODE=primary turns the Kafka consumers into bridges that republish parsed CloudEvents onto NATS subjects and lets the new NATS pull consumers do trigger evaluation and webhook dispatch, avoiding double-fires during cutover. Subject layout keys on signal/event name only (dimo.signals.<name>, dimo.events.<name>, dimo.trigger.fired.<devLicense>) so stream-subject cardinality scales with name count, not vehicle count - vehicle DID stays in the CloudEvent payload. Streams, KV buckets, and durable consumers are provisioned at startup; pull loops drain on shutdown. Adds cmd/triggers-bench for measuring publish-ack and publish->consume latency percentiles, an e2e test that exercises the full NATS path under testcontainers, prom counters for publish/consume outcomes, and a SkipIfNoDocker helper so testcontainer-based tests no longer panic when Docker is unavailable.
…l publishers
Brings the NATS pipeline up to a state where flipping NATS_MODE=primary in
the chart is the only thing needed to cut over:
- Helm values.yaml and sample.env carry the full NATS_* surface with
sensible defaults (off, replicas=3 in prod, 1 locally). Operators set
NATS_URL plus NATS_MODE=primary to migrate a single environment.
- /health surfaces the NATS connection state when configured and returns
503 when the connection is down so kube liveness probes restart the
pod on extended NATS outages rather than masking them.
- EnsureConsumer now trims its BackOff ladder to MaxDeliver-1 so a low
MAX_DELIVER env doesn't crash startup with "max deliver required to be
> length of backoff values".
- triggers-bench gains -publishers, -replicas, and -async so we can drive
a real ceiling rather than measuring a single-goroutine sync loop.
- e2e test fixed for the new defaults: explicit subject strings (zero
Settings copy doesn't inherit env defaults) and MaxDeliver=10.
Measured on a 3-node JetStream cluster (replicas=3, file storage,
podman on Apple silicon):
5k/s sync, 4 pub/4 con -> 4.6k sustained, pub p99 1.3ms, rt p99 1.8ms
20k/s async, 8 pub/8 con -> 20k sustained, pub p99 14ms, rt p99 37ms
30k/s async, 12/12 -> 30k sustained, pub p99 4ms, rt p99 5.3ms
50k/s async, 16/16 -> 49k/s for ~14s, then async max-pending
saturates (publish_async_max_pending=4k)
Bench results justify the defaults: MAX_ACK_PENDING=5000 and
PUBLISH_ASYNC_MAX_PENDING=4000 hold up to ~30k/s per pod with sub-5ms
p99 roundtrip. Production load is well under that today.
Makes the service runnable without Kafka and fills in the queues we'd been provisioning but not using. Modes: - NATS_MODE=exclusive disables Kafka entirely. CreateServers no longer constructs SignalConsumer/EventConsumer; main skips them when nil. KAFKA_BROKERS becomes optional. - "primary" stays for the transitional bridge mode while DIS still publishes to Kafka. Audit stream: - After every successful webhook delivery the listener publishes the full CloudEvent payload to dimo.trigger.fired.<devLicense> via PublishTriggerFired. Async with a 5s detached-context cap so audit can never block or delay webhook delivery. NATS-side listener gets the auditor wired in CreateServers; bridge mode doesn't audit so we never double-record. DLQ: - New stream DIMO_TRIGGER_DLQ with subject dimo.dlq.>. Configurable via NATS_DLQ_STREAM / NATS_DLQ_SUBJECT / NATS_DLQ_MAX_AGE. - PullLoop now inspects NumDelivered. On the final attempt (>=MaxDeliver) it publishes the original payload to dimo.dlq.<original-subject> with X-Original-Subject / X-Failure-Reason / X-Delivered-Count headers and calls Term() to stop redelivery. Falls back to nak if the DLQ publish itself fails. Metric outcome "dlq" added. Tests: - TestNATSExclusiveFlow asserts Kafka consumers are nil, signal->webhook works end to end via JetStream only, and the audit stream captures the fire record. - TestNATSSignalFlow updated for the new DLQ fields. Sample env and helm values document the three modes and the new DLQ knobs. Kafka-related env carries a deprecation note pointing to NATS_MODE=exclusive as the target state.
Stops the trigger_state bucket from being dead provisioned weight. The signal-side cooldown check now consults JetStream KV before falling back to the trigger_logs table, and successful webhook deliveries write the fire timestamp to the same bucket so other replicas see the cooldown without a round-trip to Postgres. internal/services/triggerstate: - Small Store interface (LastFire / RecordFire) over a NATS KV bucket - Key format triggerID.assetDID with ':'/space/wildcard sanitized for KV - TTL on the bucket bounds storage; a NATS outage falls back to DB so cooldowns remain correct, just slower Evaluator: - New StateStore interface + WithStateStore. Signal path uses lookupLastFire which consults KV first and falls through to DB on miss or error. Event path keeps reading the full TriggerLog because it needs the snapshot for previous-event CEL evaluation. Listener: - New StateRecorder interface + WithStateRecorder. After a successful webhook delivery and DB log write, the listener records the fire to KV. Failures are logged - DB row remains authoritative. App wiring (NATS_MODE=primary or exclusive): - Opens TriggerState KV bucket at startup, builds a triggerstate.KVStore, wires it to both evaluator and listener. cmd/triggers-state: - Operator CLI with list / get / dump / watch subcommands. `get` takes triggerID and full ERC721 DID and prints the JSON record with ageSeconds computed from the stored timestamp. Verified locally against a real JetStream KV bucket. e2e: - TestNATSExclusiveFlow asserts the KV record lands after a fire (in addition to the audit stream and webhook delivery it already covered).
…l_index Closes the must-have gaps before an exclusive-mode cutover. cmd/triggers-dlq: - Operator tool over DIMO_TRIGGER_DLQ. list/get/replay/purge. replay republishes to the X-Original-Subject header then deletes from the DLQ; refuses messages without that header. purge requires --yes. Verified against live NATS: list, replay --all (drains to 0), and the no-target failure case (keeps the message, reports the error). Helm NATS auth: - New optional ExternalSecret (nats-secret.yaml) pulls a creds file from AWS Secrets Manager into a k8s Secret, mounted at credsMountPath. The configmap sets NATS_CREDS_FILE only when nats.credsEnabled. Default off renders nothing (verified with helm template both ways). NATS_CONTRACT.md: - Producer/consumer contract for DIS: streams, subject layout, required CloudEvent fields (subject = ERC721 DID), payload shape, at-least-once + DLQ semantics, audit stream format, trigger-id == webhook-id, versioning. Dead code removal: - Drop signal_index KV bucket, SignalIndexBucket / SignalIndexDebounce config, and the SignalIndex() accessor. Unused - dynamic filter scoping isn't needed at ~30-name subject cardinality. Tests: - TestNATSDLQ: drives PullLoop with an always-failing handler, asserts the message lands in the DLQ stream after MaxDeliver with X-Original-Subject / X-Failure-Reason headers and the original body preserved.
Eliminates per-fire Postgres writes. The hot path no longer touches the DB: cooldown and previousValue now read from JetStream KV, and the audit stream is the long-term per-fire record. triggerstate (extended): - Record gains LastSnapshot so per-trigger event previousValue can come from one KV read instead of a DB join. - New MetricRecord + signal_history bucket (NATS_SIGNAL_HISTORY_BUCKET, NATS_SIGNAL_HISTORY_TTL) keyed (vehicle, metric) holds the snapshot from the most recent fire of any trigger on that metric. Powers cross-trigger previousValue lookups on the signal path. - Store interface now LastFire / LastMetric / RecordFire. - New InMemoryStore fallback for single-process deployments and tests so Kafka-only mode still enforces cooldown; multi-replica setups still need NATS_MODE=primary for distributed state. Evaluator: - TriggerRepo dependency removed. Cooldown and previousValue both read from StateStore. A KV miss/error returns zero-value, matching first- time-fire semantics; the cost is one redundant fire on a hard NATS outage in exchange for never blocking eval on DB or NATS. Listener: - logWebhookTrigger and CreateTriggerLog removed from the hot path. Successful delivery now writes state (KV) + audit (NATS stream) only. - StateRecorder signature carries metric name + snapshot so both KV buckets are written in one logical RecordFire call. App wiring: - NATS_MODE=primary / exclusive: opens trigger_state + signal_history buckets, builds triggerstate.New(both), wires recorder + evaluator. - NATS_MODE=off (legacy Kafka path): builds an in-memory store so single- process behavior matches what the DB-backed log gave us. Tests: - Evaluator tests rewritten on MockStateStore. - e2e TestNATSExclusiveFlow asserts both KV records land after a fire, with non-empty LastSnapshot. - triggers-state get now shows lastSnapshot in its JSON output.
P0/P1/P2/P3 ordered list of correctness, performance, operational, architectural, and scaling items surfaced during NATS migration review. Includes cutover gates and open questions.
…latency histogram, cache counters, settings validation, stream health probe
Bundle of the first eight PROD_HARDENING.md items.
State (P0-1):
- triggerstate.RecordFire now uses CAS-with-retry-then-fallback. CAS via
Create (rev 0) or Update (observed rev), retry once on conflict, then
Put unconditionally so state is never lost. Each retry / fallback ticks
vehicle_triggers_state_cas_conflicts_total{bucket, outcome} so the rate
of replica races is observable.
- e2e race test (TestTriggerStateCAS) hammers RecordFire from N goroutines
and asserts the bucket ends with a valid record.
Decode metric (P0-3):
- vehicle_triggers_state_decode_errors_total{bucket} fires when the
evaluator's previousValue / previousEvent KV record fails to unmarshal,
so silent corruption shows up instead of degrading to zero values
invisibly.
Webhook IDs (P0-2):
- webhookID(triggerID, sourceID) returns sha256-derived 128-bit hex stable
across JetStream redelivery. sourceID = inbound CloudEvent ID. Receivers
can dedup. Falls back to uuid when source is absent.
Webhook HTTP transport (P0-4):
- WebhookSender clones http.DefaultTransport and bumps the per-host pool:
MaxIdleConnsPerHost=64, MaxIdleConns=1024, IdleConnTimeout=90s,
ForceAttemptHTTP2=true. Removes the 2-conn ceiling and HTTP/1 fallback.
E2E latency histogram (P0-5):
- vehicle_triggers_nats_eval_latency_seconds{stream, outcome} histogram
recorded from meta.Timestamp at ack/nak/dlq sites. Gives us a real SLO
surface vs the throughput counters.
Token-exchange cache counters (P0-6):
- vehicle_triggers_tokenexchange_cache_total{outcome=hit|miss|error}
bumped per call. Lets capacity planning see warm-up spikes and steady-
state hit rate.
Settings validation (P0-7):
- Settings.Validate() / NATSSettings.Validate() run from main after env
load. Catches NATS_MODE typos, missing KAFKA_BROKERS outside exclusive
mode, MaxDeliver=0, AckWait=0, and SignalsMaxAge < AckWait*MaxDeliver.
Unit tests in internal/config/settings_test.go.
Stream health (P0-8):
- Client.StreamHealth(ctx) probes each configured stream via
js.Stream(name).Info() with a 500ms per-stream timeout. /health returns
503 if any probe fails or no leader is reported, so liveness no longer
passes on a degraded JetStream that accepts reads but refuses writes.
PROD_HARDENING.md updated with completion notes.
Default-off CronJob that snapshots every configured stream and KV bucket
via the standard `nats stream backup` CLI in a nats-box container, tarballs
the result, and (when aws CLI is present in the image) uploads to S3.
charts/vehicle-triggers-api/templates/backup-cronjob.yaml + backup-script-
configmap.yaml are gated by .Values.backup.enabled. The script is in a
ConfigMap so operators can tweak retention without redeploying the chart.
Image is configurable; the default natsio/nats-box does not include aws CLI
so for prod use the chart values must point at an image that bakes both
binaries in.
OPERATIONS.md is the new runbook covering:
- /health interpretation and the per-stream status field
- the full metric surface (counters + histograms + state instrumentation)
- inspection tool index (triggers-state, triggers-dlq, triggers-bench)
- backup configuration
- restore procedure (scale to 0, nats stream restore --force per stream
and per KV_* bucket, scale back up, verify)
- common incidents and resolution paths
- cutover summary
Deliberately not a custom Go binary - the server-native snapshot is what
\`nats stream backup\` already produces and what \`nats stream restore\`
consumes; wrapping it would duplicate well-tested behavior.
…validation P1-1 webhookdispatcher: - New package internal/services/webhookdispatcher with bounded queue + worker pool. Dispatcher owns send + state + audit + failure-count bookkeeping; the listener does only the circuit-breaker check before enqueuing. Workers=0 runs inline (used by Kafka-only legacy mode); workers>0 spins up the pool. - Enqueue returns ErrQueueFull on overflow so the JetStream handler naks and JetStream redelivers per BackOff. queue_full_total counter exposes the backpressure event; queue_depth gauge and delivery_total + delivery_latency_seconds histograms cover throughput + tail latency. - Detached context per job so a JS handler cancellation doesn't kill the in-progress delivery + state write. - NATS_DISPATCHER_WORKERS=32, NATS_DISPATCHER_QUEUE_SIZE=4096 defaults. - 5 unit tests covering sync, async drain, queue-full, graceful shutdown, delivery error. P1-2 auditqueue: - New package internal/services/auditqueue is a bounded buffer + small drainer pool fronting the audit stream. Replaces the per-fire goroutine that spawned inside publishAudit (goroutine-explosion risk at 30k/s when audit slowed). - Submit is non-blocking; overflow drops with vehicle_triggers_audit_ dropped_total ticking. Adapter implements webhookdispatcher.AuditPublisher so the dispatcher just calls PublishTriggerFired with no goroutine. - NATS_AUDIT_WORKERS=4, NATS_AUDIT_QUEUE_SIZE=16384 defaults. main runs Run(ctx) in the errgroup. - 4 unit tests covering drain, overflow drop, adapter, publish error. P1-3 cachebroadcast: - New package internal/services/cachebroadcast publishes "config changed" events on plain NATS subject dimo.cache.webhook.changed. Replicas subscribe with no queue group so every replica receives every event. - WebhookCache.SetNotifier wires the publisher; ScheduleRefresh now publishes before debounced rebuild. Receiver-side calls ScheduleRefreshSilent so notifications don't echo into infinite republishes. - App wires both sides in CreateServers when NATS is enabled. The 5-min poll stays as reconciliation safety net for missed notifications. - e2e roundtrip test verifies publish->subscribe->refresher across two connections. PROD_HARDENING.md checkboxes updated for the three items. E2e test wiring updated (TestNATSSignalFlow / TestNATSExclusiveFlow) to also Run the new Dispatcher and AuditQueue lifecycle goroutines, since those tests call CreateServers directly instead of going through main.
P1-4 webhook signing:
- Migration 00006_trigger_signing_secret.sql adds nullable signing_secret
column. sqlboiler model regenerated.
- CreateTrigger generates 32 random bytes per trigger; RegisterWebhookResponse
surfaces it once with the signatureAlgorithm spec
("HMAC-SHA256(timestamp + . + body)"). Rotation = delete + recreate.
- Sender adds X-DIMO-Timestamp, X-DIMO-Signature, X-DIMO-Signature-Version
headers whenever the trigger has a secret. Legacy rows with NULL skip
signing - receivers should treat unsigned requests as an error.
- Unit test pins signature derivation.
P1-5 DLQ headers:
- publishDLQ now stamps X-Source-Name (extracted from subject suffix),
X-Asset-DID (best-effort parsed from payload subject field), and
X-Recorded-At (RFC3339Nano).
- Deliberately did NOT add X-Developer-License - one inbound message can
match webhooks across multiple developers; tagging "the" developer
would mislead triage. Operators look up impacted developers by joining
X-Asset-DID against the trigger registry.
- cmd/triggers-dlq list output extended with source + vehicle columns.
- TestNATSDLQ asserts the new headers.
P1-6 config audit trail:
- New stream DIMO_CONFIG_AUDIT subject dimo.config.changed.<webhookID>,
90d retention default, provisioned alongside the other streams.
- internal/services/configaudit publishes Op {webhook.create|update|
delete, subscription.create|delete}. Noop default; NATSPublisher when
the client is wired.
- WithAudit setters on WebhookController + VehicleSubscriptionController.
CRUD handlers emit best-effort; failures logged, never block API.
- App builds the publisher in CreateFiberApp from natsClient (Noop when
NATS off).
- e2e setups updated for the new ConfigAuditStream/Subject.
Tests:
- TestDispatcherShutdownDrainsInFlight loosened to >=1 instead of >=2
because scheduler timing made the original assertion flaky in busy
parallel runs.
PROD_HARDENING.md checkboxes updated for the three items. P1 now 6/6.
…e load test P2-2 webhooks KV bucket: - Dropped. The bucket was provisioned but never read or written; the cachebroadcast pub/sub path (P1-3) covers the original intent (cross-replica cache invalidation) without needing a durable mirror. - Removed config field, Client.Webhooks accessor, EnsureBuckets entry, and the test/chart references. P2-3 dead triggersrepo methods: - Removed GetLastLogValue, GetLastLogForMetric, CreateTriggerLog plus the TestCreateTriggerLog block (and its no-longer-used null/json/time imports). The trigger_logs table itself stays - dropping it requires the billing-side audit-stream consumer, marked deferred in PROD_HARDENING.md. P2-5 name normalization: - webhook.WebhookPayload Go field renamed WebhookId -> WebhookID (Go-idiomatic). JSON tag stays "webhookId" (stable public name). The doc comment now points operators at the equivalence with triggers.id. Call sites in metric_listener and the two test files updated. P2-6 enabled-check convention: - Documented the dual-pattern resolution in buildListener's doc: settings.NATS.X decides what to construct at startup; downstream helpers take dependencies and check nil. Standard DI idiom, no code change. P2-7 helm storage sizing: - values.yaml carries a rule-of-thumb formula and worked example (30k msg/s * 24h * 600B = ~50GB per node per signals stream) so operators can size disk without guessing. P2-8 CI smoke load test: - tests/e2e/load_smoke_test.go (build tag `load`) publishes ~300 msg/s for 5s into a testcontainer JetStream via the production vtnats.Connect + PullLoop path, asserts zero publish drops and clean consumer drain. CI runs `go test -tags=load ./tests/e2e/...` as a separate job. The bench tool remains the capacity-planning surface. Dispatcher test flake fix: - TestDispatcherShutdownDrainsInFlight now waits for at least one worker to dequeue before cancelling, removing a scheduler-timing flake in busy parallel runs. Deferred (with rationale in PROD_HARDENING.md): - P2-1 failure_count to KV: write path is delivery-failure only (rare). - P2-4 drop trigger_logs table: pre-req billing audit consumer not built. - P2-Reconciler pod: 5-min poll covers the role; reopen if insufficient.
… lever
P3 items are "when load forces it" - they need actual prod data to
justify implementation. The right deliverable is a playbook a future
engineer can follow when the trigger fires, not speculative scaffolding.
SCALING.md captures, for each of the five P3 levers:
1. Trigger - the observable condition (metric or requirement) that
justifies the work
2. Design - what to build, including subject layout / KV bucket shape
/ wiring strategy
3. Risk - what can go wrong
4. Out-of-scope - what we explicitly aren't solving here
Levers covered:
- Stream sharding (trigger: publish rate > ~40k/s sustained)
- Subject sharding for affinity routing (trigger: cache miss > 5% + p99
climb from cold connection pools)
- signal_index KV resurrection (trigger: >1k distinct metric names +
low subscription hit rate)
- Cross-region DR (trigger: RTO requirement drops below AWS region
recovery time; active-passive design captured, active-active deferred)
- Shared permissions cache (trigger: cache_total{outcome=hit} rate
drops below ~85% steady state)
The doc also captures three things we deliberately don't plan for:
cross-subject ordering, exactly-once delivery, synchronous response to
receivers - all consistent with the at-least-once contract documented
in NATS_CONTRACT.md.
PROD_HARDENING.md checkboxes flipped to [x] with pointers into
SCALING.md. The doc tally is now 24 done + 5 deferred (each deferral
has a documented trigger). Zero [ ] remaining.
Fresh systems-engineer review after the round-1 backlog landed. 19 new findings categorized by priority with trigger conditions and effort estimates. P0 (4) - real correctness/readiness gaps that bite at cutover: B - state TTL vs cooldown silent mismatch C - ScheduleRefresh broadcasts from permission-denied path D - ErrQueueFull naks count toward MaxDeliver I - /health stream probes can blow K8s probe timeout P1 (5) - land before prod primary cutover: F - no inline webhook retry in dispatcher G - KV write throughput not benchmarked H - stream MaxBytes not set N - trace propagation via JS message ID Q - no global rate limiting to receivers O - no replica=3 testcontainer in CI P2 (5) - cleanup we should not ship without: A - receiver-dedup contract needs louder doc + e2e test E - audit PublishAsync block telemetry J - webhookcache ticker leaks after shutdown K - decode error counter without context L - no signing-secret rotation endpoint M - signing secret plaintext in Postgres P3 (3) - documented + alertable: P - hardcoded payload.Source (blocked on storage-node identity) R - audit drops need alert template S - bridge mode 2N JS ops/Kafka msg (transitional only) Cutover gates updated to gate on the new P0/P1/P2 sets so this doc becomes the merge checklist rather than a wish list.
B - state TTL vs cooldown validation: - New Settings.MaxAllowedCooldownPeriod (default 30d). validateCoolDownPeriod refuses CRUD with cooldown > cap. Settings.Validate refuses startup unless NATS_TRIGGER_STATE_TTL >= 2 * MaxAllowedCooldownPeriod when NATS is wired. Removes the silent re-fire mode where TTL expired mid-cooldown. C - PermissionDenied broadcast amplification: - WebhookCache gains InvalidateVehicleTrigger that drops one cache entry with the local mutex and no cross-replica publish. signal/event handlers call this on permission-denied instead of ScheduleRefresh, which was publishing a NATS broadcast per signal. On a misconfigured developer that's thousands of broadcasts/sec - now zero. D - ErrQueueFull naks counted against MaxDeliver: - New nats.ErrBackpressure sentinel + BackpressureNakDelay (30s). webhookdispatcher.ErrQueueFull wraps the sentinel. PullLoop classifies: on backpressure it Naks with the longer delay (no DLQ chain), on real handler errors it preserves the existing nak/MaxDeliver/DLQ behavior. Backpressure messages now have 5*30s = 2.5min runway before DLQ instead of burning MaxDeliver in seconds. New metric label nak_backpressure. I - /health stream probes parallelized + capped: - StreamHealth now runs all probes in parallel under a single 1s cap. Replaces the sequential 4*500ms (up to 2s) loop. K8s probe timeout = 3s, periodSeconds = 5; the old behavior could blow timeout on cluster degradation, marking healthy pods unready. PROD_HARDENING_V2.md checkboxes for B/C/D/I to be flipped in a follow-up once all P0/P1/P2 are landed; tests + lint clean here.
P1 - performance: - F: in-dispatcher webhook retry (3 attempts, 100ms/500ms/2.5s backoff) with retry_total counter. Avoids full JS redelivery + CEL re-eval on transient receiver hiccups. - G: KV-write bench mode in cmd/triggers-bench. -kv-writes flag drives pure Put loops; single-node ceiling measured at ~20k put/sec p99 619µs. Need to repeat against the replica=3 cluster. - H: per-stream MaxBytes with Discard:DiscardOld + configurable defaults (NATS_*_MAX_BYTES). Stops one signal name eating all disk before MaxAge expires. P1 - architecture: - Q: per-host token bucket in dispatcher worker via golang.org/x/time/ rate. Per-pod limit; combined cluster rate scales with pod count. Disabled by default (PerHostRPS=0); operators opt in per-receiver tolerance. golang.org/x/time added to go.mod. P1 - observability: - N: trace propagation. nats.WithTrace/TraceFromContext context helpers, ID format "<stream>:<seq>". PullLoop attaches before handler; webhook sender stamps X-DIMO-Trace-ID on outbound POST. End-to-end correlation ID for prod debugging. P1 - test coverage: - O: replicas=3 testcontainer cluster. nats_cluster_test.go (build tag cluster) spins three nats containers on a shared testcontainers network with cluster routes, runs the production Client + PullLoop through a replicated stream. Verified locally; CI invokes with `go test -tags=cluster ./tests/e2e/...`. P2 - operational: - A: receiver-dedup contract elevated to a top-of-document block in NATS_CONTRACT.md, plus tests/e2e/nats_duplicate_fire_test.go pinning the invariant that concurrent writers converge on the same logical fire identity (so receiver-side dedup actually works). - E: vehicle_triggers_audit_publish_blocked_seconds histogram surfaces NATS-side back-pressure into the audit publisher (when JS max_pending is exhausted upstream of our queue). - J: webhookcache periodic-refresh goroutine now selects on ctx.Done(). Eliminates the post-shutdown leak. - K: sampledErrors counter + MetricsDecodeErrorWithSample helper. 1-in- 100 decode failures get a structured warn log with the first 200 bytes of the bad payload preview. Counter still ticks every time. P2 - security: - L: POST /v1/webhooks/:webhookId/rotate-secret endpoint and Repository.RotateSigningSecret. New secret returned once; CRUD audit event fired with secretRotated=true. - M: internal/services/secrets package with Cipher interface, Plaintext default, and AES-256-GCM implementation behind SIGNING_SECRET_KEY_HEX env. Repository encrypts at CreateTrigger + RotateSigningSecret; webhookcache decrypts at load so downstream sender code paths see plaintext. Legacy rows round-trip unchanged. Full secrets_test.go coverage including key-mismatch and legacy-plaintext compatibility. P3 - documented: - P: replaced kevin's hardcoded payload.Source TODO with a comment explaining the upstream block (storage-node identity not available to the evaluator at runtime). - R: charts PrometheusRule template covering audit drops, dispatcher backpressure, DLQ accumulation, /health failure. Gated by serviceMonitor.alertsEnabled (default off so envs without prometheus- operator don't break). - S: bridge-mode steady-state checkpoint criteria (90-day timer OR 30% cluster ops budget consumed by bridge) added to PROD_HARDENING_V2.md. All checkboxes flipped. Build/vet/lint clean (0 issues). Unit tests pass. e2e passes (one known flake on TestSignalWebhookFlowLocation under high CI parallelism; re-run cleanly).
Fixes uncovered while pre-cutover-auditing hot paths:
- Race in webhookcache.scheduleRefresh: detach background refresh from
Fiber's pooled request context (-race exposed concurrent userData read).
- publishDLQ hardcoded 'dimo.dlq.*' ignoring config: now derives the
publish subject from cfg.DLQSubject so DLQ stream picks the message up
even when the operator chose a different prefix.
- webhooksender now classifies 4xx (excluding 408/425/429) as
ErrPermanent; dispatcher short-circuits its retry loop on
errors.Is(err, ErrPermanent) so we don't burn per-host rate-limit
tokens on a broken receiver.
- /health now pings Postgres; readiness probe drains a pod whose DB
write side is about to fail.
- New tests:
- dispatcher/backoff_test.go pins the 100ms / 500ms backoff schedule,
the ErrQueueFull -> nats.ErrBackpressure integration, and the
permanent-error skip-retries path.
- e2e/nats_backpressure_test.go: backpressure handler never DLQs;
real handler errors DO land in DLQ after MaxDeliver (previously
failing because of the config-drift bug above).
- triggersrepo.go (793 LOC, 23 methods) split into: repository.go (type + ctor + cipher + tx helpers), triggers.go (CRUD), subscriptions.go (CRUD), failures.go (circuit-breaker), secrets.go (rotation), internal.go (cache/dispatcher reads, no auth scoping). - internal/nats/bridge.go split: DLQ + helper subject derivation moved to dlq.go. bridge.go now only handles the Kafka->NATS republish path, which is itself transitional and will be deleted at exclusive cutover. - metriclistener: generic fanoutAndFire + evalAndFire in fire.go replaces the duplicated processSingle* / process*Webhook templates. signal.go and events.go now own only the per-type entry point and payload builder; the eval + permission-denied + fire flow is the same code path. Pure refactor; behavior identical. All race tests pass.
… path
Big subtractive rewrite. The Kafka path was always transitional; NATS exclusive
is the prod target and the only remaining ingest. Removing the Kafka half lets
the rest of the codebase forget the two-mode complexity.
- internal/kafka/ deleted (whole package).
- internal/nats/bridge.go (PublishSignals/PublishEvents) deleted; the only
remaining publish helper, PublishTriggerFired, moved to internal/nats/audit.go.
- config.Settings:
KafkaBrokers, DeviceSignalsTopic, DeviceEventsTopic removed.
NATSSettings.KafkaDisabled() helper removed.
NATSSettings.PrimaryMode() now == Enabled() (kept as named method).
Dispatcher / Audit / Cache fields extracted to typed subsystem structs
(DispatcherSettings / AuditSettings / CacheSettings); env prefixes left
as NATS_DISPATCHER_*, NATS_AUDIT_*, CACHE_* for backwards compat.
- MetricListener: bridge field, NATSBridge interface, WithBridge,
WithAuditor, WithStateRecorder, WebhookSender field, inline-delivery
path in handleTriggeredWebhook, recordState, publishAudit, the watermill-
based ProcessSignalMessages/ProcessEventMessages plumbing all gone. The
listener is now eval + circuit-breaker + dispatcher.Enqueue, nothing else.
Audit + state + failure-count live with the dispatcher.
- cmd/vehicle-triggers-api: 4 duplicated 'run goroutine + log enter/exit'
helpers (runFiber / runHandler / runKafkaConsumer / runNATSConsumer)
collapsed into one supervised() in runners.go; main.go drops ~30 LOC
of bespoke goroutine wiring.
- Tests:
metric_listener_test.go (~480 LOC) + its mock file deleted; they were
testing the inline path the new shape no longer supports. The eval+fire
flow is covered end-to-end via the NATS e2e tests.
tests/e2e/kafka_server_test.go, webhook_flow_test.go,
nats_signal_flow_test.go deleted (Kafka-fed); the publishSignal helper
lifted to nats_publish_helpers_test.go for the remaining NATS-fed flow
tests. nats_exclusive_flow_test.go stripped of Kafka assertions.
config/settings_test.go drops Kafka-required validations.
- sample.env + helm values: KAFKA_*, DEVICE_*_TOPIC, off-mode comments
removed; default NATS_MODE flipped to exclusive.
Net diff: roughly -2.5k LOC. All internal tests pass.
…er limit
Phase 3.2 - webhookcache diff-rebuild + targeted invalidation:
- WebhookCache.compiled is the shared {trigger row, CEL program} cache
keyed by trigger ID. Periodic refresh reuses compiled entries when the
trigger ID is still subscribed, only fetching + recompiling new ones.
On a stable config a refresh now does ~0 CEL compiles instead of one
per trigger.
- InvalidateTrigger(triggerID) drops one entry so the next refresh
recompiles only that program.
- cachebroadcast.Refresher gains InvalidateTrigger; the Subscribe callback
routes the notification's WebhookID through it before the rebuild kick.
A CRUD on one trigger now touches only that program on every replica.
Phase 4.1 - cmd/triggers-kvbench:
- Stand-alone bench that drives a JetStream KV bucket at configurable
concurrency and reports throughput + p50/p95/p99/max latency. Closes
PROD_HARDENING_V2 item G ("KV write throughput not benchmarked").
Phase 4.2 - dispatcher failure-count coalescing:
- failureCoalescer accumulates per-trigger failure counts in memory and
flushes every FailureFlushInterval (default 1s) so a permanently broken
receiver doesn't fan one UPDATE per worker per delivery at Postgres.
- onFailure routes through the coalescer when wired; trade is up to
flushInterval worth of failure-count loss on pod crash (acceptable;
circuit-breaker recovers on next failed delivery).
Phase 4.3 - cluster-shared rate limiter via KV semaphore:
- clusterLimiter stores per-host token-bucket state in JetStream KV with
a CAS update loop. N pods sharing a configured RPS now sum to RPS
aggregate, not N*RPS, so a customer with a 10 RPS receiver doesn't
get hammered when we scale horizontally.
- Falls back to the per-pod hostLimiter on persistent KV errors so a
JetStream blip degrades to "weaker but still bounded" rather than
"unlimited." Per-host nextWake cache prevents lockstep KV hammering
on a hot empty bucket.
- New NATS RATE_LIMIT_BUCKET (default webhook_rate_limit) provisioned
with the other KVs. Empty bucket name skips cluster sharing entirely.
- Dispatcher.WithClusterLimiter(kv) opts in; wired in app.go behind
PerHostRPS > 0 + non-empty bucket name.
Also: removed unused watermill ProcessSignalMessage / ProcessEventMessage
plumbing left behind by the Kafka rip; go mod tidy.
All internal -race tests pass. make lint: 0 issues.
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
Adds a JetStream-backed evaluation pipeline behind a
NATS_MODEflag and deprecates the Kafka path. The service can now run with zero Kafka dependencies once DIS publishes to NATS.Modes:
NATS_MODE=off(default) — current Kafka-only behavior, no NATS wiring runs.NATS_MODE=primary— Kafka consumers bridge into NATS, NATS consumers evaluate triggers and dispatch webhooks. Transitional while DIS still publishes to Kafka.NATS_MODE=exclusive— Kafka consumers are not created at all (KAFKA_*env becomes optional). Target state.Subject design keys on signal/event name only (
dimo.signals.<name>,dimo.events.<name>,dimo.trigger.fired.<devLicense>,dimo.dlq.>). Stream subject cardinality scales with name count (~30), not vehicle count. Vehicle DID stays in the CloudEvent payload.Queues now in use:
DIMO_SIGNALS/DIMO_EVENTS— ingest from DISDIMO_TRIGGER_AUDIT— per-fire record published after every successful webhook delivery (async, can't block delivery); keyed on developer license for billing aggregationDIMO_TRIGGER_DLQ— poison messages afterMAX_DELIVERretries land here withX-Original-Subject/X-Failure-Reason/X-Delivered-Countheaders, thenTerm()'d so JetStream stops redeliveringtrigger_stateKV — per-(trigger, vehicle) fire timestamp for distributed cooldown across replicas. Evaluator reads KV first, falls back totrigger_logson miss/error. Listener writes after every successful delivery.Operator tooling:
cmd/triggers-state—list/get/dump/watchover the trigger_state KV bucketcmd/triggers-bench— load + latency probe (parallel publishers, replicas, async/sync, p50/p99/max)cmd/nats-testpublisher— synthesizes vss CloudEvents for local testingObservability:
/healthreturns 503 when NATS is configured and disconnectedvehicle_triggers_nats_publish_total{stream, outcome}andvehicle_triggers_nats_consume_total{stream, outcome}with outcomesok/error/ack/nak/dlqBench numbers
3-node JetStream cluster, replicas=3, file storage, podman on Apple silicon:
Cutover
NATS_MODE=off. No runtime change.NATS_MODE=primary+NATS_URL=...in dev. Kafka still ingests, NATS evaluates. Watch audit stream andnats_consume_total{outcome="nak"|"dlq"}.NATS_MODE=exclusiveand dropKAFKA_*env.Test plan
TestSignalWebhookFlow(existing Kafka path) — no regressionTestNATSSignalFlow— full path via JetStream publish in primary modeTestNATSExclusiveFlow— Kafka consumers nil, webhook fires from JetStream only, audit stream captures, KV state recordedtriggers-stateCLI verified against live NATSinternal/nats/subjects.go; needs sign-off + DIS PR before flipping toexclusivein prodNATS_MODEper environment (chart default staysoff)🤖 Generated with Claude Code