Scalability Patterns and Techniques Questions

Direct answer

A pragmatic decomposition starts from real bounded contexts, not code modules, keeps data integrity intact by translating between the old and new models instead of cutting over all at once, and replaces any temptation toward a distributed two-phase commit with sagas built from idempotent, compensable steps. The strangler fig pattern, routing traffic through a facade so old and new implementations coexist during the transition, is the mechanism that makes all of this incremental and reversible rather than a risky big-bang cutover.

Identifying service boundaries

Use domain-driven design to find bounded contexts along business-capability lines (orders, billing, inventory) rather than along existing code-module lines, which usually reflect implementation history more than the current business domain. Within a candidate boundary, look for a vertical slice: API (application programming interface), business logic, and the data that logic owns, so the resulting service can operate independently without still calling back into the monolith for its own core data. Start with a boundary that is both low-risk (not on the critical write path) and clearly bounded (its data isn't heavily shared with other domains), since that combination validates the migration mechanics without betting the business on the first attempt.

Data integrity during migration

• Anti-corruption layer (ACL): a translation layer between the monolith's data model and the new service's API, so the new service isn't forced to adopt the monolith's legacy shape, and the monolith doesn't need to understand the new service's internals.
• Change data capture (CDC): stream changes from the monolith's database into the new service so reads stay consistent while write ownership is still transitioning.
• Dual writes, done carefully: if the monolith and the new service both need to write during the transition, make every write idempotent and reconcile the two stores regularly; prefer event-driven replication (the monolith publishes a change event that the new service consumes) over synchronous dual-writes, which fail if either side is briefly unavailable.

Avoiding distributed-transaction anti-patterns

Don't reach for two-phase commit (2PC) across the new service boundary. It requires every participant to be available and responsive during the commit window, which directly works against the availability and independent-deployability goals that motivated decomposition in the first place. Instead, model multi-step business flows as sagas: a sequence of local transactions, each with a defined compensating action if a later step fails, so the overall flow degrades gracefully instead of blocking on a distributed lock.

A common practical split is to orchestrate the core, high-stakes transaction (say, order to payment to shipment) where compensations matter, and let secondary, lower-stakes integrations (send a confirmation email, update an analytics feed) stay choreographed.

When to reach for the strangler fig pattern

Reach for it whenever you need the old and new implementations to coexist safely during a transition, which in practice is almost always for anything beyond a trivial, low-traffic domain. The mechanics, applied specifically at a monolith's gateway:

• Reverse-proxy routing: put a reverse proxy or API gateway in front of both the monolith and the new service, and route requests to one or the other by path, feature flag, or percentage, so the switch is a routing change rather than a deployment event.
• Traffic shadowing: mirror a copy of live production requests to the new service without using its response, so you can compare its behavior and performance against the monolith's real response under real traffic before the new service is trusted to actually serve anyone.

flowchart LR
  C[Client] --> RP[Reverse proxy /<br/>API gateway]
  RP -->|serves response| M[Monolith]
  RP -.->|mirrored, response discarded| NS[New service]
  RP -->|once validated,<br/>cut over by %| NS

Only retire the monolith's code path for a given slice once traffic has been fully cut over and the data behind it has been reconciled, not merely once the new service has been deployed.

Worked example: extracting a payments capability

Define the Payments bounded context and its API contract, then stand up an ACL that maps the monolith's existing payment calls onto that contract. Use CDC to populate the new Payments service's data store from the monolith's database so both stay consistent during the transition. Put a reverse proxy in front of the payments endpoints and start with traffic shadowing: every real payment request is also sent to the new service, its result compared against the monolith's, but only the monolith's response reaches the client. During this shadow period, downstream systems the payments path calls (a fraud check, a ledger write) effectively see close to double their steady-state call volume, since both the monolith and the shadowed new service are calling them, which is worth flagging to whoever owns that downstream capacity before shadowing begins. Once results match consistently, cut real traffic over gradually (a small percentage, then a majority, then all of it) via the reverse proxy, with the monolith's code path kept in place and able to take traffic back at any stage until the migration is fully reconciled.

Trade-offs and pitfalls

• Treating dual writes as sufficient data-integrity strategy on their own, without idempotency keys or a reconciliation job; a single missed or duplicated write during the transition window leaves the two stores silently inconsistent.
• Reaching for orchestration everywhere out of caution; it adds a coordinator every downstream service must trust and depend on, which is unnecessary overhead for flows that don't actually need centralized compensation logic.
• Shadowing traffic without accounting for its load on shared downstream dependencies, which can turn a validation exercise into a self-inflicted capacity incident.
• Declaring the migration done once the new service is receiving 100% of traffic, while the monolith's old code path and data are still live as a fallback; the migration isn't actually finished, and the risk isn't actually retired, until that fallback is deliberately removed.

Direct answer

Resharding a live database with minimal downtime is an online migration: keep the old topology serving traffic while you copy data to the new one under a throttle, use dual writes or change-data-capture (CDC) to keep the copy current, validate it against the source before trusting it, and cut routing over in a small, reversible step rather than a single big-bang switch. The hard parts are rarely the copy itself; they are choosing a shard key/count that won't need redoing soon, keeping writes ordered correctly during the overlap window, and having a rollback that is actually exercised, not just documented.

Structured elaboration

Choosing the new shard key or shard count

• Analyze real access patterns (write hotspots, cardinality, range-vs-hash trade-offs) from production telemetry rather than guessing; target per-shard headroom (commonly keeping any shard well under its saturation point) so the new topology doesn't need revisiting immediately.
• For a hot single shard suffering write contention, the two concrete techniques are rekeying (choosing a shard key that spreads the contended writes, if the current key is the root cause) and range-splitting (dividing the shard's existing key range into two or more ranges, each becoming its own shard, when the key itself is fine but the range has outgrown one node).
• For a lookup-based versus algorithmic shard-mapping layer: an algorithmic mapping (consistent hashing, which places nodes and keys as points on a circular hash space called a ring, so a topology change only reassigns the small slice of keys near it, or the simpler hash-mod-N) needs no metadata lookup and scales trivially, but it distributes tenants roughly evenly by key, which fails badly when tenant sizes vary by orders of magnitude (a large enterprise tenant next to hundreds of tiny ones). A lookup-based mapping (an explicit tenant-to-shard table) costs a metadata hop per request but lets you place large tenants on dedicated shards and pack small tenants together deliberately. For a multi-tenant system with that kind of size skew, prefer lookup-based mapping, or a hybrid: algorithmic mapping for the default pool of ordinary tenants, with a lookup override for the small number of outsized ones.

Data-migration strategy

• Online copy under CDC: stream changes from the source shard(s) via a CDC pipeline (e.g., a log-based change stream) into the target topology while a background job bulk-copies existing data in ranges. The CDC stream and the bulk copy have to be reconciled (apply CDC events that land inside an already-copied range, buffer or replay ones that land ahead of the copy cursor).
• Dual writes with a quiesce window: to guarantee transaction ordering is preserved for records that are actively being migrated, briefly pause (quiesce) writes to the specific key range being cut over, drain the in-flight CDC backlog for that range so source and target agree, flip routing for just that range, then resume writes. This bounds the ordering risk to a small, deliberately-chosen window instead of trying to reason about ordering across an open-ended dual-write period for the whole dataset.
• Split-and-merge at scale: for a large, contended shard, resharding is a split (one shard's range divides into two, each getting its own routing-table entry) rather than a full-dataset migration; the reverse operation, merge, combines two under-loaded shards back into one when growth was over-provisioned. At terabyte scale, both operations are driven by range boundaries and routing-table updates rather than moving the whole dataset, which keeps a single split or merge bounded in time and blast radius.
• A cross-topology special case: migrating a stateful store (for example, a session store) from a single-region deployment to a multi-region sharded cluster follows the same dual-write skeleton, but adds a proxy layer in front of both topologies so the application never talks to the store directly during the migration; the proxy performs parity verification (comparing responses from the old and new paths for the same key) before fully cutting traffic over, which catches migration bugs before they're user-visible.

flowchart TD
    A[Old shard topology] -->|CDC stream| B[Change pipeline]
    A -->|throttled bulk copy| C[New shard topology]
    B --> C
    C --> D[Shadow reads: compare old vs new]
    D -->|parity confirmed| E[Routing cutover, range by range]
    D -->|mismatch| F[Pause cutover, reconcile]
    E --> G[Quiesce window per range: drain, flip, resume]
    G --> H[Old topology retired]

Routing updates
Introduce a migration-aware routing layer (gateway or client-side routing table) so application code never encodes shard topology directly. Roll the new routing out in stages: internal/test traffic, a small canary of production traffic, then full cutover, with each stage able to fall back to the previous routing table without a code deploy.

Throttling
Throttle the copy and CDC-apply workers adaptively against replication lag, tail latency, and source-shard CPU/I/O, not a fixed rate, so migration traffic backs off automatically when it's competing with production load. As a worked illustration of sizing a per-shard step against an operational target: suppose the team's service-level agreement (SLA) is for each per-shard rebalance step to complete in well under two minutes even across a 200+ node cluster, and a given shard holds roughly 50 GB of data with the copier capped at 500 MB/s per shard to protect production I/O. The data-transfer portion alone takes

which leaves about 20 seconds of headroom under the 120-second target for updating ring/routing metadata and completing leader election (the process by which the new shard's replica-set nodes agree on which one of them becomes the active primary/writer) for the new shard's replica set, before the step is considered done. That headroom is the number to watch: if metadata propagation or leader election routinely eats into it, the fix is a faster metadata path, not a bigger copy-bandwidth cap.

Automating rebalancing across many shards
When rebalancing hundreds of shards simultaneously rather than one at a time, the automation has to cap total concurrent migrations (global throttle, not just per-shard), and it must keep secondary-index consistency in scope explicitly: an index that's rebuilt or re-pointed on a different schedule than its base table will silently return wrong results for the window in between, so index cutover needs to be part of the same per-range quiesce step as the base data, not a follow-up job.

Validation

• Row counts and per-range checksums between source and target.
• Shadow reads: serve a fraction of real production reads from the new topology in parallel with the old one (without returning the shadow result to the client) and compare, which surfaces correctness bugs under real traffic patterns before any client depends on the new path.
• Client-side cache invalidation during cutover: if clients or edge caches hold entries keyed by the old routing, cutting a range over without invalidating those cached entries serves stale-routed responses; the cutover step has to include a cache-invalidation signal (a version bump on the routing key, or explicit invalidation) for anything caching downstream of the router.
• End-to-end application-level smoke tests against the new topology before removing fallback to the old one.

Rollback
Keep dual writes (or the CDC stream) active until validation passes; if a problem surfaces, stop applying new CDC events, revert routing to the old topology for the affected range, and resume from the last confirmed-consistent checkpoint rather than restarting the whole migration. Because cutover happens range-by-range behind a quiesce window, rollback scope is bounded to the ranges already flipped, not the whole dataset.

Trade-offs & pitfalls

• A migration that skips the quiesce window to avoid any write pause will have an ordering gap for records touched during the exact moment of cutover; a short, well-instrumented pause on a narrow key range is a better trade than an unbounded consistency risk.
• Algorithmic shard mapping is simpler to build and reason about, but for multi-tenant systems with orders-of-magnitude tenant-size variance, defaulting to it produces shards that are balanced by tenant count and wildly unbalanced by load; verify the mapping choice against actual per-tenant load, not tenant count.
• Skipping shadow reads to save migration time is the single most common way correctness bugs escape into production, because checksums catch data-copy errors but not query-behavior differences (an index that returns results in a different order, a range boundary edge case).
• Treating secondary-index rebuild as a background job that can lag the base-table cutover is a frequent source of silent incorrect reads immediately after a migration; keep index and base-data cutover atomic per range.

Direct answer

Use the temporary read replica as the mechanism that turns an in-place resize (which can mean real downtime) into a controlled cutover: provision the replica already at the larger CPU/memory spec, let it fully catch up to the primary, briefly pause writes, promote the replica to primary, and repoint the application. The actual write-unavailability window is bounded by how long it takes to drain in-flight writes and flip the connection target, not by the resize operation itself, which is why this pattern minimizes downtime even though it isn't strictly zero-downtime.

Structured elaboration

Pre-checks

• Confirm the exact target CPU/memory spec against measured load, not a guess, and confirm a maintenance window and a communicated service level objective (SLO, the measurable target for allowed downtime or latency impact) for the operation.
• Take a fresh on-demand snapshot immediately before starting, independent of the replica strategy, as a last-resort fallback.
• Verify current replication lag baseline and that the environment supports creating a same-region replica sized larger than the current primary.
• Confirm the automation (scripts, IaC) for promotion and connection-string cutover has been tested outside of this incident, not written live.

Rolling steps

1. Create a read replica provisioned at the new, larger instance spec. Let it catch up and monitor replication lag until it's negligible and stays that way for a sustained period, not just a single low reading.
2. Briefly quiesce writes on the current primary (put the application into a short read-only or write-paused mode).
3. Promote the replica to primary. Because it was fully caught up at the moment of promotion, this preserves the data that existed at quiesce time.
4. Repoint the application's write target to the newly-promoted primary (a connection-string or routing change, ideally something that doesn't require an application redeploy).
5. Resume writes and run smoke tests against critical read and write paths.
6. Rebuild redundancy: the original (smaller) instance can be resized and re-added as a replica, or replaced, restoring the topology's normal read-replica count.

Fallback options

• If the replica fails to catch up before the maintenance window closes, abort the promotion; investigate whether replication is network- or I/O-bound, and either wait for a longer window or address the bottleneck before retrying.
• If a data mismatch or unexpected inconsistency is detected after promotion, the fallback is the pre-operation snapshot, not the old primary (which may now be behind); restore from snapshot to a fresh instance if this happens.

Monitoring

• Before: replication lag trend over a real observation window, not a single point-in-time check, plus baseline CPU/memory/connection counts to compare against post-cutover.
• During: replication lag right up to the promotion moment, since promoting a replica that's meaningfully behind means losing whatever writes happened after its last applied transaction.
• After: error rates, write and read latency, and a targeted data check (row counts or checksums on a few critical tables, or confirming the most recent known transactions are present) rather than assuming success from the absence of alarms.

Worked example

A safe promotion policy might require replication lag to stay under 1 second for a sustained 5-minute window before promotion is allowed (a stated operational threshold for this runbook, not a universal rule that applies to every workload). The actual write-pause duration during cutover isn't a fixed number worth quoting as a general fact, since it depends on the application's connection pool behavior, not on the database resize itself: it's bounded by how long the app takes to drain in-flight writes and how quickly its clients reconnect to the new endpoint after the connection target flips, which is exactly why this pattern is described as minimizing downtime, not eliminating it. If the application's reconnect and retry logic is slow or missing, the same database-side runbook produces a much longer perceived outage even though the database steps themselves didn't change.

Trade-offs & pitfalls

• This is not truly zero-downtime: promoting a replica that isn't fully caught up loses whatever writes landed after its last applied transaction, so the safety of the whole procedure hinges on verifying lag is genuinely near zero at the moment of promotion, not assuming it.
• If the application cannot tolerate even a brief write-pause, this pattern isn't sufficient on its own; a true zero-downtime requirement needs a different approach, such as a proxy layer that queues writes during cutover.
• Before building a custom replica-promotion runbook, check whether the cloud provider's native "modify instance class" operation already performs an equivalent internal promote-and-swap; if it does, it may be simpler and better-tested than a hand-rolled version of the same idea.
• The replica-promotion mechanics here (catching up, promoting, cutting over) are a practical means to a scaling end; the deeper mechanics of replication modes and failover consensus are a related but distinct topic from the scaling procedure itself.

Direct answer

Cache hit ratio is the fraction of lookups a cache can answer without going to the origin (hits divided by total lookups); a cache miss is a lookup where the value isn't present or has expired, forcing a fetch from the origin and usually a write back into the cache; cache warmup is the process of populating a cache, at startup or after a restart, before it can offer a useful hit ratio. Whether adding an application cache (memcached, Redis, or similar) is worth it comes down to a simple cost comparison: what a miss costs you (a database round trip, a slow computation, an external API call) versus what running the cache costs you (infrastructure, and the risk of serving stale data), not a fixed hit-ratio threshold you're supposed to hit.

Structured elaboration

When to add a cache

• The workload is read-heavy with requests that repeat: the same or similar queries recur often enough that a cached answer is reused rather than computed once and discarded.
• The thing being cached is expensive relative to a cache lookup: a slow database query, an external API call, or a CPU-heavy computation are all good candidates; something already fast to compute gains little.
• The data can tolerate the staleness a cache implies, or the system can actively invalidate the cache on writes; if every read must reflect the absolute latest write, caching adds a consistency problem you have to solve, not just a performance win.

What hit ratio would justify the cost

There's no universal number, because "justified" depends on the ratio between the cost avoided per hit and the cost of running the cache, not on the hit ratio in isolation. A commonly cited planning heuristic for latency- and cost-sensitive systems is to aim for roughly 70 to 80% hit ratio; treat that as a starting heuristic to validate against your own workload, not as a target derived from anything specific to your system. A system where each miss is very expensive (an external, rate-limited, or metered API call) can be worth caching even at a 50% hit ratio, while a system where the origin call is already cheap may not justify caching even at 90%.

Three practical ways to improve an existing cache's effectiveness

1. Tune time-to-live (TTL, how long a cached value is considered valid before it must be refreshed) per key type rather than using one global value: longer TTLs for stable data, shorter for volatile data, so you're not needlessly re-fetching stable data or serving stale volatile data.
2. Improve cache key design: normalize keys (strip user-specific noise that doesn't actually change the result, use consistent prefixes) so semantically identical requests share a cache entry instead of each generating its own miss.
3. Warm proactively and avoid a thundering herd on expiry: pre-populate known-hot keys at deploy or restart, and use a read-through or refresh-ahead pattern where a background job refreshes a key just before it expires, rather than letting many concurrent requests all miss on the same expired key at once.

Worked example

Assume, as a planning input rather than a measured fact, that an endpoint currently receives 10,000 requests per minute with no caching, and that each request costs 20 ms of database time. Total database time consumed per minute today:

$10{,}000 \times 20\,\text{ms} = 200{,}000\,\text{ms} = 200\,\text{s of database time per minute}$

If a cache reaches a 75% hit ratio on this endpoint, the number of requests still reaching the database is:

$10{,}000 \times (1 - 0.75) = 2{,}500 \text{ requests per minute}$

$2{,}500 \times 20\,\text{ms} = 50{,}000\,\text{ms} = 50\,\text{s of database time per minute}$

That's a 75% reduction in database load, directly tracking the hit ratio, from 200 seconds of database time per minute down to 50. Whether that reduction is "worth it" then depends on what those 150 seconds of freed-up database time are worth relative to running the cache, not on the 75% figure by itself.

Trade-offs & pitfalls

• Chasing a higher hit ratio as a goal in itself, rather than as a proxy for reduced backend cost, can lead to caching things that barely help (already-cheap lookups) while a genuinely expensive but less frequent lookup goes uncached.
• A TTL that's too long trades staleness risk for hit ratio; too short and the cache behaves closer to no cache at all under the same access pattern.
• Skewed (heavily concentrated, or Zipfian) access patterns mean a handful of keys drive most hits; eviction policy choice (LRU, least-recently-used, versus LFU, least-frequently-used) matters more under this kind of skew than under uniform access, since LRU can evict a very-frequently-used-but-not-most-recent key that LFU would keep.
• Expose hit and miss counters through your application performance monitoring (APM, application performance monitoring) or metrics stack; without that visibility, a regression in hit ratio after a code change (a key-format change, a new high-cardinality parameter) can go unnoticed until backend load spikes.

Direct answer

At 500,000 events per second, the application layer runs into four coupled bottlenecks: network ingress and connection handling, the thread/concurrency model that turns incoming bytes into work, CPU spent parsing and enriching each event, and coordination overhead from any stateful lookup the enrichment step needs. The general fix is the same shape for all four: partition the work so no single node or lock sees the full 500k/s, move blocking calls off the hot path with batching and async I/O, and keep enrichment state local wherever possible instead of doing a synchronous round trip per event.

Where each bottleneck shows up, and how to mitigate it

Network and connection handling. Producers opening and closing connections per request, or a single ingress tier absorbing all 500k/s, saturates NIC (network interface card) throughput and connection-handling overhead before CPU (central processing unit) capacity is even the limiter. Mitigate by terminating connections on a horizontally scaled ingress tier with persistent, multiplexed connections (HTTP, Hypertext Transfer Protocol, version 2, or gRPC, a binary remote-procedure-call protocol, with keep-alive) instead of one connection per event, and by partitioning producers across ingress nodes (consistent hashing on producer or tenant ID: placing producers and nodes on a shared hash ring so adding or removing a node only reassigns a small slice of producers) so no single node absorbs the full rate.

Thread / concurrency model. A thread-per-request model at this rate means either an enormous number of OS (operating system) threads (high context-switch and memory overhead) or a bounded pool that queues and adds latency. Mitigate with an event-driven or async runtime (a reactor model, or a language runtime with lightweight concurrency) so a small number of OS threads can hold many events in flight, and keep the network-handling threads free of blocking work by handing enrichment off to a separate async stage.

Parsing and enrichment CPU. JSON (JavaScript Object Notation, a common text-based data format) parsing, regex, and per-event enrichment logic are genuine CPU cost that scales linearly with event count. Mitigate by using a compact binary wire format instead of text JSON where producers can be changed, batching events so parsing overhead is amortized across a group instead of paid per event, and partitioning the CPU work itself across many consumer processes (one partition of the input stream per consumer) so total CPU scales horizontally with node count.

Coordination for stateful enrichment. If enrichment needs a lookup (a reference table, a feature value, a dedup check), a synchronous call per event to a shared store turns 500k/s of ingestion into 500k/s of round trips to that store, which will bottleneck long before ingestion does. Mitigate with a local, sharded cache per consumer (so most lookups are in-process), asynchronous batched refresh of that cache instead of per-event synchronous calls, and partitioning by the same key the enrichment lookup uses, so a given consumer's cache stays warm for the keys it actually sees.

Ingestion pipeline shape

flowchart LR
  P[Producers] --> ING[Ingress tier<br/>persistent connections]
  ING --> PART[Partitioner<br/>consistent hashing]
  PART --> LOG[Partitioned log<br/>many partitions]
  LOG --> CONS[Consumer pool<br/>one partition per consumer]
  CONS --> CACHE[Local sharded cache]
  CACHE -->|miss: async batched refresh| STORE[Reference store]
  CONS --> OUT[Enriched events out]

Partitioning threads through the whole path (ingress, log, consumers, cache) is what lets total throughput scale by adding nodes rather than by making any single node faster.

Worked example: why thread-per-event breaks down

As an illustrative assumption, not a measured figure, suppose the combined parsing-plus-enrichment latency budget per event is 5 ms end to end. By Little's Law, the average number of events that must be in flight simultaneously to sustain 500,000 events per second at that latency is:

$N = X \times R = 500{,}000 \text{ events/s} \times 0.005\text{s} = 2{,}500 \text{ concurrent events}$

If handled with one OS thread per in-flight event, and again as an illustrative assumption, suppose each thread reserves roughly 2 MB of stack:

$2{,}500 \text{ threads} \times 2\text{ MB/thread} = 5{,}000 \text{ MB} \approx 4.9 \text{ GiB just in thread stacks}$

That is memory spent before any actual event data or enrichment state is held, and it scales linearly with load: doubling throughput doubles the stack overhead in this model. This is the concrete reason to move to an async or lightweight-concurrency model instead of one OS thread per in-flight event: the same 2,500-way concurrency can be held with a small, fixed pool of OS threads multiplexing many logical tasks.

Trade-offs and pitfalls

• Synchronous enrichment gives the simplest read-after-write behavior (the enriched event is definitely enriched before you acknowledge it) but caps throughput at whatever the slowest coordination call allows. Asynchronous enrichment (acknowledge ingestion fast, enrich in a downstream consumer stage) scales further but means the ingestion acknowledgment and the enrichment are no longer atomic, so downstream consumers must tolerate a brief window before enrichment completes.
• Batching parsing and enrichment reduces per-event overhead but trades off latency: a bigger batch amortizes more CPU cost per event but delays every event in the batch until it fills or a timer fires.
• A local cache per consumer reduces coordination cost but only if the partitioning scheme actually routes related events to the same consumer consistently; if the partition key doesn't align with the enrichment lookup key, the cache stays cold and you are back to a lookup per event.
• Sizing thread pools or partition counts once at launch and never revisiting them; hot partitions and CPU-bound stages should be monitored and rebalanced as traffic shape changes, not fixed at initial capacity.