Updating Routing tables

[EkaterinaB29]

Hello,
I am trying to connect multiple nodes which run in docker containers , first to the bootstrap node , then they should update their DHTs , the nodes connect to bootstrap and send FIND_NODE_REQ but bootstrap doesn't return FIND_NODE_RES with the information about the nodes that already connected to it ( so it doesn't update it's table ) yet they send PING-PONG... Can you please tell me what I am doing wrong?

package blockchain.network;

import com.google.gson.*;
import io.ep2p.kademlia.NodeSettings;
import io.ep2p.kademlia.connection.ConnectionInfo;
import io.ep2p.kademlia.connection.MessageSender;
import io.ep2p.kademlia.exception.HandlerNotFoundException;
import io.ep2p.kademlia.model.FindNodeAnswer;
import io.ep2p.kademlia.node.KademliaNode;
import io.ep2p.kademlia.node.KademliaNodeAPI;
import io.ep2p.kademlia.node.Node;
import io.ep2p.kademlia.node.external.BigIntegerExternalNode;
import io.ep2p.kademlia.protocol.message.*;
import io.ep2p.kademlia.serialization.gson.GsonFactory;
import io.ep2p.kademlia.table.Bucket;
import io.ep2p.kademlia.table.DefaultRoutingTableFactory;
import io.ep2p.kademlia.table.RoutingTable;

import java.io.*;
import java.lang.reflect.Type;
import java.math.BigInteger;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Map;
import java.util.concurrent.*;

public class TcpMessageSender<I extends Number, C extends ConnectionInfo> implements MessageSender<I, C>, Closeable {
    private final Map<BigInteger, KademliaNodeAPI<BigInteger, TcpConnectionInfo>> registeredNodes = new ConcurrentHashMap<>();
    private final ExecutorService executorService;
    private ServerSocket serverSocket;
    private final int listeningPort;
    private final Gson gson;

