Global Snapshots Across Shards Without a Shared Clock

Snapshot isolation on one node is easy to explain and cheap to implement: a transaction records the set of committed transactions that existed when it started, and every read is filtered against that set. The snapshot is a number, the filtering is a comparison, and no locks are held for reads. Spread the data over shards and each of those three properties becomes a problem, because a number that means “everything committed before this point” only means that if all the commits were ordered by the same counter.

What a snapshot identifier must satisfy

A snapshot has to answer one question for every row version: was this version committed before my transaction started, or not? For the answer to be consistent across shards, the identifier must be comparable with the commit timestamps that shards assign, and the comparison must be stable.

Three properties are needed. The identifier must be unique: two transactions must never share one, or their snapshots cannot be distinguished in a version’s history. It must be monotonic: a transaction that starts after another commits must have a larger identifier, or it will miss the earlier commit. And it must be readable by every shard: each shard must be able to decide, given a version’s commit stamp and the snapshot identifier, whether to show the version.

A per-shard counter satisfies uniqueness and monotonicity within the shard but fails the third property, because shard A’s counter has no relation to shard B’s. A transaction that reads A then B can see A’s state as of counter 100 and B’s as of counter 40, which are unrelated instants. A row written on B at B-counter 45 may be invisible while a row written on A at A-counter 90 is visible, and no consistent global point in time corresponds to that combination.

The usual answer is a global clock of some kind, and the interesting design question is what kind. Options range from a physical clock with bounded error to a logical counter allocated by a single service to a hybrid logical clock that combines both. Each gives a different guarantee about how far apart two snapshots can be and what happens when a node’s clock is wrong.

Physical time as the snapshot

Using wall-clock time as the snapshot identifier is the simplest approach. Each committed transaction is stamped with the physical time at commit, adjusted to account for clock skew so that the stamp is not earlier than any stamp it causally follows. A reader takes the current time as its snapshot and sees versions stamped at or before it.

The property that makes this workable is a bound on clock skew. If every node’s clock is within ε of true time, then a transaction that commits at true time t is stamped no later than t + ε, and a reader that starts at true time t’ sees it if t’ > t + ε. To avoid missing a commit that happened before the reader started, the reader must wait out the uncertainty, or the stamps must be pushed forward so that a commit is never stamped earlier than its true time plus the maximum skew any node might have.

CockroachDB’s design is an explicit instance of this: timestamps are assigned from a hybrid logical clock, and a transaction’s commit timestamp is pushed forward if it would otherwise violate causality with respect to any timestamp it has observed. The result is a total order on transactions that is consistent with real-time ordering, at the cost of occasional timestamp bumps that delay a commit.

The cost of a physical-clock approach is that the guarantee depends on a bound the system cannot verify from inside. A node whose clock jumps forward by a large amount can stamp a commit in the future, which makes it visible to readers that started later but also makes the timestamp order disagree with real-time order for the interval of the jump. Systems handle this by refusing to serve when the measured skew exceeds a configured maximum, which converts a correctness problem into an availability one.

The single-counter approach and its bottleneck

The alternative is a logical counter: one service allocates snapshot numbers, and every shard’s commits are stamped with numbers from that service. Uniqueness and monotonicity are exact, and there is no skew to bound. The counter service becomes a dependency for every commit and every snapshot acquisition, which is a round trip on the critical path and a single point of failure unless the counter is itself replicated.

Replicating the counter is possible, as a replicated log with a counter per entry, but the cost is now paid per commit rather than per read. If the counter is only advanced in batches, a transaction’s snapshot may lag the true commit state by the batch interval, which means a reader can miss a commit that happened before it started. That is the opposite error from the physical clock’s, and it is a correctness violation for read-your-writes if the reader is the same session that committed.

A middle path is to allocate ranges of counter values to shards, so each shard can stamp commits locally from its own range and only occasionally request a new range. This restores local commit speed but reintroduces the ordering problem: a shard stamping from a range that is numerically ahead of another shard’s range will produce commits that appear to happen after transactions that started later. The ranges must be allocated in a way that respects the happens-before relation, which in practice means the allocation must be ordered and the shards must not stamp outside their range until it is safe.

The read path: waiting for a snapshot to be safe

Whatever the identifier is, a reader that wants a consistent view across shards must ensure that the snapshot it uses is not older than any commit it has already observed. Two mechanisms do this.

The first is to acquire the snapshot from the same authority that stamps commits, at the moment the transaction starts. If the counter service hands out snapshot S, then every commit stamped at or below S is included, and no commit above S can be included, which is exactly the definition of a snapshot. The reader’s view is consistent across shards because all shards compare against the same S.

The second is to use a timestamp and wait for safety. A reader takes the current physical time, then asks each shard to confirm that it has no pending commit with a stamp below that time that it has not yet made visible. This is a barrier: the reader cannot proceed until every shard has advanced its visible watermark past the snapshot. The cost is one round trip per query to the slowest shard, which is the same structure as a distributed read that must be linearizable.

The barrier’s cost is the reason many systems avoid cross-shard snapshots for reads that only touch one shard. A single-shard read can use the shard’s local snapshot and pay nothing. The global snapshot is only needed when a transaction reads from more than one shard, and the design question becomes how to keep most transactions single-shard so the barrier is rarely paid. That is the same partitioning argument that reduces two-phase commit’s participant count, and it appears in two-phase commit and the coordinator decision record as the reason co-location matters.

Write conflicts across shards

Snapshot isolation’s write rule is that two concurrent transactions must not write the same row, and this is enforced by a first-committer-wins check at commit time. On one node the check is a comparison against the row’s version. Across shards, a transaction that writes rows on two shards must have both checks succeed, and the two checks must be atomic with respect to each other.

The obvious implementation is to take a write lock on each row at write time rather than at commit time, which converts the conflict check into ordinary locking. That works but reintroduces the deadlock and blocking problems of a locking protocol, and it means a cross-shard transaction holds locks on both shards for its whole duration.

The alternative is to keep writes local and validate at commit, using the two-phase commit machinery to make the validation atomic. Each shard validates its own writes against its own version history, and the coordinator’s decision makes the set of validations atomic. The window between validation and decision is a window in which another transaction can write the same row, which is why validation must be paired with a lock or a reservation that survives until the decision. Systems that validate at commit typically take a short-lived lock during the validation phase, held until the decision arrives, which is exactly the prepared state of two-phase commit.

The isolation anomaly that snapshot isolation permits, write skew, is unaffected by distribution. Two transactions can read overlapping sets and write disjoint sets on different shards and both commit, producing a state that no serial order would produce. Distribution makes it harder to detect because the two transactions’ read sets are recorded on different nodes, and no single node sees the overlap. That is the problem serializable snapshot isolation across shards addresses.

What the snapshot does not cover

A snapshot fixes the read view, not the schema. A transaction that starts before a schema change and commits after it may read columns that the change removed, or write rows whose shape changed. Distributed systems handle this by versioning the schema alongside the data and by requiring DDL to be ordered with respect to transactions, which usually means DDL takes an exclusive lock that conflicts with the snapshot’s acquisition. The interaction is a common source of operational trouble, because a long-running transaction with an old snapshot blocks the schema change, and the schema change blocks new snapshots.

A snapshot also does not bound how long a version must be retained. Every shard must keep row versions that any live snapshot might need, and the oldest live snapshot determines the retention floor. A single long-running cross-shard transaction therefore pins old versions on every shard it touched, which multiplies the storage cost of one slow query by the number of shards. Vacuum and garbage collection must consult a global minimum rather than a local one, which is another place where a distributed snapshot turns a local decision into a coordinated one.