Old-School Programming: From Consensus Algorithm to Sharded Database to Distributed Transaction

Preface

Old-school programming, handcrafted with care
No shortcuts taken, no tokens burned
Born of logic, shaped at the fingertips
Reasoning in between, causality reveals itself
Every line, thought through the hard way
Every run, true to its original intent
Old-school programming — honoring the beauty of purity 🫡

Years of engineering experience have taught me that technically successful projects are rarely built on an “absolutely perfect” design. More often, they come from making appropriate trade-offs under complex constraints. Replication, sharding, transactions, consistency, and consensus... these concepts appear as separate chapters in textbooks, but in real systems they are often intertwined and tightly coupled, significantly increasing the cognitive burden on engineers. Therefore, without a deep understanding of how these mechanisms work together, it is difficult to make reliable design decisions. Conversely, the more you understand how strongly consistent transactions are implemented, the more confident you can be when facing various trade-offs. Of course, studying these mechanisms does not mean that system design should default to introducing the strongest semantics; rather, it means understanding exactly what costs support those semantics and where they ultimately push complexity.

After previously finishing CMU-15445, I felt I had already developed a very deep understanding of single-node SQL. But after reading DDIA, I still felt my understanding of transactions was missing the final layer: how exactly are the distributed transactions we use every day in DynamoDB implemented? So I dug out the Raft and Shard KV code I had written earlier while studying the MIT 6.5840 public course, and on top of that, hand-built support for the classic two-phase commit (2-phase commit, 2PC). After finishing it, my biggest takeaway was this: many abstractions and theories in distributed systems look simple on the surface, but once you actually try to turn them into working engineering, difficulties emerge endlessly from the countless details introduced by interactions among system components.

This article systematically reviews several key technical points and implementation ideas encountered in this 2PC implementation.

Origin: Why Distributed Transactions Are Needed

In the classic older version of the MIT 6.5840 labs, the main path is roughly to first implement the Raft protocol, then a single-shard KV built on Raft, and finally evolve it into Shard KV (a sharded database) by adding a controller. By that point, the system already has:

horizontal database partitioning based on shards
replication consistency within a single replica group
support for group reconfiguration and shard migration

But it is still missing one very important capability: a series of operations spanning multiple shards and requiring atomic commit—that is, distributed transactions in the database sense. Without this capability, our KV database cannot be used to implement exactly-once processing mechanisms such as the outbox pattern, which is the foundation for implementing idempotent business logic.

Therefore, our goal is to integrate a classic 2PC transaction layer on top of the existing Shard KV as non-invasively as possible, thereby turning it into a functionally complete distributed KV database.

If a transaction touches only a single shard, then it is enough to put it into that replica group’s own Raft log. But if a transaction involves multiple groups at once, two additional problems must be solved:

Ensure that all participants ultimately either commit together or roll back together (Atomicity)
Before the commit result is finalized, other transactions must not break this transaction’s (Isolation) semantics

The first problem corresponds to 2PC, and the second corresponds to 2PL (2-phase locking).

Transaction Interface Design

We define the following interface:

Clerk.Transaction(txnId string, operations []TxnOperation) []string

A transaction consists of an ordered list of operations, []TxnOperation, containing a sequence of GET/PUT/APPEND; the return values also correspond strictly one-to-one with the input positions:

GET returns the value read
PUT returns the value written
APPEND returns the final value after appending
If the transaction aborts, it returns nil

When facing instability such as network jitter, server crash/recovery, and leader switching, client retries of the same transaction are the norm, not the exception. Therefore, the semantics of Transaction() must explicitly support idempotence:

Repeated calls with the same txnID should return the result of the first transaction
Even if later calls carry different operations, the system should use the transaction content from the first version that successfully entered the state machine

A subtle but important detail here is that the system does not ultimately execute according to “what the last request looked like,” but according to “the first persisted transaction plan.” And it is not enough to merely commit successfully—the coordinator must also be able to return the complete Values idempotently, in the same order as the input operations, and with the same result on retries. Therefore, the coordinator cannot merely record that “this transaction committed successfully”; it must also persist the final results. Of course, in a real engineering implementation, we would certainly introduce some form of TTL mechanism to clean up long-expired transactions so that space usage does not grow without bound.

Coordinator

In 2PC, the coordinator mainly addresses three problems:

How to persist transaction state
How to advance state safely
How to recover after a crash

State Machine Definition

Based on previous experience implementing Shard KV, we know that the coordinator is essentially just maintaining a set of replicated state machines mapping TxnID => TxnState, with the following states:

Prepare
Commit
Abort
Committed
Aborted

Their meanings are:

Prepare: transaction information has been persisted and prepare RPCs are being sent to all participants
Commit: the commit decision has been made and is being propagated to all participants
Abort: the abort decision has been made and is being propagated to all participants
Committed/Aborted: terminal transaction states that can be returned directly to the client
Once the coordinator decides to enter Commit, it can never revert to Abort
Once the coordinator decides to enter Abort, it can never return to Commit

The reason to make Commit and Abort explicit intermediate states, rather than recording only terminal states, is that “making a decision” and “all participants have completed the final action” are not the same thing. If the coordinator crashes while propagating commit / abort, crash recovery must rely on Raft, so this state must be explicitly written into the log.

State Advancement

Raft.Start() is only a proposal, not consensus

Looking back at our previous Shard KV implementation, state advancement must follow a specific pattern: “first hand the command to Raft, then wait until the local state machine has applied it successfully before continuing.” The reason is that, in this system, the semantically authoritative persistent state is not some temporary result locally constructed by a leader thread, but the result that has entered the replicated state machine and been executed in log order. Raft.Start() can only show that the current node is able to propose a log entry; it cannot guarantee that the entry will eventually commit, cannot guarantee that the current node will still be the leader when it commits, and cannot guarantee that the entry will not be overwritten by some other log.

If we treat the transaction as already persisted immediately after Start() returns, the coordinator is effectively treating an intention that has not yet been confirmed by the cluster and not yet absorbed by the state machine as if it were a fact. In that case, subsequent logic may continue advancing prepare, commit, or abort based on a transaction state that does not actually exist. Then after crash recovery, we would discover that the state machine contains no corresponding record at all, ultimately breaking the basic guarantee that “after recovery, the state machine can always be trusted.”

Persisting Transaction Operations

When a cross-shard transaction enters the coordinator, it is first split according to the current configuration into: “groupID => the subsequence of operations handled by this replica group.” But simply saving these group-local sub-operations is not enough. In the end, the local Values returned by each participant still need to be stitched back together in the original transaction order. Therefore, the coordinator must also persist “groupID => the position indexes of these sub-operations in the original transaction.”

Clearly, the Config used when the transaction was initiated must also be stored, so that when sending RPCs, the coordinator can find the address of the corresponding replica group through groupID. In other words, we bind a transaction to the configuration at the time it was started (shard => replica group). Later we will see that this is critical for correctly implementing reconfiguration.

Crash Recovery

Although the Raft-based state machine is persistent, the entity that actually advances state is a thread, and threads are not reliable. Since the coordinator may crash, after recovery there may no longer be any thread responsible for continuing to advance transactions in the state machine that have not yet reached a terminal state—that is, the work of sending RPCs and collecting responses.

At this point we must answer: for a transaction stuck in the intermediate state Prepare/Commit/Abort, who continues advancing it? Intuitively, some kind of recovery / reconciliation driver is required:

the leader periodically scans the coordinator state machine
finds transactions in the intermediate states Prepare/Commit/Abort
if no thread is currently responsible for advancing one of them, it restarts the corresponding worker thread

Here we need to be careful that when starting a new thread, deduplication should be based on TxnID + TxnStatus, to avoid spawning multiple threads to do duplicate work. We also need to avoid deduplicating only by TxnID, because there is a corner case like this:

the thread executing the Prepare state has not fully exited yet
but it has already persisted the commit / abort decision into the state machine
a thread for advancing the new state has just started
at this point, if we see that there is already a running thread for the same TxnID, the new one would be incorrectly blocked

The difficulty of using consensus algorithms lies not only in state machine design, but also in whether you can reason clearly about the concurrent runtime semantics during state advancement.
That is also why many systems “look correct,” but fail as soon as they run.

Participant

State Definition

Unlike the coordinator, the participant side keeps only the three states Prepared, Committed, and Aborted, because it only carries local responsibilities. A participant does not decide the final fate of the transaction. It only needs to answer two questions: whether the transaction has successfully acquired locks in the current replica group, and once the final decision arrives, whether the current group should execute commit or abort. Prepared means local checks have passed and the locks are held, waiting for the coordinator’s final decision; Committed and Aborted mean the transaction has already terminated locally, and are used to support idempotent retries. Finer-grained intermediate states are meaningful only on the coordinator side, because only the coordinator needs to distinguish transaction advancement stages such as “sending prepare,” “about to enter commit,” or “sending abort.”

From 2PC to 2PL

For most people who have only read textbooks, the first thing that comes to mind when distributed transactions are mentioned is 2PC. But once you actually build it, you find that 2PC itself only completes half the puzzle—the atomic decision of “everyone commits together or everyone aborts together”—while the other half is isolation among transactions. Only together do they achieve the final goal: strong consistency.

Distributed transactions must ultimately pin down certain resources or commit conditions, thereby preventing concurrent changes that would invalidate the correctness of the current transaction’s commit. On the participant side, we need a mechanism to lock the keys involved in a transaction. This is basically a form of strict 2PL, with roughly the following semantics:

Perform a full conflict check during Prepare; if it fails, abort the transaction immediately
Only after everything passes are all keys involved in the transaction added to the read/write lock table
Once in Prepared, the locks are held until Commit / Abort
The locks are released together only after the final decision RPC arrives

If locks are not held until commit, other transactions may commit concurrent modifications between prepare and commit, breaking isolation.

Atomicity of Locking

An important constraint here is that local prepare must lock atomically: on a given participant, a transaction must either successfully acquire all locks for its keys and enter Prepared, or abort without leaving behind a half-locked state. This property depends mainly on the serialized execution of Raft state machine apply: lock conflict checking and lock-table updates happen within the same single-threaded state machine step, so there is no issue of two prepares interleaving concurrent modifications to the lock table locally. On commit, the participant executes the transaction’s operations in order, returns local results, and releases the locks; on rollback (abort), it simply releases the locks. In this way, transaction commit atomicity comes from two-phase commit, while transaction isolation is guaranteed by local locks held continuously after prepare.

ConditionCheck

To support transaction conditions similar to DynamoDB-style ConditionCheck, we adopt a relatively simplified approach and directly treat condition checking as a special kind of OpType:

type OpType string

const (
    Put    OpType = "Put"
    Append OpType = "Append"
    Get    OpType = "Get"
    // Transaction support
    TxnCondEqual    OpType = "TxnCondEqual"
    TxnCondNotEqual OpType = "TxnCondNotEqual"
    TxnCondExist    OpType = "TxnCondExist"
    TxnCondNotExist OpType = "TxnCondNotExist"
)

type TxnOperation struct {
    Key   string
    Value string
    Op    OpType
}

These condition-check operations, like ordinary Get/Put/Append, appear directly in the same transaction’s operations list, and the participant checks them all together during the Prepare phase:

If all conditions hold, the transaction continues
If any condition fails, the participant directly returns abort, and the entire transaction never enters the commit phase

In concurrency control, these condition checks are treated as read-only accesses: they participate only in local conflict checking and update the read set, but do not modify actual data during the Commit phase.

This way, we do not need to design an additional interface for “a single write operation carrying condition expression”; such requirements can instead be decomposed into multiple explicit condition checks and placed together with subsequent writes into the same 2PC transaction. The final semantics remain unchanged: the condition checks and the subsequent writes either all succeed together, or all fail together and leave no side effects.

The advantage of this design is that it is simple to implement, clear in semantics, and expressive enough for common use cases such as “conditional PUT” in DynamoDB transactions.

Correct Handling of Out-of-Order RPCs

In an unreliable network, the following can absolutely happen:

the coordinator has already decided to abort
the participant receives abort first
the delayed prepare arrives only afterward

If, under this ordering, the participant simply returns OK when receiving abort, then a transaction that should have been terminated would be prepared again and get stuck there forever—because the coordinator would mistakenly believe abort had already been completed correctly. Therefore, the participant side must support:

leaving a tombstone upon receiving an Abort RPC
if a later Prepare RPC with the same TxnID arrives, directly and idempotently returning aborted

Extending 2PC: Introducing a Commit Decision Point

Classic 2PC has a theoretical dependence on the liveness of the coordinator: once a transaction is stuck in the prepared intermediate state, and the coordinator crashes for a long time before finishing broadcasting the decision, participants may remain suspended for a long time, and the system cannot make progress. Industrial implementations usually make engineering improvements around this point. Taking systems such as TiDB/Percolator as an example, a transaction selects a primary participant during its initiation stage, and the state machine of every secondary participant records who the primary is. In this way, whether the transaction has ultimately committed no longer depends solely on the original coordinator to announce it; when a secondary cannot obtain the final decision for a long time, it can query the commit status of the primary, and then decide whether it should commit or abort locally. In this project, I implemented a minimal simulation of this idea: selecting a PrimaryGID for each transaction, so that secondaries can actively query the transaction status of the primary while waiting for the coordinator, thereby partially reducing the dependence on the continued survival of the coordinator.

Strictly speaking, after introducing a persistence mechanism such as Raft, the coordinator is no longer a single point of failure in the traditional sense. Once the transaction state and state-transition commands have been written into the replicated state machine, as long as a majority is still alive, a new leader can take over and continue to complete prepare, commit, or abort. Therefore, improvements such as primary commit discussed here are not intended to solve the problem that “once the coordinator goes down, the whole system becomes unrecoverable,” but rather to further reduce the dependence of transaction progress on the continued availability and timely responsiveness of the coordinator. In other words, it mainly improves liveness and recovery paths, rather than compensating for missing correctness guarantees.

Of course, this mechanism still has a prerequisite: the transaction must at least have progressed to the point where the primary commit has been durably completed. Only after the primary becomes a queryable commit decision point can secondaries infer from it whether they should commit locally or continue waiting; if all participants remain in the prepared state, then the transaction will still be stuck here, unless we introduce an additional garbage-collection mechanism to clean up these expired transactions.

How 2PC Interacts with MVCC and OCC

In a single-node SQL database, transactions are usually implemented based on MVCC (Multi-Version Concurrency Control): each write creates a new version at commit time, and a commit timestamp determines its visibility to other transactions. This model is essentially “single-phase”—only after the transaction commits does the new data version become visible to others.

When the system evolves into distributed transactions and introduces 2PC, a new problem appears: 2PC requires that a transaction that has successfully prepared must be able to commit, but with OCC (Optimistic Concurrency Control), it is possible that only at the final validation stage do we discover that the current transaction’s read set conflicts with another transaction’s write set. At that point, since success has already been promised, how can the transaction still be aborted? In other words, between prepare and commit, the transaction has already “decided to write,” but has not yet finally committed. How should this intermediate state be represented in the MVCC model so that successful final commit is guaranteed?

A natural extension is to introduce intents (uncommitted versions) into MVCC. From the perspective of the data model, this can be understood as an “intermediate state” in the MVCC version chain:

During the Prepare phase, each participant brings the entire transaction “almost to completion”:
- complete read/write conflict checks to avoid concurrency anomalies (write skew)
- write a temporary version with transaction identity (an intent) into the version chain
- this version already contains the final written value, but has not yet been marked committed
For ordinary reads occurring on the same node, such versions are invisible by default
- their visibility is no longer determined solely by timestamp, but also depends on transaction state (pending / committed / aborted)
In the final Commit phase, the intent is marked as an official version, making it visible to other transactions

In traditional 2PL, the semantics of prepare are “freeze the data by locking it to prevent conflicts”; in 2PC + OCC, the semantics become:

explicitly declare future write intent by writing an intent, and reserve a position in the version chain
complete conflict detection on the read/write sets in the same phase, ensuring that commit conditions are already satisfied
once in the prepared state, the intent takes on the semantics of a “lock,” preventing subsequent transactions from writing the same row or forcing them to handle the conflict
- if a later write transaction sees an intent, it cannot “simply ignore it and continue writing,” otherwise this becomes a write-write conflict; it may choose to wait, preempt, or abort itself
- later read transactions can usually still read the older committed version, but under Serializable isolation, seeing an intent tells the system that there is an unfinished write here, so this read must be included in dependency tracking, and it must eventually be determined whether it is ordered before or after the write

The core of Serializable is that the dependency graph among transactions must be acyclic, because only then can a topological sort produce an equivalent serial execution order.

Shard Migration and Transactions

At this point, we have solved transaction execution in a static system, but in a real system the configuration (the shard => replica group mapping) does not remain unchanged, whereas an implicit premise of 2PC is that the participant set must be stable.

Why Bind Transactions to Configurations

As mentioned earlier, transactions must be bound to the system configuration at the time they are initiated. Let us first explain why this is necessary—if they are not bound, the transaction’s execution context is no longer fixed, directly violating a basic condition of 2PC: the final decision must be sent to the same participants that originally took part in prepare, making recovery and idempotent semantics impossible to converge. For example, the participant set of the same transaction could drift:

when a transaction is in Prepare, some key belongs to gid=100
by the time Commit happens, that key may already have migrated to gid=101
if the coordinator does not save the transaction execution plan under the old config, but instead always looks it up again under the latest config, then the following can happen:
- Prepare was actually done by 100
- but Commit is sent to 101

Now the problem is obvious: 101 has no prepared record for this transaction. Since the locks are not held there, the operation return values it can provide may be incorrect, breaking transaction isolation.

Conditions for Advancing Reconfiguration

In the original Shard KV system, each replica group periodically polls shardctrler for new configurations, and when a configuration change is detected, it handles shard migration by itself. After transaction support is introduced, the conditions for advancing this process also need adjustment. Consider the following scenario:

a participant has already executed Prepare for some transaction
a configuration change occurs, and a shard involved in that transaction is now supposed to be moved away
can that participant still process the transaction’s Commit/Abort or not?

In fact, under the old configuration, the participant’s identity must remain valid until the transaction terminates. Otherwise, if it accepts and applies the new configuration, then later when the coordinator issues Commit or Abort to the groupID under the old configuration, the correct shard owner can no longer be found, causing the transaction to get stuck in an intermediate state.

The solution is to add an extra constraint to the original reconfiguration logic: if the shard to be sent intersects with the lock set held by pending transactions, the current configuration change does not take effect yet. Although from the perspective of the external shardctrler, the latest configuration already exists and is visible to clients, for a participant with unfinished transactions, the old configuration remains in effect until the relevant transactions finish, and only then is the new configuration applied. In other words, although neither the coordinator nor the participant can prevent shardctrler from publishing a new configuration, the participant can choose to delay the actual moment at which it switches to the new configuration, thereby preserving the key condition required by 2PC.

How Industrial Systems Handle It

Industrial systems are usually reluctant to let reconfiguration be directly blocked by possibly long-running transactions, because that would couple control-plane progress too tightly to transaction liveness. A more common solution is to design some migration protocol that understands transactional intermediate states, allowing those intermediate states to be copied and transferred along with shard data, so that after shard migration the new owner can continue taking over the transaction. But the price of that flexibility is a significant increase in the engineering complexity and testing difficulty of the data model, storage engine, and recovery mechanisms.

Testing: Vibe Coding All the Way

The emergence of LLMs has made producing code extremely cheap, but for distributed systems, complexity does not disappear out of thin air—you still have to ensure the correctness of the business logic. That means you must find ways to constrain the size and readability of AI-generated output so that validation remains within a cognitive burden humans can bear. Therefore, for core code in critical paths, LLMs cannot dramatically improve development efficiency; the final bottleneck is still your own reasoning and review of correctness.

By contrast, writing test code is a purely mental drain. Most of the time, such code has no complicated context and is relatively easy to validate at a rough level, which makes it especially suitable for completion through vibe coding. For this project, I believe AI saved me enough time on testing to roughly double my overall efficiency, especially when you need to use some testing tool you are unfamiliar with, such as the linearizability check below.

Linearizability Test

Linearizability is a read/write guarantee for a single object. It constrains whether a series of externally observable operations can be explained as some serial execution that respects real-time order.

Since transaction execution involves multiple keys, its linearizability test models a Transaction(txnID, ops) as one high-level atomic operation, rather than splitting each Get/Put/Append on each key apart for checking as in Shard KV. We run a randomized transaction test that simulates network failures and node crashes, recording for each transaction its start time, end time, input operation sequence, and final output during execution, and then feed the entire history into Porcupine. The job of the Porcupine library is to search, subject to real-time ordering constraints, whether there exists some serial order that can explain the return values and final state of this batch of transactions—if such an order exists, the history is considered linearizable.

To let Porcupine determine whether a transaction history is valid, we need to provide it with a sequential execution model. The model here is rough but effective: maintain a simple map[string]string as the abstract state, simulate the effects of Get, Put, and Append on state and return values according to the order of operations inside each transaction, and treat abort as a legal outcome that does not change state. In this way, what the linearizability check verifies is whether these transactions really behaved externally like atomic operations, rather than leaking intermediate state during execution or violating atomic commit semantics.

Closing Remarks

Returning to the sentence at the beginning of this article: the essence of software engineering has never been the pursuit of some textbook-style “perfect solution,” but rather identifying the boundaries of the problem under real constraints and making reasonable trade-offs.

From consensus algorithms, to sharded databases, to distributed transactions, although this toy project of mine is still far from a truly industrial-grade distributed database, the greatest reward from hand-building my way along this path has been that, by making a series of trade-offs between simplification and preservation, I came to deeply understand what makes strongly consistent transactions so expensive, why many systems would rather step back and adopt weaker saga semantics, and how to answer with greater confidence, when facing design trade-offs—why we should not introduce transactions.