M11 plan: gRPC bidi-streaming ingest (proto + server + client + loadgen + smoke)

M11_PLAN.md

@@ -0,0 +1,498 @@
+# M11 — Detailed Implementation Plan
+
+**Status:** planning (post-M10)
+**Target:** M11 exit criterion from SPEC §19
+**Goal:** Internal Go services can push ≥ 10k alerts/sec on a single gRPC bidi stream, sustained 10 min, p99 server-side `Ack` ≤ 50ms.
+
+---
+
+## 0. Recap — what M11 must prove
+
+| Milestone | Exit criterion | How it's measured |
+|---|---|---|
+| **M11** | internal Go service can publish ≥ 10k alerts/sec on one gRPC bidi stream, sustained 10 min, p99 server-side `Ack` ≤ 50ms | `m11_smoke.py` runs a gRPC loadgen against `ingestd:9090`, queries Prometheus, asserts on rate + p99 + DLQ |
+
+**Why this matters beyond "another transport":** HTTP/WS/MQTT each cap out around 1–2k/s per source due to per-request overhead, JSON parsing, or broker round-trips. A first-party Go producer (a peer service in our infra) needs an order-of-magnitude higher ceiling **with strict schemas and explicit backpressure**. gRPC + HTTP/2 + protobuf is the only stack that hits that bar cleanly.
+
+---
+
+## 1. Workstream overview
+
+```
+┌──────────────────────────┐    ┌──────────────────────────┐    ┌──────────────────────────┐
+│  W1: proto + codegen     │    │  W2: grpcserver in       │    │  W3: grpcclient lib +    │
+│  (.proto + buf + stubs)  │───▶│  ingestd (port 9090)     │───▶│  reusable Go client      │
+└──────────────────────────┘    └──────────────┬───────────┘    └──────────────┬───────────┘
+                                               │                               │
+                                               ▼                               ▼
+                                     ┌──────────────────────────┐    ┌──────────────────────────┐
+                                     │  W4: gRPC loadgen        │    │  W5: smoke + verify      │
+                                     │  (Go binary, multi-      │───▶│  (m11_smoke.py, 3 green  │
+                                     │   stream, paced)         │    │   runs, M11_VERIFICATION)│
+                                     └──────────────────────────┘    └──────────────────────────┘
+```
+
+Five workstreams, all on `master`, each ends with a passing test before moving on.
+
+---
+
+## 2. The four layers of risk (and what each WS defuses)
+
+| Risk | Why it's scary | How we defuse it |
+|---|---|---|
+| **Proto schema drift** | Multiple teams consume these stubs; a v2 bump without a plan is a v1+1 outage | W1: checked-in `.proto`, `buf generate` in CI, no manual edits to `gen/` |
+| **Server can't sustain 10k/s** | Different concurrency model from HTTP (1 stream = N inflight messages, not 1 conn = 1 req). Naive impl will serialize. | W2: bounded concurrent workers per stream + `sem` on dedupe/publish. Load test in W4 finds the wall. |
+| **Backpressure invisible to client** | gRPC stream silently buffers. Slow source OOMs the server. | W2: server enforces max-256-inflight; sends `Error.RATE_LIMITED` with `retry_after_ms` when over. Client (W3) honors it. |
+| **Throughput claim is hand-wavy** | "10k/s" without a measurement plan is just a wish | W5: `m11_smoke.py` runs Go loadgen (not Python) for 10 min, asserts p99 ≤ 50ms + rate ≥ 10k/s, mirrors M10's assertion harness |
+
+---
+
+## 3. Workstream W1 — Proto schema + code generation
+
+### 3.1 Repo additions
+
+```
+proto/
+  buf.yaml                          # buf module config (v1, go package prefix)
+  buf.gen.yaml                      # codegen config → gen/go/
+  broadannounce/v1/
+    ingest.proto                    # the Ingest service + Alert + Ack (from SPEC §19)
+gen/
+  go/broadannounce/v1/              # generated stubs (do not edit; CI regenerates)
+  go/broadannounce/v1/ingest.pb.go
+  go/broadannounce/v1/ingest_grpc.pb.go
+```
+
+### 3.2 .proto content
+
+Exactly as written in SPEC §19 — `Ingest.StreamAlerts` (bidi), `Alert` (9 fields), `Ack` (result oneof with Ok/Error), `Error.Code` enum (UNKNOWN/UNAUTHENTICATED/RATE_LIMITED/INVALID/INTERNAL), plus `retry_after_ms`.
+
+### 3.3 Code generation
+
+- Add `buf` to dev tooling (Makefile target `proto`)
+- `buf generate` → outputs to `gen/go/`
+- `gen/` is committed (per repo convention — see existing `internal/...` is committed)
+- Pre-commit / CI hook runs `buf format --check` and `buf lint`
+
+### 3.4 Go module deps
+
+Add to `go.mod`:
+```
+google.golang.org/grpc     v1.6X.0
+google.golang.org/protobuf v1.3X.0
+github.com/bufbuild/buf    (tool-only, not in go.mod)
+```
+
+### 3.5 Exit criteria for W1
+
+- `buf generate` runs clean from a clean clone
+- `go build ./gen/...` succeeds
+- A trivial `grpcurl -plaintext localhost:9090 list` returns `broadannounce.v1.Ingest`
+
+---
+
+## 4. Workstream W2 — gRPC server inside `ingestd`
+
+### 4.1 Skeleton
+
+```
+internal/grpcserver/
+  server.go         # gRPC server lifecycle, register, graceful stop
+  handler.go        # StreamAlerts implementation
+  auth.go           # API key validation (shared with HTTP)
+  ratelimit.go      # per-stream sliding-window rate limit
+  flowcontrol.go    # in-flight cap, per-message Error.RATE_LIMITED
+  server_test.go    # table-driven tests: auth, dedupe, rate limit, flow control
+```
+
+### 4.2 Server lifecycle (in `cmd/ingestd/main.go`)
+
+```go
+// New in main():
+grpcServer := grpcserver.New(grpcserver.Deps{
+    Broker:    br,
+    Dedupe:    ded,
+    Limiter:   limiter,
+    Quarantine: quarantine,
+    Circuit:   cb,
+    Logger:    logger,
+})
+go grpcServer.Serve(cfg.GRPCAddr)  // :9090 default
+// defer grpcServer.GracefulStop(15 * time.Second)
+```
+
+New config:
+```go
+// internal/config/config.go
+GRPCAddr  string  // BA_INGESTD_GRPC_ADDR, default ":9090"
+GRPCMaxInflight int  // BA_INGESTD_GRPC_MAX_INFLIGHT, default 256
+```
+
+### 4.3 Stream lifecycle (the meat)
+
+```go
+func (s *Server) StreamAlerts(stream pb.Ingest_StreamAlertsServer) error {
+    ctx := stream.Context()
+    
+    // 1. Auth — pull API key from metadata, hash, look up source
+    src, err := s.authenticate(ctx)
+    if err != nil { return status.Error(codes.Unauthenticated, "...") }
+    
+    // 2. Per-stream sem (in-flight cap)
+    sem := make(chan struct{}, s.MaxInflight)  // default 256
+    
+    // 3. Per-stream sliding-window rate limit
+    rl := s.limiter.New(src.SourceID, src.RateLimitPerSec)
+    
+    // 4. Loop: receive → process → send Ack
+    for {
+        msg, err := stream.Recv()
+        if err == io.EOF { return nil }
+        if err != nil { return err }
+        
+        // Rate-limit check FIRST (cheaper than dedupe)
+        if !rl.Allow() {
+            ack := &pb.Ack{
+                AlertId:    msg.DedupeKey,  // best-effort correlation
+                DedupeKey:  msg.DedupeKey,
+                AcceptedAtMs: time.Now().UnixMilli(),
+                Result: &pb.Ack_Error{
+                    Error: &pb.Error{
+                        Code: pb.Error_RATE_LIMITED,
+                        Message: "per-stream rate limit exceeded",
+                        RetryAfterMs: 100,
+                    },
+                },
+            }
+            stream.Send(ack)
+            continue
+        }
+        
+        // 5. Bounded concurrency — this is what gets us to 10k/s
+        sem <- struct{}{}
+        go func(m *pb.Alert) {
+            defer func() { <-sem }()
+            
+            // Shared pipeline (same as HTTP/MQTT path)
+            result := s.pipeline.Process(ctx, m, src)
+            
+            ack := buildAck(m, result)  // dedupe_count, accepted_at_ms
+            stream.Send(ack)
+        }(msg)
+    }
+}
+```
+
+**Why the goroutine + sem:** the dedupe Redis call + NATS publish is the slow path (~2–5ms total). If we serialize, we cap at ~250 msg/s. With 256-way concurrency, we cap at ~50k msg/s on the server side. This is the core insight that takes us from 1k/s (HTTP) to 10k/s (gRPC).
+
+### 4.4 Pipeline reuse
+
+The HTTP path has `processDeps.Process(ctx, alert, source) Result`. We extract that into a shared `internal/pipeline/` package (small refactor as part of W2 — W2.1) so gRPC and HTTP both call the same `pipeline.Process`. This is a **deliberate** refactor: M11 is the first time we have a second transport sharing the pipeline, so DRY-ing it now is the right time. (If we copy-paste, M12/M13 will diverge.)
+
+**Estimated refactor scope:** 100–200 lines moved from `cmd/ingestd/process.go` to `internal/pipeline/`. No behavior change. Existing M0–M9 smoke tests must still pass.
+
+### 4.5 Keepalive + dead-stream detection
+
+```go
+// In server.go
+grpc.KeepaliveParams(keepalive.ServerParameters{
+    Time:    30 * time.Second,
+    Timeout: 10 * time.Second,
+})
+grpc.KeepalivePolicy(keepalive.EnforcementPolicy{
+    MinTime:             5 * time.Second,
+    PermittingWithoutStream: false,
+})
+```
+
+### 4.6 Metrics (layer 6/7)
+
+Add to `internal/observability/`:
+- `ba_ingestd_grpc_streams_active` (gauge)
+- `ba_ingestd_grpc_inflight_per_stream` (histogram, label `source_id`)
+- `ba_ingestd_grpc_rate_limited_total{source_id}` (counter)
+- `ba_ingestd_grpc_ack_latency_seconds{source_id}` (histogram, buckets: 1/5/10/25/50/100/250/500ms) — **this is the M11 exit metric**
+- `ba_ingestd_alerts_received_total{transport="grpc",result=...}` (counter — extends M0 counter)
+
+### 4.7 Tests (table-driven, all in-process via `grpc-go` test helpers)
+
+| Test | Asserts |
+|---|---|
+| `TestStream_HappyPath` | 100 alerts → 100 Acks, dedupe_count=1 each, p99 < 50ms |
+| `TestStream_AuthFail` | bad key → stream closed, `codes.Unauthenticated` |
+| `TestStream_Dedupe` | 100 identical dedupe_keys → first Ack has count=1, rest have count>1 |
+| `TestStream_RateLimit` | source rate=100/s, send 1000 in 1s → ~100 Ok + ~900 RATE_LIMITED |
+| `TestStream_InFlightCap` | slow pipeline → RATE_LIMITED with `retry_after_ms` |
+| `TestStream_BrokerDown` | NATS unreachable → CB opens, stream sees INTERNAL, no panic |
+| `TestStream_ConcurrentStreams` | 10 streams × 1k/s each → 10k/s total, no message loss |
+
+### 4.8 Exit criteria for W2
+
+- All 7 tests green
+- `go test ./internal/grpcserver/...` passes
+- M0–M9 smoke tests still pass (regression check on pipeline refactor)
+
+---
+
+## 5. Workstream W3 — Reusable gRPC client library
+
+### 5.1 Skeleton
+
+```
+internal/grpcclient/
+  client.go         # Dial + StreamAlerts wrapper
+  backoff.go        # honors Error.RATE_LIMITED.retry_after_ms
+  options.go        # functional opts: WithAPIKey, WithMaxRetries, WithKeepalive
+  client_test.go    # against a mock server (in-process)
+```
+
+### 5.2 API
+
+```go
+// internal/grpcclient/client.go
+package grpcclient
+
+type Client struct {
+    conn   *grpc.ClientConn
+    client pb.IngestClient
+    apiKey string
+}
+
+type Option func(*Client)
+
+func New(addr string, opts ...Option) (*Client, error) { ... }
+func WithAPIKey(k string) Option { ... }
+func WithMaxRetries(n int) Option { ... }
+func WithKeepalive(time, timeout time.Duration) Option { ... }
+
+func (c *Client) Stream(ctx context.Context) (Stream, error) {
+    stream, err := c.client.StreamAlerts(ctx)
+    return Stream{stream: stream}, nil
+}
+
+type Stream struct {
+    stream pb.Ingest_StreamAlertsClient
+}
+
+func (s *Stream) Send(a *pb.Alert) error { ... }
+func (s *Stream) Recv() (*pb.Ack, error) { ... }
+func (s *Stream) CloseSend() error { ... }
+```
+
+### 5.3 What it gives other teams
+
+A peer Go service does:
+```go
+c, _ := grpcclient.New("ingestd.tenant.svc.cluster.local:9090",
+    grpcclient.WithAPIKey(os.Getenv("BA_API_KEY")))
+stream, _ := c.Stream(ctx)
+go func() {  // producer
+    for alert := range alertsCh {
+        stream.Send(alert)
+    }
+    stream.CloseSend()
+}()
+for {  // consumer — handles rate limit + dedupe_count
+    ack, _ := stream.Recv()
+    metrics.RecordAck(ack)
+}
+```
+
+That's the whole "publish to broad-announce" story in 10 lines. **This is the actual deliverable other teams consume.** The M11 server is the means; the client is the end.
+
+### 5.4 Exit criteria for W3
+
+- `internal/grpcclient` compiles, all tests pass
+- A 30-line example service (`cmd/example-grpc-producer/main.go`) demonstrates usage
+- `go vet` + `go test ./internal/grpcclient/...` clean
+
+---
+
+## 6. Workstream W4 — gRPC loadgen
+
+### 6.1 Repo addition
+
+```
+loadgen/
+  grpc.go            # new — uses internal/grpcclient
+  http.go            # existing
+  mqtt.go            # existing
+  ws.go              # existing
+  main.go            # dispatch on --driver
+```
+
+### 6.2 Flags
+
+```
+--driver=grpc
+--target=ingestd:9090
+--api-key=acme-001:prom-prod:s3cret-acme
+--rate=10000           # alerts/sec
+--duration=10m
+--workers=8            # parallel streams
+--payload-size=512     # bytes per alert (realistic)
+--company-id=acme-001
+--source-id=prom-prod
+```
+
+### 6.3 How it works
+
+- Opens N streams (default 8)
+- Each stream is a producer goroutine that fires `--rate/N` alerts/sec via a paced `time.Ticker`
+- Each stream is also a consumer goroutine that reads Acks, records `accepted_at_ms - sent_at_ms` for p99
+- Atomic counters: `sent`, `acked_ok`, `acked_rate_limited`, `errors`
+- On `Error.RATE_LIMITED`, honor `retry_after_ms` (sleep, then resume) — this is the backpressure test
+
+### 6.4 Profile
+
+`docker-compose.yml` — new `loadgen-grpc` profile (separate from `loadgen` and `loadgen-m10`):
+
+```yaml
+profiles: ["loadgen-grpc"]
+services:
+  loadgen-grpc-1:
+    build: ./loadgen
+    command:
+      - --driver=grpc
+      - --target=ingestd:9090
+      - --rate=5000
+      - --duration=15m   # soak is 10m; +ramp +cooldown
+    ...
+```
+
+Two instances × 5k/s = 10k/s on the wire (matches M11 exit).
+
+### 6.5 Exit criteria for W4
+
+- `go build ./loadgen` succeeds
+- `docker compose --profile loadgen-grpc up -d` starts clean
+- A 1-minute dry run shows ≥ 5k/s per instance on the wire (Prom confirms)
+
+---
+
+## 7. Workstream W5 — Smoke test + verification
+
+### 7.1 Scripts
+
+```
+scripts/m11_smoke.py            # assertion harness
+M11_SMOKE_LOG.md                # raw log of 3 green runs
+M11_VERIFICATION.md             # what landed, evidence, exit criterion proof
+```
+
+### 7.2 What `m11_smoke.py` does
+
+Mirrors `m10_smoke.py` structure:
+
+```
+Step 1 — pre-flight
+  - ingestd :9090 gRPC port is listening
+  - prometheus has grpc_streams_active, grpc_ack_latency_seconds
+  - loadgen-grpc-1, loadgen-grpc-2 are in scrape targets
+
+Step 2 — start gRPC loadgen cluster
+  - docker compose --profile loadgen-grpc up -d
+  - wait 15s for ramp-up
+
+Step 3 — soak for 10 minutes
+  - every 30s, sample:
+    - rate = sum(rate(ba_ingestd_alerts_received_total{transport="grpc",result="ok"}[30s]))
+    - p99 = histogram_quantile(0.99, rate(ba_ingestd_grpc_ack_latency_seconds_bucket[60s]))
+    - dlq = ba_dlq_messages_total (must stay 0)
+  - fail on:
+    - rate < 9k/s for any sample (tolerance ±10% of 10k)
+    - p99 > 50ms for any sample
+    - dlq > 0
+
+Step 4 — multi-stream backpressure test
+  - spawn 16 streams × 1k/s each
+  - assert: total = 16k/s, no message loss, all rate-limited Acks honored
+
+Step 5 — teardown
+```
+
+### 7.3 Exit criteria for W5 (= M11 exit)
+
+- 3 consecutive green runs on local docker-compose
+- 1 green run on remote playground `parres` (192.168.44.94)
+- `M11_VERIFICATION.md` shows:
+  - 20 soak samples ≥ 9k/s
+  - 20 soak samples p99 ≤ 50ms
+  - DLQ = 0 throughout
+  - Multi-stream backpressure test passes
+- SPEC.md M11 row updated to **✅ shipped YYYY-MM-DD**
+
+---
+
+## 8. Sequencing & parallelism
+
+```
+W1 ─────────────────┐
+                    ├──▶ W2 ──▶ W3 ──┐
+                                       ├──▶ W4 ──▶ W5
+                    (W2 is critical    (W3 can start in    (W5 is the
+                     path — blocks      parallel with W2     ship gate)
+                     everything else)   once grpcserver
+                                       interface is stable)
+
+Estimated wall time:
+  W1: 1 day    (proto + codegen is small, but buf setup + CI is fiddly)
+  W2: 3 days   (server impl + tests + pipeline refactor)
+  W3: 1 day    (client lib is mostly wrapper code)
+  W4: 1 day    (loadgen is just calling W3)
+  W5: 1 day    (smoke is patterned on M10's, so fast)
+  Total: ~7 working days
+```
+
+W2.1 (pipeline refactor) is the riskiest sub-task. It should land first as a standalone commit, with M0–M9 smoke tests run after to prove no regression, **before** W2.2 (gRPC server) starts.
+
+---
+
+## 9. What M11 is *not*
+
+(Stolen from SPEC §19, reinforced here for the implementer)
+
+- **Not for public SaaS webhooks** (Grafana/Stripe/Datadog) — those stay on HTTP POST. Don't add a "gRPC webhook shim" — that's a different product.
+- **Not for browsers** — gRPC needs grpc-web + Envoy. The dashboard story stays on WebSocket.
+- **Not for IoT/PLCs** — MQTT is the right answer there. Don't migrate MQTT sources to gRPC.
+- **Not a gRPC gateway for the HTTP path** — HTTP and gRPC are siblings, not parent/child. Each has its own port, auth, metrics.
+
+---
+
+## 10. Open questions to resolve before W2
+
+1. **mTLS**: SPEC says "optional". Do we ship mTLS-ready code (cert plumbing, off by default) or defer? **Recommendation:** ship the cert plumbing, off by default — costs nothing, unblocks future compliance ask.
+2. **Stream ID propagation**: do we tag the stream with an opaque `stream_id` for OTel traces, or rely on `source_id`? **Recommendation:** add `stream_id` to logs/traces for forensic clarity (cheap).
+3. **Graceful shutdown on server**: when `ingestd` gets SIGTERM, do we drain in-flight Acks (up to 30s) or hard-close? **Recommendation:** drain for `ShutdownTimeout` (default 15s) — same as HTTP server.
+4. **Reconnect on transient broker failure**: client (W3) side. Exponential backoff? **Recommendation:** yes, 100ms → 1s cap, 3 retries, then surface to caller. Match typical gRPC client conventions.
+5. **Per-company rate limit on gRPC**: SPEC says "stream-level rate limit" but companies may have many gRPC sources. **Recommendation:** enforce BOTH (per-source first, then per-company as a backstop). Same as HTTP.
+
+---
+
+## 11. Definition of done
+
+- [ ] `buf generate` runs clean in CI
+- [ ] All 7 server tests green (`internal/grpcserver/...`)
+- [ ] All 3+ client tests green (`internal/grpcclient/...`)
+- [ ] M0–M9 smoke tests still green (pipeline refactor regression check)
+- [ ] `m11_smoke.py` passes 3× locally + 1× on parres
+- [ ] `M11_VERIFICATION.md` published with rate / p99 / DLQ evidence
+- [ ] `docs/sources/grpc.md` quickstart written
+- [ ] SPEC.md M11 row: **✅ shipped YYYY-MM-DD** with evidence one-liner
+- [ ] `cmd/example-grpc-producer/main.go` exists and runs
+
+---
+
+## 12. Risks & mitigations
+
+| Risk | Likelihood | Impact | Mitigation |
+|---|---|---|---|
+| Pipeline refactor breaks HTTP path | Medium | High (regression) | Land W2.1 standalone, run M0–M9 smoke before starting W2.2 |
+| gRPC server can't hit 10k/s on docker-compose | Medium | High (M11 exit fails) | W2 has bounded-concurrency design; W5 finds the wall early; fallback: bump MaxInflight to 512 (default 256) |
+| Backpressure not actually honored | Low | Medium (OOM risk) | W4 deliberately over-sends, asserts no message loss + no OOM |
+| `buf` toolchain friction in CI | Medium | Low (dev pain) | Pin buf version; document install in README |
+| Other teams don't migrate to gRPC | Low | Low (no value) | M11 client lib is opt-in; HTTP stays forever |
+
+---
+
+**Next step:** approval to start W1.