raft somewhat working

SymphonyDatabaseNode/src/main/java/io/github/lumijiez/app/NodeManager.java

@@ -11,17 +11,14 @@ import org.apache.logging.log4j.Logger;
 import java.util.ArrayList;
 import java.util.List;
 
-
 public class NodeManager {
     private final Logger logger = LogManager.getLogger(NodeManager.class);
 
     private final UdpListener listener;
     private final UdpMessageSender sender;
     private final WebSocketManager ws;
-//    private final Raft raft;
+    private final Raft raft;
-
     private final List<NodeInfo> nodes = new ArrayList<>();
-
     public static final String HOST = System.getenv().getOrDefault("HOSTNAME", "localhost");
     public static final int PORT = Integer.parseInt(System.getenv().getOrDefault("UDP_PORT", "8084"));
 
@@ -29,14 +26,15 @@ public class NodeManager {
         this.listener = new UdpListener(this);
         this.sender = new UdpMessageSender(this);
         this.ws = new WebSocketManager(this);
-//        this.raft = new Raft(this);
+        this.raft = new Raft(this, sender);
 
         listener.startListening();
         ws.connectAndListen();
+        raft.start();
     }
 
     public void handleMessage(String message) {
-//        raft.processMessage(message);
+        raft.processMessage(message);
     }
 
     public List<NodeInfo> getNodes() {
@@ -47,17 +45,16 @@ public class NodeManager {
         if (!nodes.contains(node)) {
             nodes.add(node);
             sender.sendMessage(node, "NODE_REGISTERED");
-
+          //  logger.info("New node registered, updating Raft peers");
-            logger.info("New node registered, restarting Raft consensus");
-//            raft.initializeRaft();
         }
     }
 
     public void removeNode(NodeInfo node) {
         nodes.remove(node);
+      //  logger.info("Node removed, updating Raft peers");
     }
 
     public UdpMessageSender getSender() {
         return sender;
     }
-}
+}

SymphonyDatabaseNode/src/main/java/io/github/lumijiez/network/UdpListener.java

@@ -48,7 +48,7 @@ public class UdpListener {
                             buffer.flip();
 
                             String message = new String(buffer.array(), 0, buffer.limit()).trim();
-                            logger.info("Received UDP {}:{}: {}", sender.getHostName(), sender.getPort(), message);
+                          //  logger.info("Received UDP {}:{}: {}", sender.getHostName(), sender.getPort(), message);
                             nodeManager.handleMessage(message);
                         }
                     }

SymphonyDatabaseNode/src/main/java/io/github/lumijiez/network/UdpMessageSender.java

@@ -23,13 +23,9 @@ public class UdpMessageSender {
             channel.configureBlocking(false);
             ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
             channel.send(buffer, new InetSocketAddress(node.hostname(), node.port()));
-            logger.info("Sent UDP to {}:{}: {}", node.hostname(), node.port(), message);
+          //  logger.info("Sent UDP to {}:{}: {}", node.hostname(), node.port(), message);
         } catch (IOException e) {
             logger.error("Error sending UDP message: {}", e.getMessage());
         }
     }
-
-    public void sendCountMessage(NodeInfo node, int count) {
-        sendMessage(node, String.valueOf(count));
-    }
 }

SymphonyDatabaseNode/src/main/java/io/github/lumijiez/network/WebSocketManager.java

@@ -41,7 +41,6 @@ public class WebSocketManager {
 
                                 for (NodeInfo newNode : nodes) {
                                     nodeManager.registerNode(newNode);
-//                                    logger.info("Discovered node: {}:{}", newNode.hostname(), newNode.port());
                                 }
 
                             } catch (Exception e) {
@@ -79,10 +78,4 @@ public class WebSocketManager {
             logger.error("Error in WebSocketManager: {}", e.getMessage());
         }
     }
-
+}
-    public void send(NodeInfo nodeInfo) {
-        if (webSocket != null) {
-            webSocket.sendText(gson.toJson(nodeInfo), true);
-        }
-    }
-}

SymphonyDatabaseNode/src/main/java/io/github/lumijiez/raft/Raft.java

@@ -1,20 +1,177 @@
 package io.github.lumijiez.raft;
 
