Chapter 76: Consensus & Coordination

Leader Election

Leader election is the process of designating a single node as the coordinator among a group of distributed nodes. Essential for preventing split-brain scenarios and coordinating distributed work.

Why Leader Election is Needed

Single Writer: Ensure only one node writes to prevent conflicts (e.g., database primary)
Work Coordination: One leader distributes work to followers
Ordering: Leader sequences operations for consistency
Resource Management: Leader allocates resources (e.g., partition assignment in Kafka)
Caching Consistency: Leader invalidates caches across cluster

sequenceDiagram participant N1 as Node 1 participant N2 as Node 2 participant N3 as Node 3 participant ZK as ZooKeeper Note over N1,N3: All nodes compete for leadership N1->>ZK: Create ephemeral node /election/node1 N2->>ZK: Create ephemeral node /election/node2 N3->>ZK: Create ephemeral node /election/node3 ZK-->>N1: You are leader (lowest sequence) ZK-->>N2: Watch /election/node1 ZK-->>N3: Watch /election/node2 Note over N1: Leader does work N1--xZK: Connection lost ZK->>N2: Leader changed (node1 gone) Note over N2: Becomes new leader

Raft-Based Leader Election

Raft is a consensus algorithm with built-in leader election. Used by etcd, Consul, and many distributed databases.

// Raft election process
public class RaftNode {
    private NodeState state = NodeState.Follower;
    private int currentTerm = 0;
    private string votedFor = null;
    private DateTime lastHeartbeat = DateTime.UtcNow;

    public async Task RunElection() {
        // Timeout waiting for heartbeat
        if (DateTime.UtcNow - lastHeartbeat > ElectionTimeout) {
            // Become candidate
            state = NodeState.Candidate;
            currentTerm++;
            votedFor = this.Id;
            int votesReceived = 1; // Vote for self

            // Request votes from other nodes
            foreach (var node in otherNodes) {
                var response = await node.RequestVote(new VoteRequest {
                    Term = currentTerm,
                    CandidateId = this.Id,
                    LastLogIndex = log.LastIndex,
                    LastLogTerm = log.LastTerm
                });

                if (response.VoteGranted) votesReceived++;
            }

            // Majority wins
            if (votesReceived > otherNodes.Count / 2) {
                BecomeLeader();
            }
        }
    }

    public VoteResponse RequestVote(VoteRequest request) {
        // Grant vote if:
        // 1. Request term >= current term
        // 2. Haven't voted for anyone else this term
        // 3. Candidate's log is at least as up-to-date as ours

        if (request.Term < currentTerm) {
            return new VoteResponse { VoteGranted = false, Term = currentTerm };
        }

        if (votedFor == null || votedFor == request.CandidateId) {
            if (request.LastLogIndex >= log.LastIndex) {
                votedFor = request.CandidateId;
                lastHeartbeat = DateTime.UtcNow; // Reset timeout
                return new VoteResponse { VoteGranted = true, Term = currentTerm };
            }
        }

        return new VoteResponse { VoteGranted = false, Term = currentTerm };
    }

    private void BecomeLeader() {
        state = NodeState.Leader;
        Console.WriteLine($"I am the leader for term {currentTerm}");

        // Send heartbeats to maintain leadership
        _ = Task.Run(async () => {
            while (state == NodeState.Leader) {
                foreach (var node in otherNodes) {
                    await node.AppendEntries(new HeartbeatRequest {
                        Term = currentTerm,
                        LeaderId = this.Id
                    });
                }
                await Task.Delay(HeartbeatInterval);
            }
        });
    }
}

ZooKeeper-Based Leader Election

ZooKeeper provides primitives for leader election using ephemeral sequential nodes:

public class ZooKeeperLeaderElection {
    private readonly ZooKeeper zk;
    private readonly string electionPath = "/election";
    private string myPath;

    public async Task<bool> TryBecomeLeader() {
        // Create ephemeral sequential node
        myPath = await zk.CreateAsync(
            $"{electionPath}/node-",
            Encoding.UTF8.GetBytes(nodeId),
            CreateMode.EphemeralSequential
        );

        // Get all children and sort
        var children = await zk.GetChildrenAsync(electionPath);
        children.Sort();

        // Check if we're the lowest
        var mySequence = GetSequence(myPath);
        var lowestSequence = GetSequence(children[0]);

        if (mySequence == lowestSequence) {
            Console.WriteLine("I am the leader!");
            return true;
        }

        // Not leader, watch the node before us
        var myIndex = children.FindIndex(c => GetSequence(c) == mySequence);
        var nodeToWatch = children[myIndex - 1];

        await zk.ExistsAsync($"{electionPath}/{nodeToWatch}", new Watcher(async (evt) => {
            if (evt.Type == EventType.NodeDeleted) {
                // Node before us died, check again
                await TryBecomeLeader();
            }
        }));

        Console.WriteLine($"Watching {nodeToWatch}");
        return false;
    }

