Distributed storage

Partitioning assigns different portions of a dataset or key space to different storage or computation groups. The objective is often to spread load and storage growth, reduce contention, and place data near users. A partition key controls distribution; a replica count controls copies of an assigned partition. These dimensions are related but not interchangeable. Range partitioning groups adjacent ordered keys and supports range scans, but monotonically increasing keys can create a hot last range. Hash partitioning spreads keys more evenly when the hash function and key distribution cooperate, but destroys natural locality for ordered queries. Consistent hashing can reduce key movement when nodes are added or removed, while introducing virtual nodes and ownership metadata. None of these schemes automatically solves skew from a single exceptionally hot key.

For a university LMS storing submissions, partitioning by course ID seems natural for course-level queries. Yet one popular introductory course might account for half of peak uploads, producing a hotspot even with hundreds of otherwise idle shards. Adding more partitions helps only if the hot workload can itself be split or distributed. A composite key such as course ID plus randomized sub-bucket may spread writes but complicates aggregation and uniqueness checks. A range-based approach could instead isolate popular courses into dedicated shards. Assess expected read patterns, writes per key, storage volume, secondary-index overhead, and movement cost before selecting a key. A hash ring that keeps ownership stable under scale-out cannot protect the system from a single logical item being updated a million times per second.

Rebalancing is a distributed migration protocol. A source shard may continue accepting updates while target copies stream data, producing a changing boundary. The migration requires a snapshot or checkpoint, a catch-up stream, a cutover authority, and a recovery rule for interruptions. Blindly switching routing at the beginning can lose writes that are still being copied. Keeping two authoritative writers forever may violate uniqueness constraints. One strategy uses a consistent ownership epoch: source handles authoritative updates until target has caught up and an agreed cutover becomes active. Clients carrying stale routing information can be redirected or receive retryable errors. A well-documented migration states what happens to in-flight transactions and how old ownership is fenced after the switch.

Partitioning also fragments operational views. An aggregate query across shards may need scatter-gather, increasing latency and amplifying tail responses. Distributed joins and global uniqueness checks can require coordination. A request touching several shards may fail partially, even when each shard is locally reliable. Monitoring only the cluster-wide average misses hotspots; inspect p99 latency and queue depth by shard, key distribution, and movement backlog. Scale-out proposals should include a realistic estimate of metadata, repair traffic, index space, and compaction overhead. When designing a partitioned system, begin with actual access patterns and failure consequences rather than copying a default shard count from a tutorial.

Create a shard-placement simulator with configurable keys, node count, popularity skew, and ownership strategies. Visualize the keys assigned by a simple modulo hash and by a virtual-node ring, then add a node and measure reassigned keys. Count requests per shard from a seeded Zipf-like popularity distribution. Challenge learners to fix an overloaded shard without simply increasing node count, perhaps by isolating a hot key or distributing a commutative counter. Then simulate cutover: keep a routing epoch and show that an old router must not send new authoritative writes to a demoted source. The toy model omits real secondary-index costs and disk storage behavior; those require deployment-specific measurement.

A transaction groups logical actions into a contract about atomicity and isolation. Atomicity means the relevant state transitions either take effect as a unit or do not; isolation defines how simultaneous transactions can observe or interfere with each other. When data is partitioned, a transaction touching two shards needs coordination across potentially failing processes. Two-phase commit is an atomic commit protocol: a coordinator asks participants to prepare, participants durably record whether they can commit, and the coordinator then directs a shared final decision. A prepared participant may hold locks and be unable to unilaterally abort after a coordinator failure, because the commit decision could already have been durably made elsewhere. Thus classical 2PC can block despite preserving atomic decision semantics under its model. Two-phase commit alone is not a substitute for consensus about the coordinator's state.

Consider transferring a scholarship allocation between departmental ledgers. Shard A must decrease its available funds by 500 units and Shard B increase by 500. If A commits and B later fails, the total is temporarily wrong unless a protocol ensures a coordinated outcome. In 2PC, both shards prepare their changes and persist intent; only after both report readiness does the coordinator record and broadcast commit. If B cannot prepare, both abort. A failure after preparation but before learning the decision leaves an in-doubt state. Operators must know how prepared transactions are recovered and how long locks are allowed to remain. A high-level diagram that skips forced log records or crash recovery does not explain the actual safety of the protocol.

Atomicity and serializability remain distinct. A transaction could atomically commit changes at two shards yet still suffer a write skew anomaly if each checked a snapshot of the other's state under insufficient isolation. Suppose two doctors independently see that the other is on call and each goes off duty; both transactions can commit atomically, leaving no doctor available. Preventing the anomaly needs a suitable isolation or explicit invariant mechanism. Distributed transactions often choose between stronger isolation with coordination and weaker isolation with application-level conflict handling. The decision should reference concrete invariants, latency budgets, expected contention, and recovery objectives, not an abstract preference for 'ACID versus BASE'.

