Distributed Transactions with Percolator

The problem left by sharding

Multi-Raft and Sharding gives every range its own Raft group. A write to one key is atomic and durable. But a bank transfer touches two accounts that may live in two ranges, on two different Raft groups with two different leaders. Each group can commit its half independently — and a crash between the halves leaves money created or destroyed. We need atomicity across groups and a consistent snapshot for reads that span them.

The textbook answer, two-phase commit, has a well-known flaw (Distributed Transactions): if the coordinator dies after the participants voted yes, they are stuck holding locks until it comes back. Percolator (Peng and Dabek, Google, OSDI 2010), the design TiKV implements, removes the special coordinator: the transaction's state lives in the data itself, on a primary key that any client can inspect, so anyone can finish or cancel an abandoned transaction. labs/multiraft implements it on top of the Multi-Raft store.

Ingredients

A timestamp oracle (TSO): one strictly increasing counter for the whole cluster. Every transaction gets a startTS when it begins and a commitTS when it commits. (TiKV's Placement Driver provides this, Raft-replicated, handing out timestamps in batches; CockroachDB avoids a central oracle by using hybrid logical clocks plus uncertainty intervals.)

Three kinds of data per key, all inside the range's replicated state machine:

type Lock struct {
    Primary   string // the transaction's primary key: its single source of truth
    StartTS   uint64
    ExpiresAt int64  // TTL, chosen by the proposer — never by the applier
}

type WriteRec struct {
    CommitTS, StartTS uint64
    Rollback          bool // marker: this transaction was rolled back
}

type MVCCKey struct {
    Lock   *Lock             // an in-progress transaction's intent
    Writes []WriteRec        // commit records, newest first
    Data   map[uint64]string // values by the writer's startTS
}

A value becomes visible only when a write record points at it: commitTS → startTS. Until then it sits in Data behind a lock.

Reading a snapshot

A transaction reads every key as of its startTS:

case OpTxnGet:
    if k.Lock != nil && k.Lock.StartTS <= c.StartTS {
        // A transaction that started before our snapshot may still commit
        // with a commitTS below it: we cannot know the value yet.
        return ErrLocked (with the lock)
    }
    for _, w := range k.Writes { // newest first
        if w.CommitTS <= c.StartTS && !w.Rollback {
            return k.Data[w.StartTS] // the latest version committed before our snapshot
        }
    }
    return not found

The lock rule is the subtle part. A transaction with startTS = 10 holding a lock might be committed at commitTS = 15; a reader at startTS = 20 must see that value, but the commit hasn't happened yet. Ignoring the lock would return the older value and break the snapshot. So the reader stops and resolves the lock (below). Locks from transactions that started after the reader's snapshot are irrelevant to it and ignored.

Committing: two phases, one commit point

func (t *Txn) Commit() error {
    primary := t.order[0]
    // Phase 1: prewrite every key — lock it and stage the value.
    //   Fails on a write committed after our startTS (write-write conflict)
    //   or on another transaction's lock.
    prewrite(primary); prewrite(secondaries...) // in parallel

    // Phase 2: commit the primary. THIS is the commit point.
    commitTS := t.tso.Next()
    commit(primary, commitTS) // write record + remove lock, in one Raft entry

    // Secondaries: the transaction is already committed; this is cleanup.
    commit(secondaries..., commitTS) // in parallel, best effort
}

The whole transaction is committed at the instant one Raft entry — the primary's commit record — is applied. Before it, the transaction can be rolled back; after it, it must be rolled forward. Every secondary lock points at the primary, so anyone who finds a secondary lock can learn the outcome from one place.

Resolving abandoned transactions

A reader (or a writer whose prewrite hit a lock) checks the lock's primary key with a CheckTxn command that also carries the proposer's current time:

case OpCheckTxn: // applied on the primary key's range
    if w, ok := k.writeFor(c.StartTS); ok {
        return committed (w.CommitTS) or rolled-back
    }
    if k.Lock != nil && k.Lock.StartTS == c.StartTS && k.Lock.ExpiresAt > c.Now {
        return alive // still running: wait and retry
    }
    // Expired, or the primary was never prewritten: roll it back. The
    // rollback marker makes a late prewrite or commit of this
    // transaction fail, so the decision is final.
    rollback(k, c.StartTS)
    return rolled-back

Then the resolver applies the same outcome to the locked key: commit it with the primary's commitTS, or roll it back. Two properties make this safe:

  • The decision is made once, on the primary, through Raft. If the original client wakes up and tries to commit after a resolver rolled its primary back, its commit finds the rollback marker and fails.
  • Time enters as data. ExpiresAt was chosen by the transaction's client and Now by the resolver; the state machine only compares them. Replicas never consult their own clocks — the determinism rule from Multi-Raft and Sharding.

Tests and measurements

All on 3 stores with the key space split into 4 ranges, so transactions span Raft groups; simulated network, testing/synctest.

Bank invariant. 100 accounts with 100 each; 6 workers do random transfers (read two balances, write both) for 20 seconds; 2 auditors run read-only transactions that sum all 100 balances.

Run Transfers committed Aborted on conflict Audits Locks resolved by readers
Correct reads 1,252 154 (11%) 26, all with total = 10,000 106
Mutation: reads ignore pending locks 1,272 174 (12%) 28, of which 26 saw a wrong total 66

The second row is the check on the test itself: with one condition removed from OpTxnGet, almost every audit saw a debit without its credit or vice versa. A test that passes against a known-broken implementation would prove nothing.

Client crash after the commit point. A transaction writes keys in two ranges, commits the primary, and "crashes" before committing the secondary. A later reader finds the secondary's lock, checks the primary, sees committed, rolls the secondary forward, and reads the new value.

Client crash before the commit point. A transaction prewrites both keys and crashes. A reader finds the lock, sees alive, and waits; once the 3-second TTL expires (measured wait: 3.04 s), the check rolls the primary back, and the reader sees the old values. When the crashed client "wakes up" and tries to commit its primary, it gets ErrTxnAborted.

Contention. 8 workers transferring between random accounts for 10 seconds:

Accounts Committed Aborted Abort rate
10 262 462 63.8%
100 779 135 14.8%
1,000 858 18 2.1%

Optimistic concurrency control — conflicts detected at prewrite, loser retries — is excellent when conflicts are rare and wasteful when they are not: with 10 hot accounts, nearly two thirds of all work was thrown away. Hot keys are the classic weakness of this design; systems add pessimistic locking modes (TiDB's pessimistic transactions) or queue conflicting transactions on the key.

Snapshot isolation, not serializability

Percolator provides snapshot isolation: every transaction reads a consistent snapshot, and two transactions cannot both commit writes to the same key. It does not prevent write skew, where two transactions read the same keys and write different ones. Measured: Alice and Bob are both on call; T1 reads Bob's flag and turns Alice off, T2 reads Alice's flag and turns Bob off — both commit, and nobody is on call.

The single-group optimistic transactions in Iterators, Snapshots, and Transactions rejected the same scenario because they validate reads at commit. Percolator validates only writes (at prewrite). Fixes when an invariant spans keys: write the keys you read (TiDB's SELECT ... FOR UPDATE takes locks on them), or use an engine that offers serializable isolation (CockroachDB runs serializable by default, tracking read timestamps and refreshing or aborting on conflicts). Background in Isolation Levels and Anomalies.

What a transfer costs

Each transfer here is 6 Raft log entries in two groups (2 reads, 2 prewrites, 2 commits) and 2 timestamp requests, with 5 sequential round trips on the critical path: read, read, prewrite primary, prewrite secondary, commit primary (secondaries are off the critical path). On the disks measured in Raft on Real Hardware, each Raft commit costs ~12 ms at low load, so an unoptimized transfer would take ~60 ms. Production systems attack exactly this:

Optimization Idea Used by
Reads without the log Serve snapshot reads from the leaseholder/leader (ReadIndex or leases) instead of appending TiKV, CockroachDB
Parallel prewrites Prewrite all keys at once, primary included TiKV
One-phase commit All keys in one range: commit in a single Raft entry, no locks TiKV (1PC), CockroachDB
Async commit / parallel commits Consider the transaction committed once all prewrites are durable; compute commitTS from them, so the commit is one round trip TiKV async commit, CockroachDB parallel commits
Pessimistic locks Lock on read for hot keys to avoid repeated aborts TiDB

Common pitfall

Using a node's wall clock to decide whether a lock has expired. A resolver compares the lock's TTL with time.Now() inside the state machine: each replica applies the check against its own clock, so a replica whose clock is 2 seconds fast rolls the transaction back while the others still consider it alive, and the replicas' states diverge. Worse, a client whose clock runs slow sets a TTL that has already expired everywhere else. Time must enter the replicated log as data chosen by one party (the proposer), and TTLs must be generous compared with clock skew, because a transaction that is still running when its lock is declared expired is rolled back — correctly, but expensively.

Review questions

  1. Why does Percolator not suffer from 2PC's blocked participants when the coordinator dies?
  2. Why must a reader at startTS = 20 stop at a lock held by a transaction with startTS = 10, but may ignore one with startTS = 25?
  3. What is the commit point of a Percolator transaction, and what does a resolver do on each side of it?
  4. Why does CheckTxn write a rollback marker when it finds no lock on the primary at all?
  5. Why did the bank test with lock-ignoring reads fail almost every audit, while the write-skew test passed under correct snapshot isolation?