broad-announce/M11_PLAN.md

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)

// 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:

// 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)

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

// 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

// 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:

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):

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.