    private int GetSequence(string path) {
        return int.Parse(path.Split('-').Last());
    }
}

etcd-Based Leader Election

etcd uses lease-based leader election with automatic renewal:

public class EtcdLeaderElection {
    private readonly EtcdClient client;
    private readonly string electionKey = "/election/leader";

    public async Task<bool> TryBecomeLeader(string nodeId) {
        // Create lease
        var lease = await client.LeaseGrantAsync(new LeaseGrantRequest {
            TTL = 10 // seconds
        });

        // Try to acquire leadership with CAS
        var txn = new TxnRequest();
        txn.Compare.Add(new Compare {
            Key = ByteString.CopyFromUtf8(electionKey),
            Target = Compare.Types.CompareTarget.Version,
            Result = Compare.Types.CompareResult.Equal,
            TargetUnion = 0 // Key doesn't exist
        });
        txn.Success.Add(new RequestOp {
            RequestPut = new PutRequest {
                Key = ByteString.CopyFromUtf8(electionKey),
                Value = ByteString.CopyFromUtf8(nodeId),
                Lease = lease.ID
            }
        });

        var response = await client.TransactionAsync(txn);

        if (response.Succeeded) {
            Console.WriteLine("I am the leader!");

            // Keep lease alive
            _ = client.LeaseKeepAlive(lease.ID, CancellationToken.None);
            return true;
        }

        // Not leader, watch for changes
        await client.WatchAsync(electionKey, async (evt) => {
            if (evt.Type == EventType.Delete) {
                await TryBecomeLeader(nodeId);
            }
        });

        return false;
    }
}

Comparison: Raft vs ZooKeeper vs etcd

Aspect	Raft (etcd/Consul)	ZooKeeper (ZAB)	etcd (v3 API)
Consensus Algorithm	Raft	ZAB (Zookeeper Atomic Broadcast)	Raft
Complexity	Lower	Higher	Lower
Performance	High	Good	High
API Style	Key-Value	File System-like	Key-Value + Lease
Ease of Use	Easy	Moderate	Easy
Maturity	Mature (Consul), Very Mature (etcd)	Very Mature	Very Mature

Interview Question: What happens if a leader crashes immediately after being elected?

Answer: Followers detect missing heartbeats and trigger new election. In Raft, election timeout expires and a new candidate emerges. In ZooKeeper/etcd, ephemeral node deletion or lease expiry triggers watchers, and next node becomes leader. Critical: design system so partial work by crashed leader doesn't corrupt state (use fencing tokens, idempotent operations).

Distributed Locks

Distributed locks coordinate access to shared resources across multiple nodes, ensuring mutual exclusion in distributed systems.

Redis-Based Locks (SETNX + TTL)

Simple locking using Redis SET with NX (only if not exists) and expiry:

public class RedisLock {
    private readonly IDatabase redis;

    public async Task<string> AcquireLock(string resource, int ttlSeconds = 30) {
        var lockKey = $"lock:{resource}";
        var lockValue = Guid.NewGuid().ToString(); // Unique token
        var acquired = await redis.StringSetAsync(
            lockKey,
            lockValue,
            TimeSpan.FromSeconds(ttlSeconds),
            When.NotExists
        );

        if (acquired) {
            return lockValue; // Return token to verify ownership
        }
        return null; // Lock held by someone else
    }

    public async Task ReleaseLock(string resource, string lockValue) {
        var lockKey = $"lock:{resource}";

        // Lua script for atomic check-and-delete
        var script = @"
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
        ";

        await redis.ScriptEvaluateAsync(script,
            new RedisKey[] { lockKey },
            new RedisValue[] { lockValue }
        );
    }

    // Usage
    public async Task ProcessWithLock(string resource) {
        var lockToken = await AcquireLock(resource, ttlSeconds: 30);

        if (lockToken == null) {
            Console.WriteLine("Could not acquire lock");
            return;
        }

        try {
            // Critical section
            await DoWork();
        } finally {
            await ReleaseLock(resource, lockToken);
        }
    }
}

ZooKeeper-Based Locks

More robust locking using ZooKeeper ephemeral nodes:

public class ZooKeeperLock {
    private readonly ZooKeeper zk;
    private readonly string lockPath;

    public async Task<string> AcquireLock(string resource) {
        var lockDir = $"/locks/{resource}";

        // Ensure lock directory exists
        await EnsurePath(lockDir);

        // Create ephemeral sequential node
        var myPath = await zk.CreateAsync(
            $"{lockDir}/lock-",
            new byte[0],
            CreateMode.EphemeralSequential
        );

        while (true) {
            var children = await zk.GetChildrenAsync(lockDir);
            children.Sort();

            var mySequence = GetSequence(myPath);
            var lowestSequence = GetSequence(children[0]);

            if (mySequence == lowestSequence) {
                // I have the lock
                return myPath;
            }

            // Wait for the node before me
            var myIndex = children.FindIndex(c => GetSequence(c) == mySequence);
            var nodeToWatch = children[myIndex - 1];

            var tcs = new TaskCompletionSource();
            await zk.ExistsAsync($"{lockDir}/{nodeToWatch}", new Watcher(evt => {
                if (evt.Type == EventType.NodeDeleted) {
                    tcs.SetResult(true);
                }
            }));

            // Wait for notification
            await tcs.Task;
        }
    }

