PacketShard

PacketShard — protocol-sharded packet pipeline: Kafka + outbox, Akka.NET routing, at-least-once delivery (offset commit after DB write).

Read side: Debezium CDC streams shard writes back through Kafka into an idempotent Postgres projection (pg_ivm live stats), with Redis fast-path dedup — at-least-once in, exactly-once projected.

High availability (optional): the MySQL outbox tier runs as a semi-synchronous primary/replica pair behind ProxySQL, so an acknowledged packet already exists on two nodes before the client call returns — switched on with one line in .env.

Service discovery: services register themselves with Consul and the gateway resolves its backends from the catalog, so scaling a tier is a Compose change and a failed instance leaves the rotation on its own health check.

Microservice pipeline: packets enter over gRPC through a load balancer, flow through a MySQL outbox → Kafka → Akka.NET MasterNode → 5 MongoDB shards write path, and are projected by CDC (Debezium) into a Postgres read model (pg_ivm) — a CQRS split with the protocol shards as the write side and Postgres as the query side. End-to-end the pipeline guarantees at-least-once delivery: a Kafka offset is committed only after the packet has been durably written to its shard, and the read-side projection turns at-least-once input into an exactly-once result.

Contents: Architecture · Akka.NET routing · At-least-once delivery · Projects · CQRS read side · Scaling · Service discovery · AWS · High availability · Retries & dead-letter · Run it · Tests

Architecture

The diagram below illustrates the complete data flow, from packet ingress to the read model. Two invariants hold at every hop:

Durability before acknowledgment — a stage never confirms a packet until the next durable store (MySQL outbox, Kafka topic, MongoDB shard, Postgres ledger) has accepted it.
At-least-once delivery — srv_sub commits its Kafka offset only after the MasterNode confirms the shard write with "Ok" (offset commit after DB write). A crash at any point causes redelivery, never loss.

┌────────────────────────┐
│ PacketGeneratorConsole │
│ PacketGeneratorClient  │
└────────────────────────┘
             │ gRPC (HTTP/2)
             ▼
┌────────────────────────┐  "which are healthy?" ┌────────────────────────┐
│ LoadBalancer (YARP)    │ ────────────────────▶ │   Consul  :8500        │
│ :5001 gRPC · :5002 REST│ ◀──────────────────── │  catalog + health      │
└────────────────────────┘     instance list     └────────────────────────┘
             │ round-robin                                   ▲
             ▼                                               │ register
┌────────────────────────┐                                   │ + pass /health
│ srv_ingest × 3         │ ──────────────────────────────────┘
│ (gRPC, write)          │
└────────────────────────┘
             │ tx insert (durable before ack)
             ▼
┌────────────────────────┐   poll (SKIP LOCKED)   ┌───────────┐
│     MySQL Outbox       │ ◀───────────────────── │  srv_pub  │
│     (durable Q)        │                        │  (relay)  │
└────────────────────────┘                        └───────────┘
    ▲ one node by default; a primary/replica pair behind ProxySQL
      when the HA overlay is on — see "High availability" below
             │ publish (no dual-write)
             ▼
┌────────────────────────┐
│  Kafka SnapshotTopic   │
│     (5 partitions)     │
└────────────────────────┘
             │ consume
             ▼
┌────────────────────────┐
│ srv_sub × 5            │ ─── commit offset ◀── ONLY on "Ok"
│ (1 per partition)      │      (at-least-once: commit AFTER DB write)
└────────────────────────┘
             │ forward payload (TCP)        ▲
             ▼                              │ "Ok" after shard insert
┌────────────────────────┐                  │
│  MasterNode (Akka.NET) │ ─────────────────┘
│  auth · filter · route │──▶ rejected ✗ → retry ×3 → deadletter
└────────────────────────┘
             │ insert (proto-routed)
             ▼
┌────────────────────────────────────┐
│  5 MongoDB shards (write side, rs0)│
│ HTTPS │ TCP │ UDP │ ARP │ OTHER    │
│ :27018 – :27022                    │
└────────────────────────────────────┘
             │ change streams (CDC)
             ▼
┌────────────────────────────┐
│      Debezium × 5          │
│     (MongoDB connector)    │
└────────────────────────────┘
             │ pcap.<shard>.packets
             ▼
┌────────────────────────────┐
│         Kafka              │
│     (pcap.* topics)        │
└────────────────────────────┘
             │ consume (at-least-once in)
             ▼
┌────────────────────────────┐    INSERT ON CONFLICT ┌───────────────────┐
│        srv_read            │ ────────────────────▶ │ Postgres + pg_ivm │
│   (CDC consumer + API)     │  ② commit first       │   packet_ledger   │
└──────────┬─────────────────┘                       │  UNIQUE(tx_id)    │
           │                                         │  packet_stats_by_ │
           │ ① check   ③ mark (after commit)         │  proto (IMMV)     │
           ▼                                         └───────────────────┘