A saga decomposes a long business workflow into smaller local transactions and compensating actions. This can improve availability and reduce lock duration, but compensation is not a magical reversal of physical effects. A shipment that already left a warehouse cannot be undone by setting a database row back to 'not shipped'; the system may need a return procedure. A notification email cannot be recalled reliably. Compensation rules must be designed with the product and may require human intervention. Distinguish an atomic database transaction from an eventually coordinated business process whose intermediate steps can be visible. The user experience should show 'pending', 'failed and recovering', or 'completed' honestly, rather than always declaring immediate success.

Build a 2PC timeline with coordinator, shard A, and shard B. Controls trigger prepare, durable-ready recording, commit decision, failure after prepare, and recovery. Shade transaction state separately from availability of each node. The learner should find a trace where participants are blocked awaiting the coordinator yet no conflicting commit occurs. In another mode, run a saga: book a venue, reserve equipment, and issue a refund on failure; show why compensation is a new action, not time travel. Include a concurrent write-skew example to demonstrate that atomic commit does not by itself guarantee serializability. Ask students to label exactly which properties are achieved by each mechanism and what persisted records are necessary for restart safety.

Recovery reconstructs a valid operational state after process loss, disk interruption, human error, or data corruption. A write-ahead log stores enough information durably before applying changes so that a restart can redo or undo actions according to the storage engine's protocol. Replicated logs coordinate an accepted command order across nodes; database WAL primarily supports local atomicity and recovery, though implementations can combine the two. Snapshots bound replay time by capturing a consistent state at a known log position. A snapshot must be verified, versioned, and tied to the corresponding durable history. A copy of memory without a coherent checkpoint may mix updates from incompatible moments. Recovery procedures should document which artifacts are authoritative and how each is validated before serving client traffic.

Recovery design balances runtime overhead against restoration latency. Frequent snapshots consume I/O and storage bandwidth, while infrequent snapshots require longer log replay after crash. Incremental snapshots reduce repeated copying but create dependency chains; losing a necessary base invalidates later increments. A log segment can only be safely deleted after the relevant state and retention guarantees are preserved. If a follower is offline longer than log retention, the leader may need to install a full snapshot before resuming replication. Restoring a snapshot also requires restoring metadata such as applied index, configuration epoch, and deduplication state. Failing to preserve client operation IDs can allow an old retry to execute a committed financial action twice after recovery.

Imagine a distributed lab notebook whose primary disk fails after acknowledging a sequence of edits. A replica has the committed log up to index 900, and its latest snapshot covers index 850. It can restore the snapshot and replay committed entries 851–900. It must not automatically replay speculative entries at 901–905 from an abandoned leader as confirmed edits. A checksum mismatch on one log block should trigger a recovery branch, not silent acceptance. In a multi-shard notebook, snapshotting every shard at different moments may not produce a globally consistent read unless the application has a snapshot protocol or causal barrier. This is particularly important for audit trails that must explain cross-service events after a security incident.

Backups are not synonymous with replicas. A real-time follower may faithfully receive a malicious deletion; an immutable prior snapshot can still recover the deleted material. Testing only a successful restore into an isolated development process is not enough. Teams should drill full-system recovery with encrypted keys, identity permissions, network routes, schema versions, and downstream replay behavior. Define recovery time and recovery point objectives as measurable targets, then gather real restoration metrics under failure scenarios. Avoid promising zero data loss when the acknowledgement path permits asynchronous lag. Failure testing should include the unpleasant cases: a stale snapshot that starts correctly but silently excludes recent commits, a corrupt segment, and a partial rollback of just one shard.

Create a recovery visualizer that tracks snapshot index S, committed index C, and locally appended index A where S≤C≤A. Allow the learner to crash a replica, restore a snapshot at S, and replay only entries known committed through C. Display a state digest for each step and distinguish tentative entries above C. Inject a corrupt log record and require a visible verification failure before serving client reads. Ask students to test how the replay count C−S changes as snapshots become less frequent. The simulator excludes actual filesystem flush semantics and checksum collision probabilities; these details must be validated in a real storage implementation.

Wide-area placement is a product decision shaped by latency, legal requirements, failure isolation, and user expectations. If clients in two continents require a single linearizable account record, the system usually pays for communication with an authority or a quorum that can establish the latest value. A nearby stale replica can answer some read-only queries quickly, but only if the application's contract permits stale results. Replicating every record to every region increases network and storage costs and may create greater coordination load. Partitioning data by geography reduces cross-region traffic for region-local operations, while complicating accounts that migrate or interact across regions. Separate read locality, write authority, failover geography, and data residency as distinct design dimensions.

Consider a university course catalog served across three regions. Course descriptions change rarely, so edge caching with bounded staleness can deliver fast pages even if the origin region becomes unreachable. Seat enrollment changes the contract: an apparently free place is not a confirmed enrollment until the authoritative allocation has executed. The service may display cached availability with a warning but perform the actual reservation through a strongly consistent endpoint. Alternatively, predetermined regional seat quotas allow some partitioned writes without overbooking, at the cost of requiring quota transfer when demand changes. Mixing models by operation can improve performance while preserving critical invariants; the design should make the difference visible to the reader and the client API.