    public TcpMessageSender(int listeningPort) {
        this.executorService = Executors.newCachedThreadPool();
        this.listeningPort = listeningPort;

        // Initialize Gson using DefaultGsonFactory
        GsonFactory.DefaultGsonFactory<BigInteger, TcpConnectionInfo, Serializable, Serializable> factory =
                new GsonFactory.DefaultGsonFactory<>(BigInteger.class, TcpConnectionInfo.class, Serializable.class, Serializable.class);

        this.gson = new GsonFactory.DefaultGsonFactory<BigInteger, TcpConnectionInfo, Serializable, Serializable>(
                BigInteger.class,
                TcpConnectionInfo.class,
                Serializable.class,
                Serializable.class
        ).gsonBuilder().registerTypeAdapter(BigInteger.class, new JsonDeserializer<BigInteger>() {
                    @Override
                    public BigInteger deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) {
                        return new BigInteger(json.getAsString());
                    }
                })
                .registerTypeAdapter(BigInteger.class, new JsonSerializer<BigInteger>() {
                    @Override
                    public JsonElement serialize(BigInteger src, Type typeOfSrc, JsonSerializationContext context) {
                        return new JsonPrimitive(src.toString());
                    }
                })
                .create();
    }

    public void startServer() {
        try {
            serverSocket = new ServerSocket(listeningPort);
            System.out.println("Listening for incoming connections on port: " + listeningPort);

            executorService.submit(() -> {
                while (!serverSocket.isClosed()) {
                    try {
                        Socket clientSocket = serverSocket.accept();
                        handleIncomingMessage(clientSocket);
                    } catch (IOException e) {
                        System.err.println("Error accepting connection: " + e.getMessage());
                    }
                }
            });
        } catch (IOException e) {
            throw new RuntimeException("Failed to start server on port: " + listeningPort, e);
        }
    }

    private void handleIncomingMessage(Socket clientSocket) {
        executorService.submit(() -> {
            try (BufferedReader inputStream = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
                 PrintWriter outputStream = new PrintWriter(clientSocket.getOutputStream(), true)) {

                // Read the JSON content
                String json = inputStream.readLine();
                if (json == null || json.isEmpty()) {
                    System.err.println("Received empty message. Closing connection.");
                    return;
                }

                System.out.println("Received raw JSON: " + json);

                // Process the message
                processIncomingMessage(json, outputStream);

            } catch (IOException e) {
                System.err.println("Socket I/O error: " + e.getMessage());
            } finally {
                try {
                    if (!clientSocket.isClosed()) {
                        clientSocket.close();
                    }
                } catch (IOException e) {
                    System.err.println("Error closing client socket: " + e.getMessage());
                }
            }
        });
    }

    private void processIncomingMessage(String json, PrintWriter outputStream) {
        try {
            JsonElement jsonElement = JsonParser.parseString(json);
            if (!jsonElement.isJsonObject()) {
                throw new IllegalArgumentException("Invalid JSON format: " + json);
            }

            JsonObject jsonObject = jsonElement.getAsJsonObject();
            String messageType = jsonObject.get("type").getAsString();

            KademliaMessage<I, C, ?> message;
            switch (messageType) {
                case "PING":
                    message = gson.fromJson(jsonElement, PingKademliaMessage.class);
                    break;
                case "PONG":
                    message = gson.fromJson(jsonElement, PongKademliaMessage.class);
                    break;
                case "FIND_NODE_REQ":
                    message = gson.fromJson(jsonElement, FindNodeRequestMessage.class);
                    break;
                case "FIND_NODE_RESP":
                    message = gson.fromJson(jsonElement, FindNodeResponseMessage.class);
                    break;
                default:
                    throw new IllegalArgumentException("Unsupported message type: " + messageType);
            }

            if (message == null || message.getNode() == null) {
                throw new IllegalArgumentException("Message or Node cannot be null after deserialization.");
            }

            System.out.println("Processing message of type: " + messageType);
            KademliaMessage<I, C, ?> response = handleMessage(message);

            if (response != null) {
                String responseJson = gson.toJson(response);
                outputStream.println(responseJson);
                System.out.println("Sent response: " + responseJson);
            } else {
                System.err.println("No response generated for the message.");
            }
        } catch (JsonParseException e) {
            System.err.println("JSON parsing error: " + e.getMessage());
        } catch (Exception e) {
            System.err.println("Failed to process incoming message: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private void addSenderToRoutingTable(KademliaMessage<I, C, ?> message) {
        try {
            Node<I, C> senderNode = message.getNode();
            if (senderNode == null) {
                System.err.println("Sender node is null. Cannot update routing table.");
                return;
            }

            // Ensure sender node is registered
            if (!registeredNodes.containsKey(senderNode.getId())) {
                System.out.println("Registering sender node: " + senderNode.getId());
                KademliaNodeAPI<I, C> localNode = new KademliaNode<>(
                            senderNode.getId(),
                            senderNode.getConnectionInfo(),
                            new DefaultRoutingTableFactory<I, C, Bucket<I, C>>()
                                    .getRoutingTable(senderNode.getId()),
                            this,
                            new NodeSettings()
                    );
                registeredNodes.put((BigInteger) senderNode.getId(), (KademliaNodeAPI<BigInteger, TcpConnectionInfo>) localNode);
                // Add to the local node's routing table
                localNode.getRoutingTable().update(senderNode);
                System.out.println("Updated routing table with sender node: " + senderNode.getId());
            }
        } catch (Exception e) {
            System.err.println("Error updating routing table: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private KademliaMessage<I, C, ?> handleMessage(KademliaMessage<I, C, ?> message) {
        try {
            Node<I, C> senderNode = message.getNode();


            if (message instanceof PingKademliaMessage) {
                System.out.println("Responding to PING with PONG.");
                PongKademliaMessage<I, C> pongResponse = new PongKademliaMessage<>();
                pongResponse.setNode(senderNode);
                return pongResponse;
            }

            if (message instanceof FindNodeRequestMessage) {
                System.out.println("Handling FIND_NODE_REQ.");
                FindNodeRequestMessage<I, C> findNodeRequest = (FindNodeRequestMessage<I, C>) message;

                BigInteger targetId = (BigInteger) findNodeRequest.getData();
                KademliaNodeAPI<BigInteger, TcpConnectionInfo> localNode = registeredNodes.get((BigInteger) senderNode.getId());

                if (localNode != null) {
                    FindNodeAnswer<BigInteger, TcpConnectionInfo> closestNodes = localNode.getRoutingTable().findClosest(targetId);
                    FindNodeResponseMessage<I, C> response = new FindNodeResponseMessage<>();
                    response.setNode((Node<I, C>) localNode);
                    response.setData((FindNodeAnswer<I, C>) closestNodes);

                    System.out.println("Responding with closest nodes: " + closestNodes.getNodes());
                    return response;
                } else {
                    System.err.println("Local node not found for sender: " + senderNode.getId());
                }
            }

            throw new IllegalArgumentException("Unsupported message type: " + message.getClass().getSimpleName());
        } catch (Exception e) {
            System.err.println("Error handling message: " + e.getMessage());
            e.printStackTrace();
            return null;
        }
    }




    @Override
    public <U extends Serializable, O extends Serializable> KademliaMessage<I, C, O> sendMessage(
            KademliaNodeAPI<I, C> caller, Node<I, C> receiver, KademliaMessage<I, C, U> message) {

        System.out.println("Sending message from " + caller.getId() + " to " + receiver.getId());

        // Set the caller as the node in the message
        message.setNode(caller);

        TcpConnectionInfo tcpInfo = (TcpConnectionInfo) receiver.getConnectionInfo();

        try (Socket socket = new Socket(tcpInfo.getIpAddress(), tcpInfo.getPort());
             PrintWriter outputStream = new PrintWriter(socket.getOutputStream(), true);
             BufferedReader inputStream = new BufferedReader(new InputStreamReader(socket.getInputStream()))) {

            // Serialize and send the message
            String messageJson = gson.toJson(message);
            System.out.println("Serialized message: " + messageJson);
            outputStream.println(messageJson);

            // Read and deserialize the response
            String responseJson = inputStream.readLine();
            if (responseJson == null || responseJson.isEmpty()) {
                throw new RuntimeException("Received empty response from the server.");
            }

            System.out.println("Received response: " + responseJson);
            return gson.fromJson(responseJson, KademliaMessage.class);

        } catch (IOException e) {
            System.err.println("Socket communication error: " + e.getMessage());
            throw new RuntimeException("Failed to send message", e);
        }
    }

    @Override
    public <U extends Serializable> void sendAsyncMessage(KademliaNodeAPI<I, C> caller, Node<I, C> receiver, KademliaMessage<I, C, U> message) {
        executorService.submit(() -> {
            try {
                sendMessage(caller, receiver, message);
            } catch (Exception e) {
                System.err.println("Failed to send async message: " + e.getMessage());
            }
        });
    }

    @Override
    public void close() throws IOException {
        if (serverSocket != null && !serverSocket.isClosed()) {
            serverSocket.close();
        }
        executorService.shutdownNow();
    }

    public void registerNode(KademliaNodeAPI<BigInteger, TcpConnectionInfo> node) {
        registeredNodes.put(node.getId(), node);
        System.out.println("Registered node: " + node.getId());
    }
}

package blockchain.network;

import io.ep2p.kademlia.connection.ConnectionInfo;

import java.io.Serializable;

public class TcpConnectionInfo implements ConnectionInfo, Serializable {
    private final String ipAddress;
    private final int port;
    ;

    public TcpConnectionInfo(String ipAddress, int port) {
        this.ipAddress = ipAddress;
        this.port = port;
    }


    public String getIpAddress() {
        return ipAddress;
    }

    public int getPort() {
        return port;
    }

    @Override
    public String toString() {
        return "TcpConnectionInfo{" +
                "ipAddress='" + ipAddress + '\'' +
                ", port=" + port +
                '}';
    }
}

Thanks :)

[sepgh]

Hello,

First of all, I'm glad to see someone dedicated themselves to put the library into a real use.
I can't promise I'd have time to personally run the code any time soon and begin debugging, but following the logic, I'd like to suggest you to debug AbstractRoutingTable.findClosest(I destinationId) method from package io.ep2p.kademlia.table.

https://github.com/ep2p/kademlia-api/blob/main/src/main/java/io/ep2p/kademlia/table/AbstractRoutingTable.java#L120

That's where the bootstrap node grabs node informations of closest nodes to a destination.

Another thing that comes to my mind the generation of "Node ID". Are you sure they are unique? I don't see any validation logic and perhaps you'd want to add it later, but nodes should have unique ids, otherwise your bootstrap node may not return itself in the response (I guess) and that check is done by checking node ids in its routing table. You could use this code example to print the routing table of the bootstrap node and see whats up.

P.S: the logic in addSenderToRoutingTable is a bit confusing to me ...