    public async Task ReleaseLock(string lockPath) {
        await zk.DeleteAsync(lockPath);
    }
}

Redlock Algorithm (Controversial)

Redis's distributed locking algorithm across multiple Redis instances. Created to handle single-Redis failures, but controversial:

public class Redlock {
    private readonly List<IDatabase> redisInstances;

    public async Task<RedlockToken> AcquireLock(string resource, int ttlMs = 10000) {
        var lockValue = Guid.NewGuid().ToString();
        var startTime = DateTime.UtcNow;
        int successCount = 0;

        // Try to acquire lock on majority of instances
        foreach (var redis in redisInstances) {
            try {
                var acquired = await redis.StringSetAsync(
                    $"lock:{resource}",
                    lockValue,
                    TimeSpan.FromMilliseconds(ttlMs),
                    When.NotExists
                );
                if (acquired) successCount++;
            } catch {
                // Instance unavailable, continue
            }
        }

        var elapsed = (DateTime.UtcNow - startTime).TotalMilliseconds;
        var validity = ttlMs - elapsed - CLOCK_DRIFT_FACTOR;

        // Check if we got majority and lock still valid
        if (successCount >= (redisInstances.Count / 2 + 1) && validity > 0) {
            return new RedlockToken {
                Resource = resource,
                Value = lockValue,
                ValidityMs = (int)validity
            };
        }

        // Failed to acquire, release any acquired locks
        await ReleaseLock(resource, lockValue);
        return null;
    }

    public async Task ReleaseLock(string resource, string lockValue) {
        var script = @"
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
        ";

        foreach (var redis in redisInstances) {
            try {
                await redis.ScriptEvaluateAsync(script,
                    new RedisKey[] { $"lock:{resource}" },
                    new RedisValue[] { lockValue }
                );
            } catch {
                // Best effort
            }
        }
    }
}

Redlock Controversy

Martin Kleppmann criticized Redlock for unsafe assumptions about timing and clock synchronization. Key issues:

Clock jumps can cause lock holder to think lock expired when it hasn't
GC pauses can cause process to hold expired lock
Network partitions can create split-brain scenarios

Recommendation: Use ZooKeeper/etcd for critical locks, or use fencing tokens (see next section) with any locking system.

Interview Question: Why is TTL necessary for distributed locks?

Answer: TTL prevents deadlocks from crashed/stuck clients that never release locks. Without TTL, lock would be held forever. TTL must be longer than maximum expected critical section duration, but short enough to not block other clients too long. Challenge: if client runs slow (GC pause), TTL expires but client thinks it still holds lock. Solution: use fencing tokens to detect stale lock holders.

Fencing Tokens

Fencing tokens prevent zombie processes from corrupting state. A zombie is a process that holds an expired lock but doesn't realize it.

The Zombie Problem

sequenceDiagram participant C1 as Client 1 participant Lock as Lock Service participant C2 as Client 2 participant Store as Storage C1->>Lock: Acquire lock (token=1) Lock-->>C1: Success (token=1) Note over C1: GC pause for 35 seconds Note over Lock: Lock TTL expires (30s) C2->>Lock: Acquire lock Lock-->>C2: Success (token=2) C2->>Store: Write data (token=2) Store-->>C2: OK Note over C1: Wakes from GC pause C1->>Store: Write data (token=1) ❌ Note over Store: Reject! Token 1 < current token 2

Implementation

public class FencedLock {
    private readonly ILockService lockService;
    private long currentToken = 0;

    // Lock service returns monotonically increasing token
    public async Task<FencingToken> AcquireLock(string resource) {
        currentToken = Interlocked.Increment(ref currentToken);

        var acquired = await lockService.TryAcquire(resource);
        if (acquired) {
            return new FencingToken {
                Resource = resource,
                Token = currentToken,
                AcquiredAt = DateTime.UtcNow
            };
        }
        return null;
    }
}

// Storage service validates fencing tokens
public class FencedStorage {
    private readonly ConcurrentDictionary<string, long> maxTokens = new();

    public async Task<bool> Write(string key, string value, long fencingToken) {
        // Check if token is higher than any we've seen
        var maxToken = maxTokens.GetOrAdd(key, 0);

        if (fencingToken <= maxToken) {
            // Reject stale write
            throw new StaleTokenException(
                $"Token {fencingToken} is stale. Current: {maxToken}"
            );
        }

        // Update max token and perform write
        maxTokens[key] = fencingToken;
        await storage.WriteAsync(key, value);
        return true;
    }
}