Availability must be specified at the request level. A service may be reachable, serve a cached profile, and reject payment submission due to quorum loss, all at the same instant. Global uptime percentages flatten those distinctions. End-to-end service-level objectives should track which operations succeed within an agreed latency and correctness envelope. A request returning HTTP 200 with an incorrect balance is not a successful operation in the meaningful sense. A cross-region leader can reduce dependence on any one site but may raise steady-state latency, and quorum placement affects which regional outage patterns leave a majority. A strategy of placing two voters in one region and one in another tolerates different failures than a design with one in each of three regions.

There are costs outside the core theorem. Encryption keys may be region-bound, personal data may have residency requirements, and backups can silently move data across jurisdictions if configured carelessly. Observability traffic and distributed tracing can leak sensitive fields when services cross trust boundaries. Disaster recovery can fail because credentials or control-plane dependencies share the same region as the data. Model dependencies as a graph, not merely as colored dots on a map. Evaluate physical failure correlation, client routing, recovery procedures, and operational staffing. Only then select whether the product requires hot active-active writes, active-passive failover, regional read replicas, or a simpler centralized design. Complexity has a maintenance cost that must be justified by real availability or latency needs.

Build a geographic placement worksheet with three regions, simulated round-trip times, voter placement, and client distribution. For each placement calculate whether a majority remains after losing any one region and estimate lower-bound coordination latency from chosen communication paths. Label the estimates as simplified lower bounds, because queueing, serialization, application work, and durable flushes add time. Then simulate a partition separating one region from the other two. Ask which operations may remain available when writes require a quorum but static catalog reads use cached snapshots. Students should explain the difference between 'can serve any response' and 'can fulfill a confirmed transaction correctly'. The exercise does not certify regulatory compliance, which requires jurisdiction-specific review.

A research archive stores petabytes of experiment artifacts, small metadata records, and mutable access policies. Growth has made one storage shard too hot, so the team plans to move part of its key range onto two new shards. The archive cannot simply stop every researcher for several days, but it also cannot permit two concurrent owners to accept contradictory authoritative updates. First define the identity and location map. Each key has a current ownership epoch, and routers resolve keys to shard groups. The migration begins with a consistent snapshot of a source range, then streams changes produced while the snapshot copies. The target must catch up to an agreed cutover point before it becomes authoritative. The source must be fenced after cutover so stale routers cannot keep writing the previous owner.

The data migration includes reads, writes, metadata versions, encryption context, and access-control records. Copying object bytes without copying policy metadata can expose the wrong dataset to the wrong group. A client that downloads a partly moved object may see mismatched versions unless readers obey a read-routing rule. One design keeps the source authoritative during copying, reroutes reads and writes only after a cutover barrier, and preserves forwarding information temporarily. Another permits dual writes but then needs a rigorous reconciliation and acknowledgement policy. Dual writing without a durable common transaction can leave the source updated while the target misses a write; silently declaring completion would be unsafe. Students should draw the crash windows for every proposed variant.

Consider a concrete workload. The hot shard handles 9,000 queries per second and 700 writes per second, with one project contributing half the reads. Migrating the cold half of the key range may leave the hot project on the original shard with almost the same read pressure. The routing plan should use measured access distribution rather than assuming evenly distributed keys. Caching public metadata near researchers might reduce read pressure more cheaply than splitting storage ownership. If the project itself can be partitioned by experiment run, a composite key might distribute load while preserving useful locality for bulk exports. Each alternative changes index maintenance and authorization checks. Capacity planning should estimate migration network traffic, repair rate, and extra reads during the transitional period.

Introduce a cross-shard export workflow that assembles a report from several partitions. If source shard A has copied an artifact at version 17 while metadata shard B still references version 16, the export may assemble an inconsistent snapshot. A coordinated snapshot token or versioned read protocol can establish a coherent view, but carries an implementation cost. The business may accept a 'best effort latest' summary for noncritical browsing while requiring an exact version for a published experiment result. Thus the design can use different consistency contracts for different operations. These distinctions must be visible to users: a report used to support a scientific conclusion needs provenance, while a thumbnail preview may safely lag.

The final acceptance test injects a router crash, a source crash during delta copying, and a stalled target. Each attempt records an ownership epoch, final copied offset, outstanding writes, and verification hashes for migrated content. After planned cutover, send an update from an old router to the source and confirm rejection or safe redirect. Then restore one shard from an old snapshot and test how it rejoins without resurrecting previously retired authority. Verify sensitive data location and access policy alongside byte checksums. Publish a rollback plan that explains when the source can regain ownership and which new writes must be replayed. A migration that merely reports success because both servers answer ping has not established data correctness.