-import java.util.ArrayList;
+import io.github.lumijiez.app.NodeManager;
-import java.util.List;
+import io.github.lumijiez.data.models.NodeInfo;
-import java.util.Random;
+import io.github.lumijiez.network.UdpMessageSender;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.*;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public class Raft {
     private static final Random RANDOM = new Random();
+    private final Logger logger = LogManager.getLogger(Raft.class);
 
-    private static RaftStates STATE = RaftStates.FOLLOWER;
+    // Configuration Parameters
-    private static int CURRENT_TERM = 0;
+    private static final int MIN_ELECTION_TIMEOUT = 300;
-    private static String VOTED_FOR = null;
+    private static final int MAX_ELECTION_TIMEOUT = 600;
-    private static List<String> LOG = new ArrayList<>();
+    private static final int HEARTBEAT_INTERVAL = 100;
-    private static int COMMIT_INDEX = 0;
+    private static final int QUORUM_FACTOR = 2;
-    private static int LAST_APPLIED = 0;
 
+    // State Variables
+    private RaftStates state = RaftStates.FOLLOWER;
+    private int currentTerm = 0;
+    private String votedFor = null;
+    private Set<String> votesReceived = new HashSet<>();
 
-    private static final int ELECTION_TIMEOUT = 150 + RANDOM.nextInt(151);
+    private final NodeManager nodeManager;
+    private final UdpMessageSender sender;
+    private final ScheduledExecutorService electionExecutor = Executors.newSingleThreadScheduledExecutor();
 
+    public Raft(NodeManager nodeManager, UdpMessageSender sender) {
+        this.nodeManager = nodeManager;
+        this.sender = sender;
+        nodeManager.getNodes().removeIf(node -> node.hostname().equals(NodeManager.HOST) && node.port() == NodeManager.PORT);
+
+        start();
+    }
+
+    public void start() {
+        logger.info("Raft initialization. Total peers: {}", nodeManager.getNodes().size());
+        becomeFollower(1);
+    }
+
+    private void becomeFollower(int term) {
+        if (term > currentTerm) {
+            state = RaftStates.FOLLOWER;
+            currentTerm = term;
+            votedFor = null;
+            votesReceived.clear();
+            logger.debug("Transitioned to FOLLOWER. Term: {}", term);
+        }
+        scheduleElectionTimeout();
+    }
+
+    private void becomeCandidate() {
+        state = RaftStates.CANDIDATE;
+        currentTerm++;
+        votedFor = NodeManager.HOST + ":" + NodeManager.PORT;
+        votesReceived = new HashSet<>(Set.of(votedFor)); // Self vote
+
+      //  logger.info("Starting election in Term {}. Attempting to collect votes.", currentTerm);
+        sendVoteRequests();
+    }
+
+    private void becomeLeader() {
+        if (state == RaftStates.CANDIDATE &&
+                votesReceived.size() >= (nodeManager.getNodes().size() / QUORUM_FACTOR + 1)) {
+            state = RaftStates.LEADER;
+            logger.info("LEADER ELECTION SUCCESS: Becoming LEADER in Term {}", currentTerm);
+            votesReceived.clear();
+            sendHeartbeats();
+        }
+    }
+
+    private void sendVoteRequests() {
+        String voteRequest = String.format("REQUEST_VOTE|%d|%s:%d",
+                currentTerm, NodeManager.HOST, NodeManager.PORT);
+
+        for (NodeInfo peer : nodeManager.getNodes()) {
+            sender.sendMessage(peer, voteRequest);
+        }
+    }
+
+    private void sendHeartbeats() {
+        if (state != RaftStates.LEADER) return;
+
+        String heartbeat = String.format("HEARTBEAT|%d|%s:%d",
+                currentTerm, NodeManager.HOST, NodeManager.PORT);
+
+        for (NodeInfo peer : nodeManager.getNodes()) {
+            sender.sendMessage(peer, heartbeat);
+        }
+
+        // Reschedule heartbeats
+        electionExecutor.schedule(this::sendHeartbeats, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
+    }
+
+    private void scheduleElectionTimeout() {
+        int timeout = MIN_ELECTION_TIMEOUT + RANDOM.nextInt(MAX_ELECTION_TIMEOUT - MIN_ELECTION_TIMEOUT);
+
+        electionExecutor.schedule(() -> {
+            if (state != RaftStates.LEADER) {
+                logger.debug("Election timeout triggered. Starting new election.");
+                becomeCandidate();
+            }
+        }, timeout, TimeUnit.MILLISECONDS);
+    }
+
+    public void processMessage(String message) {
+        String[] parts = message.split("\\|");
+        if (parts.length < 3) return;
+
+        try {
+            String type = parts[0];
+            int messageTerm = Integer.parseInt(parts[1]);
+            String sender = parts[2];
+
+            // Term comparison and potential state change
+            if (messageTerm > currentTerm) {
+                becomeFollower(messageTerm);
+            }
+
+            switch (type) {
+                case "REQUEST_VOTE":
+                    handleRequestVote(messageTerm, sender);
+                    break;
+                case "HEARTBEAT":
+                    handleHeartbeat(messageTerm, sender);
+                    break;
+                case "VOTE_GRANTED":
+                    handleVoteGranted(messageTerm, sender);
+                    break;
+            }
+        } catch (NumberFormatException e) {
+            logger.error("Error processing message: {}", message);
+        }
+    }
+
+    private void handleRequestVote(int term, String candidate) {
+        boolean voteGranted = false;
+
+        if (term >= currentTerm &&
+                (votedFor == null || votedFor.equals(candidate))) {
+            voteGranted = true;
+            votedFor = candidate;
+            currentTerm = term;
+            scheduleElectionTimeout();
+        }
+
+        String response = String.format("VOTE_GRANTED|%d|%s", term, voteGranted);
+        sender.sendMessage(NodeInfo.fromString(candidate), response);
+    }
+
+    private void handleHeartbeat(int term, String leader) {
+        if (term >= currentTerm) {
+            scheduleElectionTimeout();
+            becomeFollower(term);
+        }
+    }
+
+    private void handleVoteGranted(int term, String candidate) {
+        if (state == RaftStates.CANDIDATE && term == currentTerm) {
+            votesReceived.add(candidate);
+            logger.debug("Vote received from {}. Total votes: {}/{}",
+                    candidate, votesReceived.size(), (nodeManager.getNodes().size() / QUORUM_FACTOR + 1));
+
+            becomeLeader();
+        }
+    }
+
+    public void shutdown() {
+        electionExecutor.shutdownNow();
+    }
 }