Real-Time and Streaming System Design Questions

Situation: Last year, our realtime messaging service (handles presence and delivery for 2M daily users) experienced a cascading outage during a traffic spike after a marketing campaign — latency jumped, message delivery dropped to ~40%, and SLO error budget was exhausted within 30 minutes.

Task: As the on-call SRE lead, I needed to restore service quickly, coordinate stakeholders (product, engineering, customer support, and execs), minimize customer impact, and ensure a clear postmortem and remediation plan.

Action:

• Immediate coordination: I declared an incident, created a shared incident channel, and assigned clear roles (communications, mitigation, RCA owner). I ran a 10-minute triage to align priorities.
• Technical triage: I examined monitoring dashboards (latency histograms, queue depth, pod CPU/mem) and logs. We identified a combination of overloaded message brokers (increased ack latency) and a recent deployment that increased per-message CPU cost.
• Mitigation: I ordered an immediate rollback of the deployment while engineering scaled broker replicas and increased consumer parallelism via a parameter flag. I applied a temporary rate limiter at the ingress proxy to protect backends.
• Communication: I provided 15-minute updates to stakeholders and a customer-facing status page update. I coordinated with customer support to prioritize high-value customers.
• Post-incident: Led the postmortem session within 48 hours, documented timelines, decisions, and contributing factors; assigned remediation tasks with owners and deadlines.

Result:

• Service returned to normal delivery rates within 22 minutes of rollback; message delivery climbed from 40% to 99.5%. Incident impact: ~0.8% of daily messages delayed >1 minute; no data loss. We preserved customer SLAs and reduced potential churn.

What I changed afterwards:

• Rolled out stricter pre-deploy load tests and a canary policy that exercises CPU-heavy paths.
• Implemented automated broker autoscaling and improved alerts for broker ack latency and consumer lag thresholds.
• Created an incident runbook with clear role assignments and a templated customer communication cadence.
• Tracked remediation completion in our reliability roadmap and measured a 60% reduction in similar incidents over the next quarter.

This experience reinforced the importance of decisive coordination, fast risk-mitigating technical actions (rollback + throttling), and strengthening expectations (canaries, tests, runbooks) to prevent recurrence.

Requirements clarified:

• Strict per-instrument ordering
• ≤10 ms tail latency (in-region)
• Replay of missed updates up to 1 minute
• At-most-once delivery to downstream processors
• 100k updates/sec aggregate

High-level architecture:

• Ingest layer: low-latency collectors (co-located with exchange gateways) receive raw updates, validate, normalize, attach wall-clock and source metadata.
• Sequencer layer: shard by instrument (hash on instrument id). Each shard is a single-threaded sequencer process that assigns monotonically increasing sequence numbers per instrument and appends the envelope to an in-memory append-only ring buffer. Single-threaded per-shard preserves strict ordering and minimizes lock overhead.
• Transport layer: primary: UDP multicast (or PGM) per shard/region for fan-out; secondary: TCP/QUIC per-client for reliable fallback. Messages carry (instrument, seq, TS, payload).
• Replay storage: in-memory ring buffer (circular) per sequencer sized for 1 minute of traffic; optionally backed by local NVMe for fast spillover. Each message lookup by sequence number.
• Subscriber model: clients subscribe to instrument partitions. They consume via multicast; if they miss packets or experience loss, they request replay via a light-weight RPC to sequencer (give last-seen seq -> sequencer streams missed messages over TCP/QUIC).
• At-most-once guarantee to downstream processors: deliver messages exactly once from sequencer to downstream by using sequence numbers + checkpointing. Downstream processors must checkpoint highest contiguous sequence consumed to a durable store (etcd/RocksDB). Sequencer avoids re-sending already-ACKed ranges to prevent duplicates. For multicast consumers that can’t ACK (market data clients), system provides sequence numbers and relies on their idempotent processing or replay semantics; for internal downstream processors require ACKs and sequencer enforces one-time delivery by only sending un-ACKed messages.
• Backpressure & flow control:
  • Multicast: no backpressure; sequencer keeps short buffer and relies on replay for slow/missing consumers.
  • TCP/QUIC subscribers: implement credit-based flow control; if a consumer falls behind, sequencer will stop pushing and rely on replay pull. Apply configurable slow-consumer policies: e.g., drop connection after X ms or move to replay-only.
  • Protect sequencer: per-subscriber write queue limits and global emitter rate limiting; apply circuit-breakers and shed non-critical load.
• Scalability:
  • Horizontal by adding sequencer shards (consistent hashing). 100k updates/sec -> with per-shard single-thread throughput easily handles many instruments; benchmark to size CPU (modern core can do ~50-200k UDP msgs/sec depending on payload).
  • Co-locate sequencers with ingest collectors to reduce network hops; use kernel-bypass (DPDK) or SO_REUSEPORT + busy-poll (Linux) to reduce latency.