// Usage
public class DistributedWorker {
    public async Task ProcessWithFencing(string taskId) {
        var token = await fencedLock.AcquireLock($"task:{taskId}");
        if (token == null) {
            Console.WriteLine("Could not acquire lock");
            return;
        }

        try {
            // Do work
            var result = await Process(taskId);

            // Write result with fencing token
            await storage.Write(
                $"result:{taskId}",
                result,
                token.Token // Include token
            );
        } catch (StaleTokenException) {
            Console.WriteLine("Lock expired, write rejected");
        }
    }
}

Fencing Token Sources

Where to get monotonically increasing tokens:

ZooKeeper: Use zxid (transaction ID) or node version
etcd: Use lease ID or key modification revision
Database: Sequence or auto-increment column
Raft: Use log index or term number

// Using ZooKeeper version as fencing token
public async Task<long> AcquireLockWithToken(string resource) {
    var lockPath = await zk.CreateAsync(
        $"/locks/{resource}",
        new byte[0],
        CreateMode.Ephemeral
    );

    var stat = await zk.GetDataAsync(lockPath);
    return stat.Version; // Use version as fencing token
}

// Using etcd lease ID as fencing token
public async Task<long> AcquireLockWithToken(string resource) {
    var lease = await etcd.LeaseGrantAsync(new LeaseGrantRequest { TTL = 30 });

    await etcd.PutAsync(new PutRequest {
        Key = $"/locks/{resource}",
        Value = nodeId,
        Lease = lease.ID
    });

    return lease.ID; // Use lease ID as fencing token
}

Interview Question: Can fencing tokens prevent all timing issues in distributed systems?

Answer: Fencing tokens prevent stale writes from zombie processes, but don't solve all timing issues. They require storage system support (reject writes with old tokens). They don't prevent: race conditions between different resources, read-modify-write races (need optimistic locking/CAS), or issues from clock skew. Fencing is one tool, not a complete solution. Combine with proper distributed algorithms and eventual consistency patterns.

Distributed Barriers

Distributed barriers synchronize N processes, ensuring all reach a point before any proceed further. Like threading barriers, but across machines.

Use Cases

MapReduce: wait for all mappers before starting reducers
Distributed testing: synchronize test execution across machines
Coordinated deployment: ensure all nodes ready before switching traffic
Batch processing: wait for all partitions to complete

sequenceDiagram participant W1 as Worker 1 participant W2 as Worker 2 participant W3 as Worker 3 participant ZK as ZooKeeper W1->>ZK: Create /barrier/worker1 W2->>ZK: Create /barrier/worker2 Note over W1,W2: Workers wait W3->>ZK: Create /barrier/worker3 Note over ZK: All 3 workers present ZK->>W1: Barrier satisfied ZK->>W2: Barrier satisfied ZK->>W3: Barrier satisfied Note over W1,W3: All proceed

Implementation with ZooKeeper

public class DistributedBarrier {
    private readonly ZooKeeper zk;
    private readonly string barrierPath;
    private readonly int requiredNodes;

    public DistributedBarrier(ZooKeeper zk, string name, int requiredNodes) {
        this.zk = zk;
        this.barrierPath = $"/barriers/{name}";
        this.requiredNodes = requiredNodes;
    }

    public async Task Enter(string nodeId) {
        // Ensure barrier directory exists
        await EnsurePath(barrierPath);

        // Register this node
        await zk.CreateAsync(
            $"{barrierPath}/{nodeId}",
            new byte[0],
            CreateMode.Ephemeral
        );

        Console.WriteLine($"{nodeId} entered barrier");

        // Wait until all nodes present
        while (true) {
            var children = await zk.GetChildrenAsync(barrierPath);

            if (children.Count >= requiredNodes) {
                Console.WriteLine($"{nodeId} barrier satisfied, proceeding");
                return;
            }

            // Watch for changes
            var tcs = new TaskCompletionSource();
            await zk.GetChildrenAsync(barrierPath, new Watcher(evt => {
                if (evt.Type == EventType.NodeChildrenChanged) {
                    tcs.SetResult(true);
                }
            }));

            await tcs.Task;
        }
    }

    public async Task Leave(string nodeId) {
        // Remove this node
        await zk.DeleteAsync($"{barrierPath}/{nodeId}");
    }
}

// Usage in distributed workers
public class DistributedWorker {
    public async Task ExecuteWithBarrier() {
        var barrier = new DistributedBarrier(zk, "phase1", requiredWorkers: 5);

        // Phase 1 work
        await DoPhase1Work();

        // Wait for all workers to complete phase 1
        await barrier.Enter(this.WorkerId);

        // Phase 2 work (only starts when all workers ready)
        await DoPhase2Work();

        await barrier.Leave(this.WorkerId);
    }
}

