Javaraft: فروشگاه ارزش کلیدی توزیع شده مبتنی بر قایق

// If election timeout elapses without receiving AppendEntries RPC from current leader or granting vote to candidate:
// convert to candidate (§5.2)
public void reset() {
    cancel();

    long minTimeout = config.getElectionTimeoutMillisMin();
    long maxTimeout = config.getElectionTimeoutMillisMax();
    long timeout = minTimeout + random.nextInt((int)(maxTimeout - minTimeout));

    electionFuture = scheduler.schedule(() -> {
        electionManager.startElection();
    }, timeout, TimeUnit.MILLISECONDS);
}

/**
 * On conversion to candidate, start election:
 * Increment currentTerm, vote for self, reset election timer, send RequestVote RPCs to all other servers.
 */
public void startElection() {
    lockManager.getStateWriteLock().lock();
    lockManager.getLogReadLock().lock();
    try {
        if (nodeState.getCurrentRole() == Role.LEADER) {
            return;
        }

        nodeState.setCurrentRole(Role.CANDIDATE);
        nodeState.incrementTerm();
        nodeState.setVotedFor(nodeState.getNodeId());

        int currentTerm = nodeState.getCurrentTerm();
        int lastLogIndex = raftLog.getLastIndex();
        int lastLogTerm = raftLog.getLastTerm();

        List<CompletableFuture<VoteResponseDto>> voteFutures = new ArrayList<>();
        ExecutorService executor = Executors.newCachedThreadPool();

        for (String peerUrl : config.getPeerUrls().values()) {
            CompletableFuture<VoteResponseDto> voteFuture = CompletableFuture
                    .supplyAsync(() -> requestVote(
                            currentTerm,
                            nodeState.getNodeId(),
                            lastLogIndex,
                            lastLogTerm,
                            peerUrl
                    ), executor)
                    .orTimeout(config.getElectionRpcTimeoutMillis(), TimeUnit.MILLISECONDS)
                    .exceptionally(throwable -> {
                        return new VoteResponseDto(currentTerm, false);
                    });
            voteFutures.add(voteFuture);
        }

        int majority = (config.getPeerUrls().size() + 1) / 2 + 1;
        AtomicInteger voteCount = new AtomicInteger(1);

        // If votes received from majority of servers: become leader (§5.2).
        for (CompletableFuture<VoteResponseDto> future : voteFutures) {
            future.thenAccept(response -> {
                lockManager.getStateWriteLock().lock();
                try {
                    if (nodeState.getCurrentRole() != Role.CANDIDATE || nodeState.getCurrentTerm() != currentTerm) {
                        return;
                    }
                    if (response != null && response.isVoteGranted()) {
                        int newVoteCount = voteCount.incrementAndGet();
                        if (newVoteCount >= majority) {
                            stateManager.becomeLeader();
                        }
                    }
                } finally {
                    lockManager.getStateWriteLock().unlock();
                }
            });
        }
        stateManager.resetElectionTimer();
    } finally {
        lockManager.getLogReadLock().unlock();
        lockManager.getStateWriteLock().unlock();
    }
}

...
public class RaftReplicationManager {
    ...

    public boolean handleClientRequest(LogEntry entry) {
        return clientRequestHandler.handle(entry);
    }

    public void startLogReplication() {
        logReplicator.start();
    }

    public AppendEntriesResponseDTO handleAppendEntries(AppendEntriesDTO dto) {
        return appendEntriesHandler.handle(dto);
    }

    public void initializeIndices() {
        logReplicator.initializeIndices();
    }
}

// Upon election: send initial empty AppendEntries RPCs (heartbeat) to each server; repeat during idle periods
// to prevent election timeouts (§5.2)
public void start() {
    lockManager.getStateReadLock().lock();
    try {
        if (nodeState.getCurrentRole() != Role.LEADER) return;
    } finally {
        lockManager.getStateReadLock().unlock();
    }

    for (String peer : config.getPeerUrls().values()) {
        if (!pendingReplication.getOrDefault(peer, false)) {
            pendingReplication.put(peer, true);
            executor.submit(() -> replicateLoop(peer));
        }
    }
}

