Design a Distributed Key-Value Store

// Python 3.10+ Async Quorum Consensus Coordinator
import asyncio

class CoordinatorNode:
    def __init__(self, ring, replication_factor: int = 3, write_quorum: int = 2):
        self.ring = ring
        self.N = replication_factor
        self.W = write_quorum

    async def put(self, key: str, value: str) -> bool:
        # 1. Locate next N distinct replica nodes on hash ring
        replicas = self.ring.get_preference_list(key, self.N)
        
        # 2. Fire concurrent asynchronous writes to all N replicas
        tasks = [asyncio.create_task(node.write(key, value)) for node in replicas]
        
        acks = 0
        for completed_task in asyncio.as_completed(tasks, timeout=2.0):
            try:
                success = await completed_task
                if success: acks += 1
                # Quorum reached! Return success without waiting for slowest replica
                if acks >= self.W: return True
            except Exception: continue
            
        raise TimeoutError("Quorum consensus write failed to meet W threshold")

// C++20 Quorum Coordinator
#include <vector>
#include <future>
#include <string>

class Coordinator {
    int N = 3, W = 2;
public:
    bool put(const std::string& key, const std::string& value, const std::vector<Node*>& replicas) {
        std::vector<std::future<bool>> futures;
        for (Node* node : replicas) {
            futures.push_back(std::async(std::launch::async, [=]() {
                return node->write(key, value);
            }));
        }
        int acks = 0;
        for (auto& fut : futures) {
            if (fut.get()) acks++;
            if (acks >= W) return true; // Quorum satisfied
        }
        return false;
    }
};

// Java 17+ CompletableFuture Quorum Router
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class CoordinatorNode {
    private final int N = 3, W = 2;

    public boolean put(String key, String value, List<StorageNode> replicas) throws Exception {
        AtomicInteger acks = new AtomicInteger(0);
        CountDownLatch latch = new CountDownLatch(W);

        for (StorageNode node : replicas) {
            CompletableFuture.supplyAsync(() -> node.write(key, value))
                .thenAccept(success -> {
                    if (success && acks.incrementAndGet() <= W) {
                        latch.countDown();
                    }
                });
        }
        return latch.await(2, TimeUnit.SECONDS);
    }
}

// Go 1.20+ Channel-based Quorum Consensus
package main

import (
    "context"
    "time"
)

func (c *Coordinator) Put(ctx context.Context, key, val string, replicas []Node) bool {
    ackChan := make(chan bool, len(replicas))
    
    for _, node := range replicas {
        go func(n Node) {
            ackChan <- n.Write(key, val)
        }(node)
    }

    acks := 0
    timeout := time.After(2 * time.Second)
    for i := 0; i < len(replicas); i++ {
        select {
        case ok := <-ackChan:
            if ok { acks++ }
            if acks >= c.W { return true }
        case <-timeout:
            return false
        }
    }
    return false
}

// TypeScript / Node.js Promise Quorum Aggregator
export class CoordinatorNode {
    constructor(private N = 3, private W = 2) {}

    async put(key: string, val: string, replicas: StorageNode[]): Promise<boolean> {
        return new Promise((resolve, reject) => {
            let acks = 0;
            const timer = setTimeout(() => reject(new Error("Quorum timeout")), 2000);

            replicas.forEach(async (node) => {
                try {
                    const ok = await node.write(key, val);
                    if (ok && ++acks === this.W) {
                        clearTimeout(timer);
                        resolve(true);
                    }
                } catch (e) { /* retry or hinted handoff */ }
            });
        });
    }
}