Double Barrier

Ensures all nodes enter AND leave together (useful for synchronized cleanup):

public class DoubleBarrier {
    public async Task Enter(string nodeId) {
        await zk.CreateAsync($"{barrierPath}/{nodeId}", new byte[0], CreateMode.Ephemeral);

        // Wait for all to enter
        while (true) {
            var children = await zk.GetChildrenAsync(barrierPath);
            if (children.Count >= requiredNodes) break;
            await WatchForChanges();
        }
    }

    public async Task Leave(string nodeId) {
        // Don't delete yet, just signal ready to leave
        await zk.SetDataAsync($"{barrierPath}/{nodeId}", Encoding.UTF8.GetBytes("ready"));

        // Wait for all to signal ready
        while (true) {
            var children = await zk.GetChildrenAsync(barrierPath);
            var allReady = true;

            foreach (var child in children) {
                var data = await zk.GetDataAsync($"{barrierPath}/{child}");
                if (Encoding.UTF8.GetString(data.Data) != "ready") {
                    allReady = false;
                    break;
                }
            }

            if (allReady) break;
            await WatchForChanges();
        }

        // All ready, now delete
        await zk.DeleteAsync($"{barrierPath}/{nodeId}");
    }
}

Interview Question: What happens if a node crashes while at a barrier?

Answer: With ephemeral nodes (ZooKeeper), crashed node's registration is automatically deleted when connection lost, but other nodes still wait forever if barrier requires exact count. Solutions: (1) Use timeout - proceed after T seconds even if not all present, (2) Track crashes and reduce required count, (3) Use double-check pattern - verify node is actually alive before counting, (4) Have coordinator that can declare barrier satisfied with N-1 nodes if one confirmed dead.

Sequence Numbers & Distributed Counters

Generate globally unique, ordered identifiers in distributed systems. Useful for ordering events, generating IDs, and coordinating work.

ZooKeeper Sequential Nodes

ZooKeeper's sequential flag automatically appends monotonically increasing number:

public class DistributedSequence {
    private readonly ZooKeeper zk;
    private readonly string sequencePath;

    public async Task<long> NextValue() {
        // Create ephemeral sequential node
        var path = await zk.CreateAsync(
            $"{sequencePath}/seq-",
            new byte[0],
            CreateMode.EphemeralSequential
        );

        // Extract sequence number from path
        // e.g., /sequences/myseq/seq-0000000123
        var sequence = long.Parse(path.Split('-').Last());

        // Optionally delete node (we only needed the number)
        await zk.DeleteAsync(path);

        return sequence;
    }
}

// Usage: generate order IDs
public async Task<string> CreateOrder() {
    var orderNumber = await sequence.NextValue();
    var order = new Order {
        Id = $"ORD-{orderNumber:D10}",
        CreatedAt = DateTime.UtcNow
    };
    await SaveOrder(order);
    return order.Id;
}

Distributed Counter

Implement counter with atomic increments across nodes:

public class DistributedCounter {
    private readonly ZooKeeper zk;
    private readonly string counterPath;

    public async Task<long> Increment(int delta = 1) {
        while (true) {
            // Read current value and version
            var result = await zk.GetDataAsync(counterPath);
            var currentValue = BitConverter.ToInt64(result.Data, 0);
            var version = result.Stat.Version;

            // Try to increment with version check (optimistic locking)
            var newValue = currentValue + delta;
            try {
                await zk.SetDataAsync(
                    counterPath,
                    BitConverter.GetBytes(newValue),
                    version
                );
                return newValue;
            } catch (KeeperException.BadVersionException) {
                // Someone else modified it, retry
                continue;
            }
        }
    }

    public async Task<long> GetValue() {
        var result = await zk.GetDataAsync(counterPath);
        return BitConverter.ToInt64(result.Data, 0);
    }
}

// Redis-based counter (simpler, uses atomic INCR)
public class RedisCounter {
    private readonly IDatabase redis;

    public async Task<long> Increment(string key, int delta = 1) {
        return await redis.StringIncrementAsync(key, delta);
    }

    public async Task<long> GetValue(string key) {
        var value = await redis.StringGetAsync(key);
        return (long)value;
    }
}

Snowflake IDs

Twitter's Snowflake generates sortable, unique 64-bit IDs without coordination:

public class SnowflakeIdGenerator {
    // 64-bit ID structure:
    // 1 bit unused
    // 41 bits timestamp (milliseconds since epoch)
    // 10 bits machine ID
    // 12 bits sequence number

    private const long Twepoch = 1288834974657L; // Twitter epoch
    private const int MachineIdBits = 10;
    private const int SequenceBits = 12;
    private const long MaxMachineId = -1L ^ (-1L << MachineIdBits); // 1023
    private const long MaxSequence = -1L ^ (-1L << SequenceBits); // 4095

