# 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.