Locks and leases in distributed applications are prone to misuse and are a common source of bugs. Let’s look at one particular case of how they can go wrong.
In “Process Pauses” we saw that a lease is a kind of lock that times out and can be assigned to a new owner if the old owner stops responding (perhaps because it crashed, it paused for too long, or it was disconnected from the network). You can use leases when a system requires there to be only one of some thing. For example:
-
Only one node is allowed to be the leader for a database shard, to avoid split brain (see “Handling Node Outages”).
-
Only one transaction or client is allowed to update a particular resource or object, to prevent it from being corrupted by concurrent writes.
-
Only one node should process a given input file to a big processing job, to avoid wasted effort due to multiple nodes redundantly doing the same work.
It is worth thinking carefully about what happens if several nodes simultaneously believe that they hold the lease, perhaps because of a process pause. In the third example, the consequence is only some wasted computational resources, which is not a big deal. But in the first two cases, the consequence could be lost or corrupted data, which is much more serious.
For example, Figure 9-4 shows a data corruption bug due to an incorrect implementation of locking. (The bug is not theoretical; HBase used to have this problem.) Say you want to ensure that a file in a storage service can be accessed by only one client at a time, because if multiple clients tried to write to it, the file would become corrupted. You try to implement this by requiring a client to obtain a lease from a lock service before accessing the file. Such a lock service is often implemented using a consensus algorithm, discussed further in the related discussion.
The problem is an example of what we discussed in “Process Pauses”. If the client holding the lease is paused for too long, its lease expires. Another client can then obtain a lease for the same file and start writing to the file. When the paused client comes back, it believes (incorrectly) that it still has a valid lease and continues writing to the file. We now have a split-brain situation: the clients’ writes clash and corrupt the file.
Diagram showing an incorrect distributed lock implementation where Client 1's lease expires during a pause, leading Client 2 to acquire a lease and write data, resulting in file corruption when Client 1 continues executing.
Figure 9-4. An incorrect implementation of a distributed lock: client 1 believes that it still has a valid lease, even though it has expired, and thus corrupts a file in storage
Figure 9-5 shows a different problem that has similar consequences. In this example there is no process pause, only a crash by client 1. Just before client 1 crashes, it sends a write request to the storage service, but this request is delayed for a long time in the network. (Remember from “Network Faults in Practice” that packets can sometimes be delayed by a minute or more.) By the time the write request arrives at the storage service, client 1’s lease has already timed out, allowing client 2 to acquire it and issue a write of its own. The result is corruption similar to that in Figure 9-4.
Diagram showing a similar file corruption scenario as in the previous figure, except that this time it is caused not by a process pause, but by Client 1's write request being significantly delayed on the network.
Figure 9-5. A message from a former leaseholder might be delayed for a long time and arrive after another node has taken over the lease.
Fencing off zombies and delayed requests
The term zombie is sometimes used to describe a former leaseholder that has not yet found out that it lost the lease and is still acting as if it were the current leaseholder. Since we cannot rule out zombies entirely, we have to instead ensure that they can’t do any damage in the form of split brain. This is called fencing off the zombie.
Some systems attempt to fence off zombies by shutting them down—for example, by disconnecting them from the network, shutting down the VM via the cloud provider’s management interface, or even physically powering down the machine. This approach is sometimes known as shoot the other node in the head (STONITH), though we prefer not to use such violent terminology. In any case, it is not particularly effective: this approach does not protect against the large network delays depicted in Figure 9-5; all the nodes could shut one another down; and by the time a zombie has been detected and shut down, it may be too late and data may already have been corrupted.
A more robust fencing solution, which protects against both zombies and delayed requests, is illustrated in Figure 9-6.
Diagram illustrating a fencing mechanism where clients obtain increasing fencing tokens from a lock service to ensure safe write operations to storage in chronological order.
Figure 9-6. Making access to storage safe by allowing writes only in the order of increasing fencing tokens
Let’s assume that every time the lock service grants a lock or lease, it also returns a fencing token, which is a number that increases every time a lock is granted (e.g., incremented by the lock service). We can then require that every time a client sends a write request to the storage service, it must include its current fencing token.
Note
There are several alternative names for fencing tokens. In Chubby, Google’s lock service, they are called sequencers, and in Kafka they are called epoch numbers. In consensus algorithms, which we will discuss in the related discussion, the ballot number (Paxos) or term number (Raft) serves a similar purpose.
In Figure 9-6, client 1 acquires the lease with a token of 33, but then it goes into a long pause and the lease expires. Client 2 acquires the lease with a token of 34 (the number always increases) and sends its write request to the storage service, including the token. Later, client 1 comes back to life and sends its write to the storage service, including its token value, 33. However, the storage service remembers that it has already processed a write with a higher token number (34), so it rejects the request with token 33. A client that has just acquired the lease must immediately make a write to the storage service, and once that write has completed, any zombies are fenced off. These operations are similar to the optimistic concurrency control (OCC) technique we saw in “Pessimistic versus optimistic concurrency control”, except that fencing is permanent, while concurrency control failures can be retried.
If ZooKeeper is your lock service, you can use the transaction ID zxid or the node version cversion as a fencing token. With etcd, the revision number along with the lease ID serves a similar purpose. The FencedLock API in Hazelcast explicitly generates a fencing token.
This mechanism requires that the storage service has some way of checking whether a write is based on an outdated token. Alternatively, it’s sufficient for the service to support a write that succeeds only if the object has not been written by another client since the current client last read it, similarly to an atomic CAS operation. For example, object storage services support such a check; Amazon S3 calls it conditional writes, Azure Blob Storage calls it conditional headers, and Google Cloud Storage calls it request preconditions.
Fencing with multiple replicas
If your clients need to write to only one storage service that supports conditional writes, the lock service is somewhat redundant, since the lease assignment could have been implemented directly based on that storage service. However, once you have a fencing token, you can use it with multiple services or replicas and ensure that the old leaseholder is fenced off on all of them.
For example, imagine the storage service is a leaderless replicated key-value store with LWW conflict resolution (see “Leaderless Replication”). In such a system, the client sends writes directly to each replica, and each replica independently decides whether to accept a write based on a timestamp assigned by the client.
As illustrated in Figure 9-7, you can put the writer’s fencing token in the most significant bits or digits of the timestamp. You can then be sure that any timestamp generated by the new leaseholder will be greater than any timestamp from the old leaseholder, even if the old leaseholder’s writes happened later.
In Figure 9-7, client 2 has a fencing token of 34, so all its timestamps starting with 34… are greater than any timestamps starting with 33… that are generated by client 1. Client 2 writes to a quorum of replicas, but it can’t reach replica 3. This means that when the zombie client 1 later tries to write, its write may succeed at replica 3 even though it is ignored by replicas 1 and 2. This is not a problem, since a subsequent quorum read will prefer the write from client 2 with the greater timestamp, and read repair or anti-entropy will eventually overwrite the value written by client 1.
Diagram illustrating the use of fencing tokens in a leaderless replicated database, showing Client 1 and Client 2 with different fencing tokens interacting with three replicas, one of which is offline.
Figure 9-7. Using fencing tokens to protect writes to a leaderless replicated database
As you can see from these examples, it is not safe to assume that only one node is holding a lease at any one time. Fortunately, with a bit of care you can use fencing tokens to prevent zombies and delayed requests from doing any damage.