    private readonly long machineId;
    private long sequence = 0L;
    private long lastTimestamp = -1L;

    public SnowflakeIdGenerator(long machineId) {
        if (machineId > MaxMachineId || machineId < 0) {
            throw new ArgumentException($"Machine ID must be 0-{MaxMachineId}");
        }
        this.machineId = machineId;
    }

    public long NextId() {
        lock (this) {
            var timestamp = GetTimestamp();

            if (timestamp < lastTimestamp) {
                throw new InvalidOperationException("Clock moved backwards!");
            }

            if (timestamp == lastTimestamp) {
                // Same millisecond, increment sequence
                sequence = (sequence + 1) & MaxSequence;
                if (sequence == 0) {
                    // Sequence overflow, wait for next millisecond
                    timestamp = WaitNextMillis(lastTimestamp);
                }
            } else {
                // New millisecond, reset sequence
                sequence = 0;
            }

            lastTimestamp = timestamp;

            // Compose ID
            return ((timestamp - Twepoch) << (MachineIdBits + SequenceBits))
                   | (machineId << SequenceBits)
                   | sequence;
        }
    }

    private long GetTimestamp() {
        return DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
    }

    private long WaitNextMillis(long lastTimestamp) {
        var timestamp = GetTimestamp();
        while (timestamp <= lastTimestamp) {
            timestamp = GetTimestamp();
        }
        return timestamp;
    }
}

// Usage: No coordination needed!
var generator = new SnowflakeIdGenerator(machineId: 42);
var id = generator.NextId(); // 280445456304152576

Interview Question: Compare distributed sequence generation approaches.

Answer: ZooKeeper sequential: Strongly consistent, globally ordered, but requires coordination (slower). Database sequences: Consistent within partition, but single point of contention. Snowflake IDs: No coordination needed (very fast), roughly time-ordered, but requires machine ID management and clock synchronization. UUIDs: No coordination, but not ordered, larger (128-bit). Choose based on needs: strict ordering (ZK), performance (Snowflake), or simplicity (UUID).

Lease-Based Systems

Leases provide time-bounded ownership of resources. After lease expires, ownership automatically reverts without requiring explicit release.

Lease Concepts

Time-Bounded: Lease expires after fixed duration
Renewal: Holder can extend lease before expiry
Automatic Cleanup: No orphaned resources if holder crashes
Grace Period: Brief window after expiry before reclaim

sequenceDiagram participant Client participant LeaseService participant Resource Client->>LeaseService: Acquire lease (TTL=30s) LeaseService-->>Client: Lease granted (ID=123) Client->>Resource: Access resource (lease=123) Note over Client: Work for 20s Client->>LeaseService: Renew lease (ID=123) LeaseService-->>Client: Renewed (expires +30s) Note over Client: More work Note over Client: Crash / Network partition Note over LeaseService: Lease expires after 30s LeaseService->>Resource: Revoke access (lease=123) Note over Resource: Available for new lease

Implementation

public class LeaseService {
    private readonly ConcurrentDictionary<string, Lease> leases = new();

    public async Task<Lease> AcquireLease(string resource, int ttlSeconds) {
        var lease = new Lease {
            Id = Guid.NewGuid().ToString(),
            Resource = resource,
            ExpiresAt = DateTime.UtcNow.AddSeconds(ttlSeconds),
            TtlSeconds = ttlSeconds
        };

        if (leases.TryAdd(resource, lease)) {
            StartExpiryTimer(lease);
            return lease;
        }

        // Resource already leased
        return null;
    }

    public async Task<bool> RenewLease(string leaseId) {
        var lease = leases.Values.FirstOrDefault(l => l.Id == leaseId);
        if (lease == null) return false;

        // Extend expiry
        lease.ExpiresAt = DateTime.UtcNow.AddSeconds(lease.TtlSeconds);
        return true;
    }

    public async Task ReleaseLease(string leaseId) {
        var lease = leases.Values.FirstOrDefault(l => l.Id == leaseId);
        if (lease != null) {
            leases.TryRemove(lease.Resource, out _);
        }
    }

    private void StartExpiryTimer(Lease lease) {
        _ = Task.Run(async () => {
            while (DateTime.UtcNow < lease.ExpiresAt) {
                await Task.Delay(1000);
            }

            // Lease expired
            if (leases.TryRemove(lease.Resource, out _)) {
                OnLeaseExpired(lease);
            }
        });
    }

    private void OnLeaseExpired(Lease lease) {
        Console.WriteLine($"Lease {lease.Id} for {lease.Resource} expired");
        // Trigger cleanup, notify waiters, etc.
    }
}

// Client with automatic renewal
public class LeaseHolder {
    private readonly LeaseService leaseService;
    private Lease currentLease;
    private CancellationTokenSource renewalCts;

    public async Task<bool> AcquireAndMaintain(string resource) {
        currentLease = await leaseService.AcquireLease(resource, ttlSeconds: 30);
        if (currentLease == null) return false;

        // Start renewal loop
        renewalCts = new CancellationTokenSource();
        _ = Task.Run(async () => await RenewalLoop(), renewalCts.Token);

        return true;
    }

    private async Task RenewalLoop() {
        // Renew at 2/3 of TTL
        var renewInterval = TimeSpan.FromSeconds(currentLease.TtlSeconds * 2 / 3);

        while (!renewalCts.Token.IsCancellationRequested) {
            await Task.Delay(renewInterval, renewalCts.Token);

            var renewed = await leaseService.RenewLease(currentLease.Id);
            if (!renewed) {
                Console.WriteLine("Failed to renew lease!");
                break;
            }
        }
    }

    public async Task Release() {
        renewalCts?.Cancel();
        if (currentLease != null) {
            await leaseService.ReleaseLease(currentLease.Id);
        }
    }
}

Lease Expiry Detection

Holder must detect when lease expires and stop using resource:

public class LeaseAwareWorker {
    private Lease currentLease;

    public async Task DoWork() {
        while (true) {
            // Check lease still valid before each operation
            if (DateTime.UtcNow >= currentLease.ExpiresAt) {
                Console.WriteLine("Lease expired, stopping work");
                return;
            }

            // Perform small unit of work
            await ProcessItem();

            // Check again after work
            if (DateTime.UtcNow >= currentLease.ExpiresAt) {
                Console.WriteLine("Lease expired during work, discarding results");
                return;
            }

            // Commit results
            await CommitResults();
        }
    }
}

// Or use cancellation token tied to lease
public class TokenBasedLeaseWorker {
    public async Task DoWork(Lease lease, CancellationToken leaseToken) {
        try {
            while (!leaseToken.IsCancellationRequested) {
                await ProcessItem(leaseToken);
                await CommitResults(leaseToken);
            }
        } catch (OperationCanceledException) {
            Console.WriteLine("Lease expired, cleaning up");
        }
    }
}

Real-World Example: GFS Chunk Leases

Google File System uses leases for chunk replicas. Master grants lease to one replica (primary), which orders all writes. Lease expires after 60s, renewable.

Interview Question: What's the trade-off between short and long lease durations?

Answer: Short leases (1-10s): Fast failover (new holder can take over quickly), but more renewal overhead, more sensitive to network hiccups. Long leases (60s+): Less overhead, more resilient to temporary issues, but slow failover. Sweet spot often 30s: balance of low overhead and reasonable failover time. Also consider: use shorter leases for critical resources, longer for stable background work. Adaptive leases can adjust TTL based on renewal success rate.

Coordination Avoidance

Coordination is expensive in distributed systems (latency, availability, scalability). Minimize coordination for higher throughput.

Why Coordination is Expensive

Latency: Consensus requires multiple network round-trips
Availability: Coordination service outage blocks all operations
Scalability: Coordination is a bottleneck (all nodes wait)
Complexity: More moving parts, more failure modes

graph TB subgraph "With Coordination" A1[Node 1] --> C1[Coordinator] A2[Node 2] --> C1 A3[Node 3] --> C1 C1 --> D1[Decision] end subgraph "Without Coordination" B1[Node 1] --> E1[Local Decision] B2[Node 2] --> E2[Local Decision] B3[Node 3] --> E3[Local Decision] end style E1 fill:#90EE90 style E2 fill:#90EE90 style E3 fill:#90EE90

Coordination-Free Techniques

1. CRDTs (Conflict-Free Replicated Data Types)

Data structures that can be updated concurrently without coordination and always converge:

// G-Counter (Grow-only counter)
public class GCounter {
    private Dictionary<string, int> counts = new();

    public void Increment(string nodeId) {
        counts[nodeId] = counts.GetValueOrDefault(nodeId, 0) + 1;
    }

    public int GetValue() {
        return counts.Values.Sum();
    }

    public void Merge(GCounter other) {
        foreach (var kvp in other.counts) {
            counts[kvp.Key] = Math.Max(
                counts.GetValueOrDefault(kvp.Key, 0),
                kvp.Value
            );
        }
    }
}

// OR-Set (Add-wins set)
public class ORSet<T> {
    private Dictionary<T, HashSet<Guid>> elements = new();

    public void Add(T element) {
        var tag = Guid.NewGuid();
        if (!elements.ContainsKey(element)) {
            elements[element] = new HashSet<Guid>();
        }
        elements[element].Add(tag);
    }

    public void Remove(T element) {
        if (elements.ContainsKey(element)) {
            elements[element].Clear();
        }
    }

    public bool Contains(T element) {
        return elements.ContainsKey(element) && elements[element].Any();
    }

    public void Merge(ORSet<T> other) {
        foreach (var kvp in other.elements) {
            if (!elements.ContainsKey(kvp.Key)) {
                elements[kvp.Key] = new HashSet<Guid>();
            }
            elements[kvp.Key].UnionWith(kvp.Value);
        }
    }
}

2. Optimistic Concurrency Control

Assume no conflicts, detect them after the fact:

public class OptimisticUpdate {
    public async Task<bool> UpdateAccount(string accountId, decimal amount) {
        while (true) {
            // Read current state + version
            var account = await db.GetAccount(accountId);
            var currentVersion = account.Version;

            // Perform business logic
            account.Balance += amount;
            account.Version++;

            // Try to update with version check
            var updated = await db.ExecuteAsync(@"
                UPDATE accounts
                SET balance = @balance, version = @version
                WHERE id = @id AND version = @currentVersion",
                new { id = accountId, balance = account.Balance,
                      version = account.Version, currentVersion }
            );

            if (updated > 0) {
                return true; // Success
            }

            // Conflict detected, retry
            await Task.Delay(Random.Shared.Next(10, 100)); // Backoff
        }
    }
}

3. Partitioning to Avoid Coordination

Partition data so each partition can operate independently:

// User accounts partitioned by user ID
// Each partition can update counters independently
public class PartitionedCounter {
    public async Task IncrementUserCounter(string userId, string metric) {
        var partition = GetPartition(userId);

        // No cross-partition coordination needed
        await partition.Increment($"user:{userId}:{metric}");
    }

    private IPartition GetPartition(string userId) {
        var hash = userId.GetHashCode();
        var partitionId = Math.Abs(hash) % totalPartitions;
        return partitions[partitionId];
    }
}

// Aggregation happens later (eventually consistent)
public async Task<long> GetTotalCount(string metric) {
    var tasks = partitions.Select(p => p.GetCount(metric));
    var counts = await Task.WhenAll(tasks);
    return counts.Sum();
}

4. Eventual Consistency

Accept temporary inconsistency, guarantee eventual convergence:

// Each node maintains local view, syncs periodically
public class EventuallyConsistentCache {
    private ConcurrentDictionary<string, CacheEntry> localCache = new();

    public void Set(string key, string value) {
        localCache[key] = new CacheEntry {
            Value = value,
            Timestamp = DateTime.UtcNow,
            NodeId = this.NodeId
        };

        // Asynchronously propagate to other nodes
        _ = PropagateToOthers(key, value);
    }

    public string Get(string key) {
        // Return local value (might be stale)
        return localCache.TryGetValue(key, out var entry) ? entry.Value : null;
    }

    private async Task PropagateToOthers(string key, string value) {
        foreach (var node in otherNodes) {
            try {
                await node.Merge(key, value, this.NodeId, DateTime.UtcNow);
            } catch {
                // Best effort, eventual consistency
            }
        }
    }

    public void Merge(string key, string value, string nodeId, DateTime timestamp) {
        // Last-write-wins conflict resolution
        if (!localCache.TryGetValue(key, out var existing) ||
            timestamp > existing.Timestamp) {
            localCache[key] = new CacheEntry {
                Value = value,
                Timestamp = timestamp,
                NodeId = nodeId
            };
        }
    }
}

When You Can't Avoid Coordination

Some operations inherently require coordination:

Unique Constraints: Ensuring email uniqueness across all users
Global Ordering: Strict event sequencing
Atomic Multi-Object Updates: Transfer between accounts
Consensus: All nodes must agree on a value

Strategy: Minimize coordination surface area. Push coordination to edges, keep core coordination-free.

Interview Question: How would you build a distributed counter that supports both increments and decrements without coordination?

Answer: Use PN-Counter (Positive-Negative Counter) CRDT: each node maintains two G-Counters (increment-only), one for increments, one for decrements. Value = sum(increments) - sum(decrements). To increment: add to increment counter on local node. To decrement: add to decrement counter on local node. Merge: merge both G-Counters independently. No coordination needed, eventual consistency guaranteed. Limitation: can't enforce lower bound (e.g., prevent negative balance) without coordination.

Summary

Consensus and coordination are essential but expensive in distributed systems. Key takeaways:

Leader Election: Use Raft (etcd/Consul), ZooKeeper, or etcd leases. Prefer simpler systems when possible.
Distributed Locks: Redis for simple cases, ZooKeeper for robustness. Always use TTL to prevent deadlocks.
Fencing Tokens: Prevent zombie processes from corrupting state with monotonically increasing tokens.
Barriers: Synchronize N processes using ephemeral nodes and watches.
Sequences/Counters: ZooKeeper sequential nodes for ordering, Snowflake IDs for coordination-free generation.
Leases: Time-bounded ownership with automatic cleanup. Renew before expiry.
Avoid Coordination: Use CRDTs, optimistic concurrency, partitioning, and eventual consistency when possible.

The art of distributed systems: minimize coordination while maintaining necessary consistency guarantees. Coordination is a powerful tool but should be used sparingly for performance and availability.