┌────────────────────────────┐                              ▲ SELECT
│       Redis                │                              │ (no agg)
│    (fast-path filter)      │ ◀────────────────────────────┘
└────────────────────────────┘           web client (GET /stats/*)
                                  ④ Kafka offset commit — always last

Key stages:

gRPC ingress: clients stream packets over gRPC (HTTP/2) to a YARP load balancer, which round-robins them across srv_ingest instances. Each instance writes the packet to the MySQL outbox inside a transaction — durable before the client call returns. The gateway learns which instances exist from Consul rather than from a list it has to be told about (see Service discovery).
Outbox relay: srv_pub polls the outbox with FOR UPDATE SKIP LOCKED and publishes to Kafka, eliminating the dual-write problem: nothing is lost if Kafka is down.
Akka.NET routing: the MasterNode is an Akka.NET actor system that authenticates, filters by protocol, and routes each packet to one of 5 MongoDB shard nodes (see Akka.NET routing).
CQRS read side: Debezium captures changes from the MongoDB shards and streams them via Kafka to srv_read, which projects the data into a Postgres read model (pg_ivm), with Redis as a fast-path duplicate filter.

Akka.NET routing (inside the MasterNode)

Why an actor system — isolation, lock-free concurrency, and an explicit backpressure point

The MasterNode is not a monolithic handler — it is an Akka.NET actor system exposed as a TCP server. Every inbound connection and every shard destination is an actor, which gives the routing stage three properties for free:

Isolation — a malformed packet or a failing shard crashes one actor, not the process; the supervision strategy restarts it while the rest of the pipeline keeps flowing.
Lock-free concurrency — actors process one message at a time from their mailbox, so the auth → filter → route sequence needs no shared-state locking even with 5 srv_sub replicas pushing packets concurrently.
Explicit backpressure point — the "Ok" reply is generated only after the shard actor’s insert succeeds, which is exactly the signal srv_sub waits for before committing its offset.

Message flow — TCP listener → connection handler → filter → 5 shard actors, and the “Ok” back

        TCP (from srv_sub × 5)
             │
             ▼
┌──────────────────────────┐
│   Akka.IO TCP listener   │  accepts connections,
│      (server actor)      │  one handler per socket
└────────────┬─────────────┘
             │ Received(payload)
             ▼
┌──────────────────────────┐
│    Connection handler    │  deserialize PacketMessage,
│    (per-connection)      │  authenticate API key (hash)
└────────────┬─────────────┘
             │ authenticated ✓        ✗ auth fail → reject
             ▼
┌──────────────────────────┐
│      Protocol filter     │  inspect packet `proto`,
│         + router         │  pick target shard
└────────────┬─────────────┘
             │ route by protocol
   ┌─────┬───┴──┬──────┬───────┐
   ▼     ▼      ▼      ▼       ▼
┌─────┐┌─────┐┌─────┐┌─────┐┌───────┐
│HTTPS││ TCP ││ UDP ││ ARP ││ OTHER │   5 ShardNode actors —
│shard││shard││shard││shard││ shard │   each owns one MongoDB
└──┬──┘└──┬──┘└──┬──┘└──┬──┘└───┬───┘   connection, inserts doc
   └──────┴──────┴──────┴───────┘
             │ insert OK
             ▼
      reply "Ok" ──▶ back to srv_sub ──▶ _consumer.Commit(result)

The reply path is the heart of the delivery guarantee: “Ok” flows backwards from the shard insert to the Kafka commit, so the offset moves only when the data is already on disk in Mongo.

When something fails — which failures are reported, which are retried, and which burn an attempt

The routing stage distinguishes a packet that cannot be stored from infrastructure that is momentarily unavailable. Confusing the two either quarantines good packets or retries a poison one forever, so each failure has a defined answer:

What fails	What the MasterNode does	What `srv_sub` does
A shard's MongoDB node is down	`ShardWriterActor` catches the write error and replies `Write failed: …` — the actor stays alive	treats it as a rejection: counts an attempt, re-queues, dead-letters after 3
Invalid API key	replies `Invalid API Key` and closes the connection	the connection drops; nothing is routed
Malformed payload	`proto` cannot be parsed, so it routes to the `OTHER` shard rather than dropping the packet	normal path — it is stored, not lost
A connection handler actor crashes	only that socket's actor dies; the listener and the five shard actors are untouched	reconnects on the next message
The whole MasterNode is down	nothing to reply	`SendAsync` returns `null` — rewind the offset, wait 2s, retry. Not counted as an attempt

That last row is the important asymmetry. An unreachable MasterNode is a transport failure and the packet is still good, so retrying forever is correct. An explicit rejection is a processing failure, so it is counted and eventually quarantined. See Retries & dead-letter.

Two details worth stating precisely, because the actor model is often assumed to do more than it does here:

The shard writers do not actually crash on a failed write. ShardWriterActor catches the exception and answers with an error string, so no supervision restart occurs on the Mongo path — isolation shows up as the other four shards keep working, not as a restart. A shard that is down produces rejections, not an outage.
There is no custom SupervisorStrategy. Anything that does throw gets Akka's default one-for-one restart, which is the right behaviour for a per-connection handler and is why one bad socket cannot take the listener with it.

The write path's own database tier has a separate failover story — see High availability.

At-least-once delivery (offset commit after DB write)

How the guarantee is built — a transactional outbox on the producer side, commit-after-write on the consumer

Two mechanisms combine into an end-to-end at-least-once guarantee, one on each side of Kafka:

Producer side — transactional outbox. srv_ingest never talks to Kafka directly. It writes each packet into the MySQL Outbox table inside the same DB transaction as its business write (IOutbox.AddAsync). The PublishOutboxJob relay then polls the table (concurrency-safe FOR UPDATE SKIP LOCKED reservation), publishes reserved rows to the SnapshotTopic Kafka topic via KafkaMessagePub, and marks them processed; CleanupOutboxJob deletes processed rows. This removes the dual-write problem: if Kafka is down, packets simply wait in MySQL.

Consumer side — commit after DB write. srv_sub consumes the topic and forwards each packet’s payload over a TCP connection to the MasterNode. It commits the Kafka offset only when the MasterNode replies "Ok" — i.e. only after the packet is durably inserted into its MongoDB shard. This is the processed → _consumer.Commit() pattern:

var processed = await _forwarder.SendAsync(envelope.Payload, stoppingToken);
if (processed) _consumer.Commit(result);

Failure analysis — what happens when a component dies mid-flight

Crash point	Outcome
after outbox insert, before Kafka publish	relay re-reserves the row on restart → published later, nothing lost
after Kafka publish, before offset commit	packet redelivered to `srv_sub` → forwarded again (at-least-once)
after shard insert, before offset commit	packet redelivered → duplicate insert absorbed by read-side dedup
after Postgres commit, before Redis mark	redelivery re-INSERTs → `ON CONFLICT DO NOTHING` absorbs it, no double count

Every row above is an executable test rather than a claim — see Tests.

Why duplicates are expected — and why that makes the read side idempotent

Duplicates are therefore possible by design — and that is exactly why the read side makes its projection idempotent (see CQRS read side).

Projects

Project map — what each service and library does

PacketGeneratorConsole / PacketGeneratorClient: gRPC clients that generate randomized test packets and stream them into the system.
LoadBalancer: YARP-based API gateway — round-robins gRPC (HTTP/2) traffic across the ingest services and REST traffic to srv_read, with JWT auth, per-caller rate limiting and optional TLS. Its destinations are written as discover://<service>:<port> and resolved from whichever registry Discovery:Provider names.
srv_ingest: gRPC service that receives packets and writes them to the MySQL outbox inside a DB transaction.
srv_pub: worker that relays packets from the MySQL outbox to Kafka (FOR UPDATE SKIP LOCKED reservation, publish, mark processed, cleanup).
srv_sub: worker that consumes Kafka, forwards packets over TCP to the MasterNode, and commits the offset only on "Ok" (at-least-once).
MasterNode: Akka.NET TCP server — actor pipeline that authenticates the API key, filters each packet by proto, and routes it to the matching MongoDB shard.
srv_read: CQRS read-side service. CdcConsumer consumes CDC events from Debezium/Kafka and hands each to ProjectionHandler, which projects it into a Postgres read model (pg_ivm) using Redis as a fast-path duplicate filter. Exposes a read API (GET /stats/*).
Shared libraries: Shared (PacketMessage, SnapshotMessage, ProtocolType, serializer, API-key hasher), kafka (KafkaMessagePub, TopicRepository, producer Message), outbox (outbox table, Outbox/Relay, publish + cleanup jobs, MySQL persistence), and discovery (IServiceDirectory with Consul, DNS and static implementations, plus the hosted service that registers a process with Consul and deregisters it on shutdown).
PacketShard.Tests: one test project covering all of the above, split into a fast in-process lane and a Testcontainers-backed lane — see Tests.

The 5 shards — one MongoDB instance per “main package” type

Shard	Protocol(s)	Host port
1	HTTPS / TLS / SSL	27018
2	TCP	27019
3	UDP	27020
4	ARP	27021
5	OTHER (everything else, e.g. ICMP, DNS)	27022

All shards store into database pcap, collection packets.

Outbox notes — MySQL implementation details

EF provider is Pomelo MySQL; the outbox transaction uses RepeatableRead isolation.
Outbox Id is CHAR(36) (a UUID()).
The reservation stored procedure GetDataFromTempTable is created on startup.
Both srv_ingest and srv_pub run IOutboxInitializer.InitializeAsync on startup (with retry, since MySQL may still be warming up). It is idempotent, so whichever wins the race is fine.
The connection string is read from ConnectionStrings:Outbox, falling back to SqlConnStr. That fallback is what lets the HA overlay repoint the apps at ProxySQL without touching the base compose file or appsettings.json.

CQRS read side — CDC → Postgres (pg_ivm) + Redis

Why CDC — the shards hold the post-routing truth

The pipeline above is the write side. The read side adds an analytics/query model without touching it: Debezium captures what actually landed in the Mongo shards (post-routing truth) and streams it to a dedicated microservice that projects it into Postgres, where a pg_ivm incrementally maintained view keeps per-protocol summaries live. The full read-side flow — shards → Debezium → Kafka pcap.* topics → srv_read → Postgres/Redis, with the numbered ①–④ processing order — is shown in the tail of the main architecture diagram above.

Why CDC and not just a new consumer group on SnapshotTopic? Because the shards hold the post-routing truth — what survived auth → filter → routing. Rejected and dead-lettered packets never reach the shards, so reading the shards (via CDC) summarizes what was actually stored, not what was merely published.

Correctness — at-least-once in, exactly-once projected

Kafka is at-least-once (by design — see At-least-once delivery), so the projection is made idempotent. ProjectionHandler (srv_read/ProjectionHandler.cs) applies a deliberate, crash-safe order to every event; CdcConsumer around it is only Kafka plumbing — subscription, the consume loop, and when to commit:

Redis fast-path — a read-only EXISTS rm:tx:<id> check skips known duplicates before they cost a Postgres round-trip. Redis is a filter, never the source of truth.
Postgres commit — one transaction applies both guards:
- dedup: INSERT … ON CONFLICT (transaction_id) DO NOTHING (permanent, not TTL-bound);
- ordering: client_state upsert with WHERE EXCLUDED.version > client_state.version (drops “hello from the past” for last-value views; inert for commutative counts).
Redis mark — SET rm:tx:<id> happens only after the commit.
Kafka ack — commit the offset last.

A crash between steps 2 and 3/4 is safe: on redelivery Redis still says “not seen”, the re-INSERT hits ON CONFLICT DO NOTHING, and nothing is lost or doubled. This is the Postgres-first order — the only crash-safe one. Reversing steps 2 and 3 would make Redis lie: the fast path would skip a transaction Postgres never stored, and the packet would vanish without a trace. Note the symmetry with the write side: both srv_sub and srv_read follow the same principle — persist first, acknowledge (commit the offset) last.

Splitting the handler out of the consumer is what makes that order testable: it can be driven against a real Postgres and a real Redis with no broker in the loop (see Tests).

The per-protocol summary is a pg_ivm IMMV (postgres/init.sql): count/min/max are commutative, so a trigger maintains them on every INSERT — no REFRESH, no query-time aggregation.

Scaling

How each stage scales — coordination delegated to the datastore or to Kafka

Every stateless stage scales horizontally; coordination is delegated to the datastore or to Kafka instead of app-level locks:

srv_ingest ×3 behind YARP — ingest instances are stateless; the load balancer round-robins gRPC streams, and durability lives in the shared MySQL outbox. The replica count is not configured on the gateway: instances register themselves with Consul and the gateway resolves them, so a fourth replica joins the rotation on its own.
srv_pub ×3 — all producers write to the same MySQL outbox; the relay reserves rows with FOR UPDATE SKIP LOCKED, so the 3 instances never double-publish.
srv_sub ×5 — Kafka gives one consumer per partition per group, so the topic is created with 5 partitions (TopicPartitions, set in kafka/TopicRepository.cs) and each of the 5 consumers gets its own partition — parallelism without rebalancing churn.
MasterNode (Akka.NET) — concurrency inside a single process comes from the actor model: one handler actor per connection, one shard actor per MongoDB node, each with its own mailbox.

Replica counts are set in docker-compose.yml via deploy.replicas.

Verify the scaling — commands

docker compose up -d --build            # replicas come from deploy.replicas
docker compose ps                       # srv_pub-1..3, srv_sub-1..5
# proof all 5 consumers are active (5 partitions across 5 CONSUMER-IDs, lag ~0):
docker exec packetshard-kafka kafka-consumer-groups \
  --bootstrap-server localhost:9092 --describe --group ConsumerGroup

Service discovery

What it replaced — a gateway that had to be told what exists

The gateway used to name its backends: ingest-1, ingest-2, ingest-3. That list is a second place the replica count lives, and it drifts — scaling meant editing gateway config and restarting it, and a dead instance kept its slot until YARP's own probes noticed. Now instances publish themselves and the gateway reads the registry.

Only the addresses became dynamic. Routes, load-balancing policy, health-check policy and HTTP version stay in appsettings.json, where they are readable:

"Destinations": {
  "ingest": { "Address": "discover://srv-ingest:8080" }
}

ServiceDestinationResolver (YARP's IDestinationResolver) expands that into one destination per live instance — ingest[0], ingest[1], … — and re-resolves when the set changes. Addresses that are not discover:// pass through untouched, so a cluster can mix discovered and pinned destinations. A lookup that fails returns the last known instances rather than none: an unreachable registry is not evidence the backends went away.

Which registry answers is a deployment setting, not an application one:

`Discovery:Provider`	Backed by	How a change is noticed	Used by
`Static` (default)	the `Discovery:Fallback` map	it never changes	no infrastructure at all
`Consul`	the agent's health endpoint	blocking query — a parked connection, not a poll	`docker-compose.yml`
`Dns`	A records, Cloud Map on AWS	a refresh interval; DNS cannot push	`aws/`

Callers depend on IServiceDirectory and never learn which one answered — which is why the AWS deployment needs no code change, only Discovery__Provider=Dns. The port in the sentinel is there for Dns: Consul stores each instance's port, but an A record carries an address and nothing else.

Who registers — and the one surprise

Service	Registers as	Port	Check
`srv_ingest`×3	`srv-ingest`	8080	HTTP `GET :8081/health`
`srv_read`	`srv-read`	8080	HTTP `GET :8080/health`
`MasterNode`	`masternode`	8000	TCP connect — it speaks a line protocol, not HTTP
`LoadBalancer`	`gateway`	5001	HTTP `GET :5002/health`

Registration runs in the background and retries rather than blocking startup, and each check starts critical so a slow-booting instance never receives traffic it cannot serve. A clean shutdown deregisters; a kill is swept up by DeregisterCriticalServiceAfter.

The surprise: srv_ingest listens twice. Its service port carries gRPC, which is HTTP/2-only — plaintext has no ALPN to negotiate with — while Consul's HTTP check speaks HTTP/1.1. So the same /health is exposed on a second HTTP/1.1 listener (HealthPort, 8081). The gateway and the agent probe the same endpoint on different ports.

srv_sub resolves the MasterNode the same way when MasterNode__Service is set, rotating instances on reconnect; leave it unset and the fixed MasterNode__Host is dialled as before.

Verify it — commands

docker compose up -d --build
open http://localhost:8500                       # catalog UI

curl -s 'localhost:8500/v1/health/service/srv-ingest?passing=true' \
  | jq -r '.[] | "\(.Service.ID)  \(.Service.Address):\(.Service.Port)"'

docker compose logs -f loadbalancer | grep "instance(s) of"

Add a replica — copy the srv_ingest-3 block, give it its own hostname: (that is what keeps its service id distinct) — and the gateway picks it up with no config edit and no restart. Kill one and the reverse happens: the check goes critical, Consul drops it, the gateway re-resolves.

Honest caveats

One agent, one server. -bootstrap-expect=1 has no quorum, so it is a single point of failure for discovery — though not on the data path: in-flight requests never touch Consul and the last known instances keep serving. A service that starts during an outage cannot register. Production wants 3 or 5.
No ACLs, no TLS. The agent runs open inside the Compose network. Discovery:Consul:Token exists for the ACL case; nothing sets it here.
Consul is not a load balancer. It answers "which instances are healthy"; YARP still decides where each request goes and keeps its own health checks. The overlap is deliberate — Consul catches a dead process, YARP catches one that is reachable but failing requests.

AWS (optional)

The deployment — the whole pipeline on ECS Fargate, in aws/

aws/ holds Terraform for the entire pipeline on ECS Fargate: NLB → YARP → srv_ingest → RDS MySQL outbox → Kafka (KRaft on EFS) → srv_sub → MasterNode → 5 Mongo shards → Debezium → srv_read → Postgres/pg_ivm, with API Gateway fronting GET /stats/* over a VPC Link. Roughly 17 tasks, about $8–9 a day, and a clean terraform destroy. Full instructions in aws/README-AWS.md.

.github/workflows/deploy-aws.yml is the delivery half: on a push to main it builds the seven images, pushes them to ECR and rolls the services, authenticating through GitHub OIDC so there is no access key in the repo. It never runs Terraform — infrastructure stays a deliberate apply.

It is a hybrid-demo profile on purpose: only the outbox uses a managed service (RDS MySQL); Kafka, Postgres, Mongo and Redis stay as containers so the project's own images and guarantees — pg_ivm, rs0 change streams, the partition count — are the ones actually deployed. README-AWS.md lists the managed upgrade path (MSK, Atlas, ElastiCache) and is honest that terraform validate has not been run against a real provider registry.

Why it needs no application patch — the provider is the only thing that changes

AWS replaces the three fixed ingest replicas with a Cloud Map service whose multivalue A records list every healthy task. That is a different registry, not a different design, so it lands entirely in configuration:

docker-compose:  Discovery__Provider=Consul   ->  ConsulServiceDirectory
aws/ (ECS):      Discovery__Provider=Dns      ->  DnsServiceDirectory
neither:         Discovery__Provider=Static   ->  StaticServiceDirectory

discover://srv-ingest:8080 in appsettings.json is read the same way in all three. The gateway, ServiceDestinationResolver and srv_sub are byte-identical between a laptop and the VPC.

Two things are genuinely different on AWS, and both are handled by the provider rather than by the app: nothing registers itself — ECS registers a task with Cloud Map when it starts it, so the registration hosted service is not even added — and watches become polls, since DNS cannot push a change. The port lives in the sentinel because an A record cannot carry one.

Expanding the record into per-task destinations is the part that matters for throughput. Left as a single DNS name, SocketsHttpHandler picks one address and multiplexes every gRPC stream onto one task, so YARP's round-robin never sees the rest.

High availability (optional)

The overlay — semi-sync MySQL + Orchestrator + ProxySQL, switched in one line of .env

The single mysql service is the write path's one hard dependency: if it is down, srv_ingest cannot accept a packet at all, because the outbox insert is the durability guarantee. ha/ replaces it with a semi-synchronous primary + replica pair, Orchestrator for failover detection and promotion, and ProxySQL routing the apps to whichever node is currently writable.

It ships as a compose overlay, so the main docker-compose.yml needs no edits. The mode is one line in .env:

# HA mode: semi-sync MySQL + Orchestrator + ProxySQL
COMPOSE_FILE=docker-compose.yml:ha/docker-compose.ha.yml

# single-node mode: comment the line out

Everything after that is the usual docker compose up -d --build.

The overlay makes three couplings so the base file stays untouched:

The old mysql is parked, not deleted — compose cannot remove a service during a merge, but profiles: ["disabled"] means nothing ever starts it. Parking alone is not enough, though: compose pulls a profiled service back in when an active service still depends_on it, so the app overrides drop that dependency with depends_on: !override (compose ≥ 2.24).
The apps wait for a writer. srv_ingest-1..3 and srv_pub gain depends_on: ha-bootstrap (service_completed_successfully), so they never start against a node that is still super_read_only.
The connection string is repointed via ConnectionStrings__Outbox → ProxySQL on :6033. The base file's SqlConnStr is still present in the merged environment and is simply outranked, which is what makes the toggle symmetric in both directions.

Component	Role	Port
`mysql-master`	semi-sync source; writable only by runtime appointment	—
`mysql-slave`	semi-sync replica, `super_read_only` until promoted	—
`proxysql`	routes the app to hostgroup 0 (the writer), follows `super_read_only`	6033 (app), 6032 (admin)
`orchestrator`	topology detection, promotion, re-parenting; web UI + API	3000
`ha-bootstrap`	one-shot: appoints the initial writer, registers the topology	—

Data flow through a failover — what happens to in-flight packets when the primary dies

Steady state — every app connection goes to ProxySQL, which keeps exactly one node in the writer hostgroup and decides which by polling super_read_only:

srv_ingest × 3 ─┐
                ├──▶ ProxySQL :6033 ──▶ hostgroup 0  ┌──────────────┐
srv_pub (relay)─┘      (writer only)    ═══════════▶ │ mysql-master │  read_only = OFF
                                                     └──────┬───────┘  (appointed)
                                        hostgroup 1         │ semi-sync AFTER_SYNC
                                        (parked, no traffic)│ commit waits for the
                                                     ┌──────▼───────┐ replica's ack
                                                     │ mysql-slave  │  super_read_only = ON
                                                     └──────────────┘

The source commits only after a replica has the binlog event, so an acknowledged outbox row exists on two nodes before srv_ingest returns to the client — the same durability before acknowledgment invariant the rest of the pipeline follows.

When the primary dies:

  ① mysql-master gone          ② promote                    ③ ProxySQL re-elects
┌──────────────┐            ┌──────────────┐              ┌──────────────┐
│ mysql-master │  ✗         │ mysql-slave  │              │ mysql-slave  │
│   (down)     │            │ super_ro=OFF │              │ hostgroup 0  │◀── writes resume
└──────────────┘            └──────────────┘              └──────────────┘
      │                            ▲                             ▲
      │ writes fail                │ SET GLOBAL (runtime only,   │ monitor sees
      ▼                            │ never persisted)            │ read_only flip
  srv_ingest returns an error ─────┴─────────────────────────────┘
  srv_pub's PublishAsync throws → transaction rolls back

Nothing in flight is lost, because every stage already assumes this can happen:

In flight when the primary dies	What happens
gRPC call mid-`AddAsync`	the insert fails, the client gets an error and retries — the packet was never acknowledged
outbox rows reserved but not yet published	`PublishAsync` throws, the transaction rolls back, `IsProcessing` clears; the next relay tick re-reserves them
rows published to Kafka but not marked processed	the same rows are reserved again after promotion and re-published — Kafka is at-least-once by design, and the read side's `ON CONFLICT DO NOTHING` absorbs the duplicate
rows already marked processed	committed on the old primary and acked by the replica before the commit returned, so they survive the promotion

The failure mode this design refuses is a second writer. A crashed primary that restarts comes back super_read_only (persisted in mysqld-auto.cnf), so it cannot accept writes on the way up; ProxySQL leaves it in the reader hostgroup until something appoints it. Promotion is always SET GLOBAL, which does not survive a restart — so the appointment has to be made deliberately, every time.

Automatic promotion does not currently work. Orchestrator issues SHOW SLAVE STATUS, which MySQL 8.4 removed in favour of SHOW REPLICA STATUS, so topology discovery fails with Error 1064 and step ② never fires on its own. Replication, semi-sync and ProxySQL's re-election all work — a promotion done by hand propagates correctly:
docker exec kafkaflowshard-mysql-slave mysql -uroot -proot \
  -e "SET GLOBAL super_read_only = OFF; SET GLOBAL read_only = OFF;"
Restoring automatic failover means pinning the HA nodes to mysql:8.0; openark/orchestrator has no 8.4-compatible release.

Writability is a runtime appointment (SET GLOBAL, never SET PERSIST): any node that restarts comes back read-only, which is the split-brain failsafe. Details and failover drills are in ha/README-HA.md.

Retries & dead-letter

Three outcomes — commit, retry, or quarantine

srv_sub creates the 5sdelay (retry) and deadletter topics in code at startup (DeadLetterProducer.EnsureTopicsAsync, same as the main topic). Each consumed message resolves to one of three outcomes:

MasterNode result	Action
replies `Ok`	commit ✓
replies but rejects (e.g. shard write failed, malformed payload)	count an attempt → re-queue to `SnapshotTopic` (attempt header `+1`), or `deadletter` once the limit is hit; then commit
unreachable (TCP can’t connect)	rewind offset + wait 2s, retry — not counted as an attempt

Attempt count travels in a Kafka header (attempts); the dead-lettered copy also carries x-failure-reason.
Limit is MaxAttempts (default 3) — a poison message is tried 3× then dead-lettered.
Transient outages don’t burn attempts, so a MasterNode restart won’t dump good packets. This distinction matters for the delivery guarantee: an unreachable MasterNode is a transport failure (rewind and wait — the packet is still good), while an explicit rejection is a processing failure (count it, and quarantine the packet after 3 strikes).

Watch the dead-letter topic fill — commands

docker exec packetshard-kafka kafka-topics --bootstrap-server localhost:9092 --list
# force rejections to see it fill: stop a shard so its writes fail
docker compose stop mongo-arp
docker exec -it packetshard-kafka kafka-console-consumer \
  --bootstrap-server localhost:9092 --topic deadletter --from-beginning

Run it

Option A — everything in Docker (recommended)

cd PacketShard
docker compose up --build

This starts Consul, Zookeeper + Kafka, MySQL, the 5 MongoDB shard nodes, then MasterNode, srv_sub, srv_pub, the ingress tier (LoadBalancer + srv_ingest) and the read side (Debezium, Postgres, Redis, srv_read). Watch the logs: srv_pub publishes, srv_sub forwards, MasterNode prints [shard:Https] saved ... etc.

The catalog UI at http://localhost:8500 shows every service that came up and whether its health check is passing — a quick read on what is actually alive (see Service discovery).

Which MySQL topology comes up depends on .env — a single mysql node by default, or the primary/replica pair behind ProxySQL if the HA overlay is enabled. docker compose ps will tell you which you got.

Inspect what landed in a shard:

docker exec -it packetshard-mongo-https mongosh --eval 'db.getSiblingDB("pcap").packets.find().limit(5)'
docker exec -it packetshard-mongo-arp   mongosh --eval 'db.getSiblingDB("pcap").packets.countDocuments()'

Option B — infra in Docker, apps on the host

cd PacketShard
# Start only Kafka + MySQL + the 5 Mongo shards.
# In HA mode the MySQL service is `mysql-master` (plus `mysql-slave proxysql orchestrator
# ha-bootstrap`) rather than `mysql` — `docker compose config --services` lists what your
# current .env resolves to.
docker compose up -d zookeeper kafka mysql mongo-https mongo-tcp mongo-udp mongo-arp mongo-other

docker compose logs -f srv_pub srv_sub masternode

# In separate terminals (defaults already point at localhost):
dotnet run --project MasterNode
dotnet run --project srv_sub
dotnet run --project srv_pub

Tests

Two lanes — 174 tests split by cost, not by layer

PacketShard.Tests covers the pipeline in two lanes. The split is by cost, not by layer — the unit lane runs anywhere in a few seconds, the infrastructure lane starts real databases via Testcontainers:

dotnet test                                       # 174 tests, ~33s
dotnet test --filter "Category=Unit"              # 133 tests,  ~5s, no Docker required
dotnet test --filter "Category=Infrastructure"    #  41 tests, ~33s, needs Docker

Those are wall-clock times, which is what you actually wait for. xunit's own Duration: line reports only the time spent inside test bodies — it says 2 s and 26 s for the two lanes, excluding build, test-host startup and, for the infrastructure lane, container startup. Most of that lane's wall clock is databases booting rather than assertions running: it starts a fresh Postgres (and, for the projection tests, a Redis) per test rather than sharing one. The MySQL suite shares a container per class and empties the table between tests instead, because MySQL boots an order of magnitude slower.

A full run costs about the same as the infrastructure lane alone — xunit runs test classes in parallel, so the 133 in-process tests finish while the containers are still coming up.

Both lanes are tagged explicitly, so Category=Unit selects the fast one by name. --filter "Category!=Infrastructure" picks the same 133 tests today, but it also sweeps up anything added later without a trait — handy as a CI gate that fails loudly if a new container test forgets its tag, and the wrong choice if you want only what is known to be in-process.

CI runs the two lanes as separate jobs (.github/workflows/ci.yml), so a broken branch is reported by the fast one without waiting on Docker.

Why parts of it need real databases — the guarantees live in the engine, not the C#

Most of the guarantees this README claims are enforced by the database, not by the C#: ON CONFLICT (transaction_id) DO NOTHING, WHERE EXCLUDED.version > client_state.version, FOR UPDATE SKIP LOCKED, and the pg_ivm trigger. Against a mock every one of those tests would pass while the system was broken, so they run against the real engines:

Postgres is built from postgres/Dockerfile, so the container carries pg_ivm and applies postgres/init.sql through the official entrypoint — the schema under test is the deployed one. Redelivering an event leaves one ledger row and an IMMV count of 1; eight concurrent projections of the same transaction insert exactly once.
MySQL proves the reservation: 40 rows, 4 concurrent relay workers each asking for all 40, asserting the union is 40 distinct ids. That is FOR UPDATE SKIP LOCKED and nothing else.
Redis + Postgres together execute the crash-point table above — the durable write commits, the Redis marker is deliberately never set, and the redelivery must be absorbed by Postgres dedup rather than double-counted.

The MasterNode needs no containers — TestProbes in place of the five Mongo writers

The routing stage takes its shard Props as a constructor argument, so the five MongoDB writers can be swapped for TestProbes: a packet with proto: "UDP" is asserted to reach the UDP probe and no other. The auth gate is tested the same way — an invalid API key must produce "Invalid API Key", close the connection, and route nothing.

Failure paths — tested as first-class behaviour, not an afterthought

A test suite that only covers success would miss the point of a durable pipeline. The suite pins the unhappy paths too: a failing Kafka publish rolls the outbox transaction back and leaves the row pending for the next tick; a failing Postgres write leaves no Redis marker behind; an abandoned reservation returns to the pool when it expires; a malformed CDC value is skipped rather than crashing the consumer.

Live pipeline

Live pipeline — interleaved logs from srv_pub, srv_sub and masternode

Interleaved output of docker compose logs -f srv_pub srv_sub masternode: the five srv_sub replicas (srv_sub-1..5) each forward packets and get MasterNode response: Ok, while masternode routes them to the protocol shards — [shard:Other] saved DNS …, [shard:Arp] saved ARP …, etc.

Navigation Menu

Repository files navigation

PacketShard

Architecture

Akka.NET routing (inside the MasterNode)

At-least-once delivery (offset commit after DB write)

Projects

CQRS read side — CDC → Postgres (pg_ivm) + Redis

Scaling

Service discovery

AWS (optional)

High availability (optional)

Retries & dead-letter

Run it

Tests

Live pipeline

About

Releases

Packages

Contributors

Languages

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
LoadBalancer		LoadBalancer
MasterNode		MasterNode
PacketGeneratorClient		PacketGeneratorClient
PacketGeneratorConsole		PacketGeneratorConsole
PacketShard.Tests		PacketShard.Tests
Shared		Shared
assets		assets
aws		aws
debezium		debezium
discovery		discovery
ha		ha
kafka		kafka
outbox		outbox
postgres		postgres
protos		protos
srv_ingest		srv_ingest
srv_pub		srv_pub
srv_read		srv_read
srv_sub		srv_sub
.dockerignore		.dockerignore
.env		.env
.gitignore		.gitignore
PacketShard.slnx		PacketShard.slnx
README.md		README.md
docker-compose.yml		docker-compose.yml

Latest commit

History

Folders and files

Repository files navigation

PacketShard

Architecture

Akka.NET routing (inside the MasterNode)

At-least-once delivery (offset commit after DB write)

Projects

CQRS read side — CDC → Postgres (pg_ivm) + Redis

Scaling

Service discovery

AWS (optional)

High availability (optional)

Retries & dead-letter

Run it

Tests

Live pipeline

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages