Replication, Partitioning, and Sharding Questions

Situation: We have one hot partition whose integer keyspace [L,R] is overloaded and must be split into s contiguous shards. Keys are mappable to a weight (requests or bytes) per small bucket; moving a key costs its weight. Goal: choose s−1 cut points so total weight moved across the cluster is minimized (i.e., reuse as much data on original nodes as possible).

Algorithmic approach (DP on weighted 1D partitioning)

1. Discretize the integer space into n contiguous buckets (choose granularity so n is manageable; histogram from telemetry).
2. Let w[i] be weight in bucket i and origin[i] be current node owning bucket i.
3. Define cost(a,b,node) = total weight in buckets [a..b] that are NOT currently on node (i.e., weight that must move if [a..b] assigned to node).
4. We want to partition 1..n into s segments and assign each segment to an existing node (or new node) to minimize sum cost of segments.
5. DP formulation: dp[t][j] = min cost to partition first j buckets into t segments. Transition:
   dp[t][j] = min_{i < j, node} (dp[t-1][i] + cost(i+1, j, node))
   Keep backpointers for reconstruction.
6. If assigning only to fresh new shards (no reuse), cost(i+1,j,new)=sum w[i+1..j]. If reusing nodes, allow node choices limited to nodes with significant overlap to reduce search.

Complexity:

• Precompute prefix sums for weights and per-node prefix sums: O(n * Nnodes).
• Evaluating cost(i+1,j,node) is O(1) with prefix sums.
• DP naive complexity O(s * n^2 * Nnode). With optimizations (Knuth/Yao conditions don't directly apply), reduce by:
  • Limiting candidate i via beam/pruning (only recent boundaries).
  • Only considering a small set of nodes per interval (top-k overlapping nodes).
    Practically with n1k, s10 this is fine.

Greedy / clustering alternative:

• Greedy: iteratively cut the heaviest subrange, assigning each cut to the node that preserves most weight. Fast O(n log n) but may be suboptimal.
• 1D k-means / k-median on weighted coordinates approximates cuts when moving to fresh shards.

Operational online steps

1. Measure: build histogram of request/traffic by bucket and current ownership map.
2. Compute split plan with DP (simulate cost, choose assignments).
3. Dry-run: simulate expected movement and load per target shard; ensure no node overload.
4. Migrate online with small consistent-range moves:
   • Create target shard containers and routing entries (shadow routing).
   • Throttle and stream data for each moved range (use changefeeds or binary logs).
   • Use a drain-and-cutover: replicate new shard, allow read from both, switch writes atomically (feature-flag/routing update).
   • Monitor latency / error budgets; rollback if SLA violated.
5. Post-migration: garbage-collect old data and update metadata.

Edge cases & notes

• Choose bucket size to trade accuracy vs compute.
• Hotspot within single bucket → refine granularity.
• Concurrent resharding: coordinate with global placement service to avoid conflicts.
• Safety: rate-limit migrations, perform in rolling fashion, and include circuit-breakers.

Why this works

• DP leverages 1D contiguous structure and per-node overlap to explicitly minimize moved weight.
• Practical optimizations and operational safeguards keep resharding online and low-risk.

Requirements & constraints first: throughput (ingest TPS), tenant count and size skew, query patterns (recent reads vs long-range analytics), storage limits, retention SLAs, and compaction CPU/I/O budget.

Partitioning tradeoffs

• Per-tenant partitions
  • Pros: isolation (hot tenants don't affect others), simple quota/backup/restore, easier per-tenant GC.
  • Cons: many small partitions if tenants are numerous → metadata overhead and inefficient compaction.
  • Best when tenant count is moderate and size-skew is high.
• Time-range partitions (e.g., hourly/daily)
  • Pros: predictable lifecycle for retention/eviction, efficient range queries by time, easy TTL by dropping whole partitions.
  • Cons: cross-tenant contention in same partition; harder tenant-level quotas.
  • Best when tenants are many and per-tenant data volume is small.
• Hybrid (tenant × time sharding)
  • Pros: combines isolation and time lifecycle; you can drop per-tenant time segments; limits partition explosion by bucketing (e.g., tenant groups or hashing).
  • Cons: more complex routing and metadata; choose granularity carefully (daily per-tenant for big tenants, weekly for small).

Retention & compaction strategy

• Retention: enforce multi-tier retention: hot (raw) for N days, warm (downsampled/aggregated) for M months, cold (highly compressed or external archive) for years. Automate transitions via partition lifecycle policies.
• Compaction: schedule background compaction during low-traffic windows, prioritize compaction for hot partitions but limit I/O using rate limiting. Use incremental compaction (L0→L1) and size-tiered policies to avoid write amplification.
• Downsampling: store rollups (min/avg/max, histograms) at fixed intervals; keep mapping so queries choose finest resolution available.

Query patterns

• Recent queries: route to hot partitions or ingest nodes with minimal compaction interference; use in-memory indexes, store recent N hours in memtables or tiered SSDs to ensure low latency.
• Historical queries/analytics: run on compacted, read-optimized partitions or OLAP replicas; use batch workers that can read compressed blocks and reconstruct downsampled results.

SRE operational considerations

• Monitoring/alerts: track partition count, partition size distribution, compaction backlog, GC throughput, write/read latencies, tenant hot-spot metrics.
• Autoscaling & throttling: autoscale ingestion frontends and compaction workers; implement tenant-level ingress throttles and backpressure to protect compaction.
• Storage reclaim: dropping partitions (time or tenant) should be atomic in metadata; reclaim storage asynchronously with prioritized deletion and observability. Validate tombstone propagation to avoid resurrected data.
• Failure modes: design compaction idempotency, checkpoint compaction progress, and provide rollback for accidental retention policy changes.
• Runbooks: define compaction emergency steps (pause ingestion, force compaction on affected shards, evict caches) and size-based alerts to trigger tenant remediation.

Recommendation: prefer hybrid with adaptive granularity (per-tenant daily for large tenants, time-only for small tenants), automated lifecycle moving data from hot->warm->cold, background rate-limited compaction, and robust monitoring + tenant throttles to maintain reliability.

Situation: We need a small cross-tenant report that joins user records and orders stored on different shards. As an SRE I’d evaluate three layers of options—application, middleware, and database—balancing performance, correctness, and operational complexity.

Approach summary:

• Application-level fan-out-then-join

  • Description: Query each shard (parallel HTTP/DB calls), fetch relevant user and order subsets, do the join in the app.
  • Performance: Good for small result sets; parallelism reduces latency but increases outbound connections and CPU/memory in app.
  • Correctness: Must handle partial failures, retries with idempotency, consistent snapshot via read timestamps if supported.
  • Trade-offs: Simple to implement, easy to instrument; expensive for larger datasets and harder to paginate across shards.

• Middleware / aggregator layer

  • Description: Introduce a dedicated aggregator service (or use an existing query router) that orchestrates shard queries, merges results, and exposes a single API.
  • Performance: Centralizes fan-out logic, enables connection pooling and throttling, caching, and batching.
  • Correctness: Aggregator can enforce global time-bounded reads, retries, and backpressure. Adds single point to monitor/scale.
  • Trade-offs: Extra component to deploy/operate; can be scaled horizontally and placed behind circuit breakers.

• Database-layer (cross-shard joins / global index)

  • Description: Use DB features: distributed SQL (e.g., Vitess/CockroachDB), global secondary index, or a materialized read-replica that co-locates join keys.
  • Performance: Best for low-latency joins; avoids pulling all data to app. But may require schema changes or expensive rebalancing.
  • Correctness: Can provide transactional guarantees depending on DB; easier to maintain strong consistency.
  • Trade-offs: Operational complexity, migrations, possible write amplification.

Caching and consistency strategies:

• Query result caching (TTL) for reports with acceptable staleness; use cache invalidation on user/order updates or time-based windowing.
• Read-through caches or materialized views updated asynchronously (event-driven): excellent read performance, eventual consistency; track inconsistency windows in SLOs.
• Use change data capture (CDC) to build a denormalized reporting store (Elasticsearch/OLAP) for fast joins; document RPO/RTO and freshness SLAs.

Operational considerations:

• Instrument per-shard latency, error rates, and fan-out amplification; create SLOs for report latency and freshness.
• Implement retries with exponential backoff and per-shard circuit breakers to avoid cascading failures.
• Limit parallelism and enforce request quotas to protect shards.
• Provide observability (traces showing per-shard calls), and fallbacks (partial results with warnings) when strict correctness is not possible.

Recommendation:

• For small ad-hoc reports: application-level parallel fan-out with strong monitoring and caching.
• For frequent reports or strict latency: build an aggregator + event-driven denormalized store (materialized view) to serve joins with acceptable freshness guarantees.
• For long-term scale and strict consistency: evaluate a DB-level distributed SQL or global index migration.

Situation/context: In a sharded DB, cross-shard traffic and cache churn reduce performance and can create hotspots during resharding/failover. As SRE, the goal is to place caches and design invalidation so reads are local, writes don’t poison other shards’ caches, and resharding/failover avoids thundering.

Proposal — cache placement (hierarchy)

• CDN: edge-cache for static assets and user-specific but mostly-read content (public profiles, images). Cache key includes version and coarse shard id when appropriate to avoid cross-shard misses.
• Application (in-process/L1): per-instance LRU small cache for ultra-hot short-lived objects (rate-limit to avoid staleness). Use for idempotent reads and to absorb bursts.
• Redis (L2, per-shard): colocate Redis replicas with each DB shard (same AZ/host pool). Each Redis serves keys for that shard only — keys include shard prefix. This keeps cache hits local and avoids cross-AZ DB hops.

Cache population / invalidation patterns

• Primary pattern: cache-aside for reads; application reads Redis-by-shard, fallbacks to DB shard, then writes cache with a TTL.
• Invalidation: use versioned keys (namespace version per logical object) rather than deletes to avoid races: on write, increment object version in shard-local metadata and write-through or publish invalidation.
• Use shard-local pub/sub: when a write occurs, the shard publishes invalidation/version bump on that shard’s channel so only local Redis/app instances invalidate or refresh.
• For cross-shard objects (joins), materialize denormalized copies per shard and keep them TTLed; update asynchronously with change-data-capture (CDC) to avoid synchronous cross-shard invalidation.

Resharding / failover mitigations

• During resharding, maintain a read-through proxy that maps keys to old/new shard and includes sticky routing; warm target shard caches proactively (bulk migrate hot keys) using usage metrics.
• Use a "grace" TTL policy: on move, retain old shard’s cache entries for a short grace period while populating new shard to prevent transient hotspots.
• Apply rate-limits and smooth backoffs on cache misses during topology change; use request coalescing (single-flight) to prevent stampedes.
• For failover, prefer promoting local Redis replicas; in multi-AZ failover, fail fast to routed healthy shard-local caches rather than falling back to remote DBs.

Operational practices

• Monitor per-shard cache hit/miss, latency, and invalidation rates; alert on rising cross-shard DB calls.
• Use synthetic load to identify hot keys; preemptively shard or introduce per-key throttles.
• Automate cache warmers and CDC-based replay for consistency post-reshard.
• Instrument metrics for invalidation lag and tombstone counts; tune TTLs/versions accordingly.

Why this works

• Shard-local Redis reduces cross-shard traffic and latency.
• Versioned keys + pub/sub localize invalidation and avoid race conditions.
• Graceful TTLs, warm-up, and coalescing prevent cache-driven hotspots during reshard/failover while keeping availability high.

Requirements (clarify): 200M global users, 5 regions, regional data residency, SLOs: Read 99.99% avail, Write 99.9% avail, <100ms regional read p95, friend-graph queries.

High-level architecture:

• Per-region read/write gateways fronted by regional API tier and an edge cache (CDN + regional in-memory cache like Redis).
• Global control plane for metadata (user-to-shard map) in a strongly-consistent store (e.g., Spanner/etcd) deployed per legal region needs.
• Data plane: sharded user graph stored in region-aware storage clusters (primary in-residency region), with asynchronous geo-replication for cross-region reads when allowed.

Replica topology & residency:

• Each user’s authoritative shard (primary) lives in the user's residency region (to satisfy residency/legal).
• Within-region: replica set of N=3 (1 primary, 2 sync replicas) for write safety and fast reads. Use quorum writes (majority) to meet 99.9% write availability.
• Cross-region: one async read-replica per other region for low-latency reads where residency permits; if residency forbids, remote replicas are not created and reads route to primary or cached data.
• Storage choices: graph-optimized store (e.g., Titan/JanusGraph on Cassandra/Scylla) or a relational store with adjacency lists and denormalized friend-materialized views for common queries.

Sharding strategy:

• Hybrid: user-id hash sharding for even distribution, but colocate high-degree users (supernodes) specially: detect hot users and isolate them onto dedicated shards (vertical split). Partition by (region, shard-id).
• Range or consistent-hash to allow re-sharding. Store user->shard mapping in metadata service.

Friend-graph queries / cross-shard approach:

• For one-hop friends: keep adjacency lists in the user's primary shard and maintain a regional cache of friend lists (LRU + TTL). For mutual-friend/2-hop or heavy traversals:
  • Use scatter-gather within-region first (fan-out to shards that hold friend adjacency), with parallel RPCs and timeout budget.
  • For cross-region traversals, prefer eventual answers: consult regional async replicas + cached precomputed mutual-friend indexes for common queries.
  • Precompute and store heavy/expensive queries (mutual friends, recommendations) in a nearline job (Beam/Spark) and materialize per-region for p95 latency.
  • For supernodes, use targeted denormalization (store partial friend-of-friend indices) to avoid fan-out storms.

Consistency & latency trade-offs:

• Strong consistency for writes to primary + sync replicas in-region (quorum). Reads served from local sync replica when possible (read-after-write within region by redirecting to primary or using session tokens).
• Allow bounded staleness for regional reads from async replicas or caches to meet latency SLO.

Operational: SRE monitoring & SLO enforcement

• Define SLOs and error budgets per region and per API (read latency p95 <100ms, read availability 99.99%, write 99.9%).
• Observability:
  • Metrics: per-shard RPC latency, errors, QPS, tail latencies, replication lag, cache hit ratio, circuit-breaker tripping, CPU/memory, disk IOPS.
  • Distributed tracing for cross-shard queries to measure fan-out fan-in times.
  • Synthetic transactions per region simulating reads/writes and friend-graph queries.
  • Alerts: multi-tier (page for SLO burn >x% in 1h, runbook-trigger for replication lag > threshold).
• Automation & mitigation:
  • Auto-scaling groups for DB frontends and read-replicas; autoscale read cache capacity.
  • Circuit breakers and backpressure: limit fan-out concurrency, fallback to cached/approximate results, queue requests during hotspots.
  • Canary deployments, automated rollback if SLOs are violated.
• Error budget policy:
  • Track burn rate per region; if burn exceeds threshold, engage corrective policy: freeze non-essential releases, scale resources, reduce background jobs (throttling precompute pipelines), enable stricter caching to reduce load.
• Incident response & postmortem:
  • Runbooks for replication lag, hotspotting, region outage. Post-incident blameless reviews and capacity planning updates.
• Capacity & testing:
  • Chaos engineering for region failover, replica loss, network partition tests.
  • Regular re-sharding rehearsals and warm-up of caches after failovers.

Edge cases & trade-offs:

• Tradeoff between strict residency and global performance — where residency prohibits replication, rely on edge caches, higher read latency possible; mitigate via precompute and client-aware UX.
• Supernode handling required to avoid fan-out storms; use rate limits and denormalization.

This design balances regional residency, low-latency reads, high availability, and SRE-driven observability and control to enforce SLOs and manage error budgets.