// Sends heartbeats/log entries to peer, adjusting sleep time to match heartbeat interval
private void replicateLoop(String peer) {
    int backoff = config.getHeartbeatIntervalMillis();
    while (true) {
        lockManager.getStateReadLock().lock();
        try {
            if (nodeState.getCurrentRole() != Role.LEADER) break;
        } finally {
            lockManager.getStateReadLock().unlock();
        }

        lockManager.getLogWriteLock().lock();
        lockManager.getStateWriteLock().lock();
        lockManager.getStateMachineWriteLock().lock();
        long startTime = System.currentTimeMillis();
        try {
            boolean ok = replicate(peer);
            if (ok) updateCommitIndex();
        } finally {
            lockManager.getStateMachineWriteLock().unlock();
            lockManager.getStateWriteLock().unlock();
            lockManager.getLogWriteLock().unlock();
        }
        long duration = System.currentTimeMillis() - startTime;
        long sleepTime = Math.max(0, backoff - duration);
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        }
    }
    pendingReplication.put(peer, false);
}

// If last log index >= nextIndex for a follower: send AppendEntries RPC with log entries starting at nextIndex.
private boolean replicate(String peer) {
    int ni = nextIndex.get(peer);
    int prevIdx = ni - 1;
    int prevTerm = (prevIdx >= 0) ? raftLog.getTermAt(prevIdx) : 0;
    List<LogEntry> entries = raftLog.getEntriesFrom(ni);

    AppendEntriesRequestDto dto = new AppendEntriesRequestDto(
            nodeState.getCurrentTerm(), nodeState.getNodeId(),
            prevIdx, prevTerm, entries, raftLog.getCommitIndex()
    );

    String peerId = config.getPeerUrls().entrySet().stream()
            .filter(entry -> entry.getValue().equals(peer))
            .map(Map.Entry::getKey)
            .map(String::valueOf)
            .findFirst()
            .orElse("unknown");

    try {
        String url = peer + "/raft/appendEntries";
        ResponseEntity<AppendEntriesResponseDto> res = restTemplate.postForEntity(url, dto, AppendEntriesResponseDto.class);
        AppendEntriesResponseDto body = res.getBody() != null ? res.getBody() : new AppendEntriesResponseDto(-1, false);
        if (body.getTerm() > nodeState.getCurrentTerm()) {
            nodeStateManager.becomeFollower(body.getTerm());
            return false;
        }
        if (body.isSuccess()) {
            nextIndex.put(peer, ni + entries.size());
            matchIndex.put(peer, ni + entries.size() - 1);
            return true;
        } else {
            int newNextIndex = Math.max(0, ni - 1);
            nextIndex.put(peer, newNextIndex);
            return false;
        }
    } catch (ResourceAccessException e) {
        return false;
    } catch (Exception e) {
        return false;
    }
}

private void updateCommitIndex() {
    int majority = (config.getPeerUrlList().size() + 1) / 2 + 1;
    int term = nodeState.getCurrentTerm();
    for (int i = log.getLastIndex(); i > log.getCommitIndex(); i--) {
        int count = 1;
        for (int idx : matchIndex.values()) {
            if (idx >= i) count++;
        }
        if (count >= majority && log.getTermAt(i) == term) {
            log.setCommitIndex(i);
            applyEntries();
            break;
        }
    }
}

public AppendEntriesResponseDto handle(AppendEntriesRequestDto dto) {
    lockManager.getLogWriteLock().lock();
    lockManager.getStateWriteLock().lock();
    lockManager.getStateMachineWriteLock().lock();
    try {
        int term = nodeState.getCurrentTerm();
        int leaderTerm = dto.getTerm();

        // Reply false if term < currentTerm (§5.1)
        if (leaderTerm < term) return new AppendEntriesResponseDto(term, false);

        // If RPC request or response contains term T > currentTerm: set currentTerm = T, convert to follower (§5.1)
        if (leaderTerm > term) {
            stateManager.becomeFollower(leaderTerm);
            term = leaderTerm;
        }

        nodeState.setCurrentLeader(dto.getLeaderId());

        // Reply false if log doesn't contain an entry at prevLogIndex whose term matches prevLogTerm (§5.3)
        if (dto.getPrevLogIndex() > 0 &&
                (!raftLog.containsEntryAt(dto.getPrevLogIndex()) ||
                        raftLog.getTermAt(dto.getPrevLogIndex()) != dto.getPrevLogTerm())) {
            stateManager.resetElectionTimer();
            return new AppendEntriesResponseDto(term, false);
        }

        // If an existing entry conflicts with a new one (same index but different terms), delete the existing
        // entry and all that follow it (§5.3). Append any new entries not already in the log.
        int index = dto.getPrevLogIndex() + 1;
        List<LogEntry> entries = dto.getEntries();
        if (!entries.isEmpty()) {
            for (int i = 0; i < entries.size(); i++) {
                int logIndex = index + i;
                if (raftLog.containsEntryAt(logIndex) && raftLog.getTermAt(logIndex) != entries.get(i).getTerm()) {
                    raftLog.deleteFrom(logIndex);
                    raftLog.appendAll(entries.subList(i, entries.size()));
                    break;
                }
            }
            if (!raftLog.containsEntryAt(index)) {
                raftLog.appendAll(entries);
            }
        }

        // If leaderCommit > commitIndex, set commitIndex = min(leaderCommit, index of the last new entry)
        if (dto.getLeaderCommit() > raftLog.getCommitIndex()) {
            int lastNew = dto.getPrevLogIndex() + entries.size();
            raftLog.setCommitIndex(Math.min(dto.getLeaderCommit(), lastNew));
            applyEntries();
        }
        stateManager.resetElectionTimer();
        return new AppendEntriesResponseDto(term, true);
    } finally {
        lockManager.getStateMachineWriteLock().unlock();
        lockManager.getStateWriteLock().unlock();
        lockManager.getLogWriteLock().unlock();
    }
}

// If command received from the client: append entry to local log, respond after entry applied
// to state machine (§5.3)
public boolean handle(LogEntry clientEntry) {
    lockManager.getStateReadLock().lock();
    lockManager.getLogWriteLock().lock();
    int entryIndex;
    try {
        if (!raftNodeState.isLeader()) return false;
        raftLog.append(clientEntry);
        entryIndex = raftLog.getLastIndex();
    } finally {
        lockManager.getLogWriteLock().unlock();
        lockManager.getStateReadLock().unlock();
    }

    long start = System.currentTimeMillis();
    long timeoutMillis = raftConfig.getClientRequestTimeoutMillis();

    while (true) {
        lockManager.getLogReadLock().lock();
        lockManager.getStateReadLock().lock();
        try {
            if (raftNodeState.getCurrentRole() != Role.LEADER) {
                return false;
            }
            if (raftNodeState.getLastApplied() >= entryIndex) {
                return true;
            }
        } finally {
            lockManager.getStateReadLock().unlock();
            lockManager.getLogReadLock().unlock();
        }

        if (System.currentTimeMillis() - start > timeoutMillis) {
            return false;
        }
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
}

@GetMapping("/get")
public ResponseEntity<String> get(@RequestParam String key) {
    final String NO_OP_ENTRY = "NO_OP_ENTRY";
    try {
        handleRead(new WriteRequestDto(NO_OP_ENTRY, Long.MAX_VALUE, NO_OP_ENTRY, NO_OP_ENTRY));
    } catch (Exception e) {
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("Failed to process read request: " + e.getMessage());
    }

    lockManager.getStateMachineReadLock().lock();
    try {
        String value = kvStore.get(key);
        if (value == null) {
            return ResponseEntity.status(HttpStatus.NOT_FOUND)
                    .body("Key not found: " + key);
        }
        return ResponseEntity.ok(value);
    } catch (Exception e) {
        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                .body("Error retrieving key: " + e.getMessage());
    } finally {
        lockManager.getStateMachineReadLock().unlock();
    }
}

private void handleRead(WriteRequestDto request) {
    LogEntry entry = new LogEntry(
            nodeState.getCurrentTerm(),
            request.getKey(),
            request.getValue(),
            LogEntry.Operation.PUT,
            request.getClientId(),
            request.getSequenceNumber()
    );
    boolean committed = replicationManager.handleClientRequest(entry);
    if (!committed) throw new RuntimeException("Can't process the read at this moment");
}

@Override
public void apply(LogEntry entry) {
    if (entry == null) {
        throw new IllegalArgumentException("Log entry cannot be null");
    }

    // deduplication check
    String clientId = entry.getClientId();
    long sequenceNumber = entry.getSequenceNumber();
    Long lastSequenceNumber = kvStore.getLastSequenceNumber(clientId);
    if (lastSequenceNumber != null && sequenceNumber <= lastSequenceNumber) {
        return;
    }
...
}