• Latency optimizations:
  • Single-threaded sequencers avoid locking.
  • Keep hot-path in-memory; avoid syscalls in path; use batching for NIC transmissions but small batch sizes to meet sub-10ms tail.
  • Use UDP multicast inside region; pin threads/cores, use CPU isolation, hugepages.
  • Measure p50/p95/p99 and tune NIC queues, interrupt affinity, and batching.
• Operational considerations:
  • SLOs: 99.9% deliveries <10ms tail; replay latency <50ms.
  • Monitoring: end-to-end latency, packet loss, sequencer queue depth, replay request rate, consumer lag, ACK gaps.
  • Alerting: elevated packet loss, slow consumers, replay storm.
  • Capacity planning: headroom for bursts; autoscale sequencer pool; pre-warm ring buffers.
  • DR and recovery: sequencer persistent checkpoints to NVMe every few seconds; on failover, new leader replays from last checkpoint.
  • Testing: chaos/packet-loss injection, soak tests at >2x load, synthetic consumer slowdowns.
    Trade-offs:
• UDP multicast gives best latency and fan-out but requires consumer-side replay and cannot enforce at-most-once for public clients; internal downstreams use ACKed TCP paths to guarantee at-most-once.
• Using full reliable middleware (Kafka) simplifies replay but adds latency and doesn't meet strict sub-10ms tail without heavy tuning.

This design meets strict ordering (per-shard sequencer), sub-10ms in-region tail (co-located single-threaded sequencers + multicast/DPDK), 1-minute replay (in-memory ring buffer), and at-most-once for internal downstream via sequence+ACK/checkpointing, while providing clear operational controls for backpressure and reliability.

Start with goals: prevent OOM, bound tail-latency, and keep high throughput while supporting many concurrent connections. Key strategies:

Pooling and slab allocators

• Use fixed-size slab arenas for common buffer sizes (e.g., 512B, 4KB, 64KB). Allocation becomes O(1), fragmentation drops, and frees return to pool quickly.
• Keep per-size free lists with low-lock or per-thread caches to avoid contention under high concurrency.
• For rare large allocations use a separate allocator to avoid polluting slabs.

Eviction strategies & memory limits

• Enforce a hard global memory cap for all buffers to prevent OOM kills. Above that, deny new buffers or drop connections.
• Soft limits per-tenant/connection trigger graceful measures: signal backpressure, slow reads, or prioritize critical traffic.
• Eviction: LRU or size-aware policies that drop/flush least-recently-used or largest idle buffers. For streaming, prefer flushing outbound (apply partial sends) before dropping stateful connections.

Avoiding OOM during spikes

• Reserve a small emergency pool for control traffic (health checks, RPCs) so critical control plane survives.
• Apply admission control: if global usage > soft limit, start rejecting new connections or ask upstream to retry with exponential backoff.
• Rate-limit per-connection bursts and use token-buckets so single clients can’t exhaust memory.
• Autoscale instances based on memory pressure and queue length for cloud deployments.

Trade-offs: larger per-connection buffers vs backpressure

• Larger buffers reduce syscalls and increase throughput but increase memory footprint and risk of OOM/GC pressure.
• Backpressure (smaller buffers + flow-control) reduces memory but increases client latency and complexity (retries, congestion signals).
• Practical approach: keep moderate-sized pooled buffers, prefer flow-control for long-lived or many connections, and allow configurable buffer caps per service class.

Monitoring & SRE ops

• Emit metrics: pool usage, allocation rates, evictions, soft/hard-limit breaches, per-connection memory.
• Alert on sustained soft-limit crossings and high eviction rates. Use playbooks: scale, tighten admission control, rotate clients.

This combination gives predictable memory bounds, low fragmentation, and operational controls to survive spikes while balancing throughput and latency.

Immediate actions (first 15–30 minutes) — reduce customer impact

• Triage & declare incident, assemble small cross-functional war room (on-call, deploy owner, messaging team, SRE).
• Implement hot mitigations to stop new failures: route new traffic away from suspect deployment (traffic split / feature flag off / scale to previous revision), pause rolling deploys, and enable aggressive retry/backoff on clients if safe.
• If broker/queue is overwhelmed, enable temporary rate-limiting or reject non-critical notifications to preserve core paths.
• Communicate externally: status page update and Slack/PagerDuty notifications with ETA for next update.

Data to gather (concurrent, automated where possible)

• Metrics: error rates (message-not-delivered), throughput, latency, queue lengths, consumer lag, resource utilization (CPU, memory, network) across services and brokers; compare to baseline 1–24h.
• Logs: clustered logs from notification service, push gateway, broker logs, consumer/worker logs; include request ids, trace ids, payload sizes, error stacks.
• Traces: distributed traces for representative failed flows to find where failures surface (enqueue vs delivery vs client ack).
• Deployment metadata: commit/artifact, config changes, feature flags, DB/schema migrations, config maps, container images.
• Broker/DB state: queue retention, POISON messages, backpressure indicators, disk/io saturation.
• Recent alerts, CI/CD pipeline events, and health-check failure timestamps.

Likely causes to investigate (prioritized)

1. Deployment regression: codebug, changed retry semantics or validation causing drops — check diff, rollback candidate.
2. Schema or API change: producer/consumer contract mismatch causing deserialization/validation failures.
3. Broker saturation or configuration change: retention, max inflight, connection limits, or partition imbalance causing rejects.
4. Memory pressure or OOM kills on workers/clients causing dropped messages or failed acknowledgments.
5. Configuration/secret change: auth failures to push gateways or 3rd-party providers.
6. Backpressure ripple: downstream slow consumers leading to queue bloat and timeouts upstream.

Triage workflow

• Reproduce small-scale: send test message, follow end-to-end trace.
• If rollback reduces errors in canary/old revision, promote rollback and continue investigation on new build.
• If broker saturation, scale brokers, tune config, or offload less critical traffic.

Post-incident remediation & preventive measures

• Root-cause analysis with timeline, contributing factors, and action items; publish postmortem within agreed SLA.
• Remediation examples: fix bug, revert schema change, tune broker configs, increase resource limits, fix memory leaks.
• Preventive controls:
  • Improve testing: contract tests for producer/consumer, integration tests exercising broker limits, chaos tests for backpressure.
  • Canary & progressive rollout policies with automated health gates tied to error-rate and queue metrics.
  • SLOs and error budgets update; automated runbooks mapping alerts to mitigations.
  • Observability: add dashboards, alerts for early signs (consumer lag, queue length growth, deserialization errors), more granular traces and sampling.
  • Automate safe rollback and circuit-breakers; implement client-side backoff and idempotency to avoid duplication/loss.
• Review runbooks and run a fire-drill to validate new mitigations.

Approach: maintain per-client priority queues and global memory tracked by an allocator. When memory > threshold, apply graduated policy: 1) refuse/ drop low-priority messages, 2) compress medium-priority messages (batch + compress), 3) aggregate/ coalesce low-volume clients, while always preserving high-priority messages. Emit metrics for drops/compressions and support graceful degradation (backoff hints, rate-limited reinstatement).

Pseudocode (Python-style):

# thresholds (bytes)
MEM_WARN = 0.7 * MEM_CAP
MEM_CRITICAL = 0.9 * MEM_CAP

# per-client structures
clients = { client_id: { "queues": { "high":[], "med":[], "low":[] }, "meta": {...} } }

def current_memory() -> int:
    return allocator.allocated_bytes()

def apply_backpressure():
    mem = current_memory()
    if mem < MEM_WARN:
        return  # healthy

    # Step 1: protect high priority
    preserve_high()

    # Step 2: graduated actions
    if mem >= MEM_CRITICAL:
        # aggressive: drop low, compress med, aggregate opportunistically
        drop_low_priority()
        compress_medium_priority(batch=True)
        aggregate_small_clients()
        emit_metric("backpressure.mode","critical")
    else:
        # moderate: only compress medium and delay low
        compress_medium_priority(batch=False)
        delay_low_priority_with_backoff()
        emit_metric("backpressure.mode","warn")

    record_metrics()

def drop_low_priority():
    for cid in clients:
        while memory_exceeds(MEM_CRITICAL) and clients[cid]["queues"]["low"]:
            msg = clients[cid]["queues"]["low"].pop(0)
            allocator.free(msg)
            emit_metric("messages.dropped", {"client":cid, "priority":"low"})

# compression and aggregation helpers omitted for brevity

Key concepts:

• Prioritized queues ensure high-priority unaffected.
• Actions are ordered: delay -> compress -> aggregate -> drop.
• Emit detailed metrics (dropped_count, compressed_bytes_saved, affected_clients).
• Graceful degradation: send backoff hint to clients, rate-limit recovery, and incremental rollbacks.

Rollout strategy & safety guards:

• Feature-flag rollout per region/cluster; start with internal customers.
• Canary small % of traffic, monitor error budget, drop-rate, latency.
• Kill-switch and automated rollback if >X% increase in high-priority latency or error-rate.
• Per-customer caps to avoid noisy-neighbor effects; never drop high-priority beyond a tiny emergency rate.
• Continuous visibility: dashboards and alerts for drop/compress events and SLO violations.

Edge cases:

• Sudden spike from single client → enforce per-client token bucket.
• Persistent memory leak → pair with OOM monitors and auto-scale/evict policy.