Girish

Dockerfile

@@ -0,0 +1,21 @@
+# Build Stage
+FROM golang:1.25-alpine AS builder
+WORKDIR /app
+
+# Cache dependencies
+COPY go.mod go.sum ./
+RUN go mod download
+
+# Copy the rest of the code and build
+COPY . .
+RUN go build -o /fs-node
+
+# Run Stage
+FROM alpine:latest
+WORKDIR /app
+
+# Copy the compiled binary from the builder stage
+COPY --from=builder /fs-node /app/fs-node
+
+# Start the server
+CMD ["/app/fs-node"]

Readme.md

@@ -1,39 +1,48 @@
-# Distributed File Server (DFS)
+# 🚀 Go Distributed File Server (DFS)
 
-A work-in-progress **distributed file storage system** built from first principles in Go.  
-This project focuses on understanding **networking, systems design, and distributed systems internals** by building each layer manually instead of relying on frameworks.
+A decentralized, peer-to-peer, content-addressable file storage network built natively in Go.
 
----
+This project aims to bypass the scalability bottlenecks of traditional Distributed File Systems (like the HDFS NameNode memory limit) by utilizing a decentralized Hash Ring for metadata-free peer routing, and a deeply nested Content-Addressable Storage (CAS) engine.
 
-## 🎯 Project Goals
+## 🧱 Architecture & Features
 
-- Learn distributed systems by **building**, not just reading
-- Implement a fault-tolerant, scalable file storage system
-- Gain deep understanding of:
-  - TCP networking
-  - Message framing & protocols
-  - Data replication
-  - Node coordination
+### 1. Peer-to-Peer Networking
 
----
+- **Custom TCP Transport Layer:** Server-side `Listen`, `Accept`, and active outbound `Dial` capabilities.
+- **Message Framing:** Custom `LengthPrefixDecoder` to handle raw TCP byte streams reliably.
+- **Wire Protocol:** Structured binary message encoding (`MessagePayload`) using Go's `encoding/gob`, supporting `PUT`, `GET`, and `DELETE` commands.
+
+### 2. Decentralized Routing & Replication
+
+- **Consistent Hash Ring:** Uses SHA-1 and binary tree search (`sort.Search`) to map files to nodes mathematically, completely eliminating the need for a central metadata database.
+- **Dynamic Peer Discovery:** Nodes dynamically join the ring upon TCP handshake.
+- **Automated Replication:** Files uploaded to any node are automatically broadcast to all connected peers, with local CAS lookups to prevent infinite replication loops.
+
+### 3. Local Storage Engine (CAS)
 
-## 🧱 Current Status
+- **Content Addressable Storage:** Files are stored based on the SHA-1 hash of their key, ensuring deduplication.
+- **Optimized Disk I/O:** Hashes are split into deeply nested directory structures (e.g., `a1/b2/c3/...`) to prevent OS-level directory limitations during massive file ingestion.
+- **Deep Clean:** The `DELETE` command not only removes the file but recursively prunes empty parent directories.
 
-### ✅ Implemented
-- Custom **TCP transport layer**
-- Server-side `Listen` and `Accept` loop
-- Incoming connection handling
-- Manual testing using **Telnet**
-- Clean separation of transport logic
+### 4. Cluster Orchestration
 
-### 🔜 In Progress / Planned
-- Message framing (length-prefixed protocol)
-- Wire protocol (PUT / GET / DELETE)
-- Local file storage engine
-- Peer discovery & membership
-- Data replication & fault tolerance
+- **Containerized Testing:** Fully containerized using a multi-stage `Dockerfile` (Alpine Linux) and `docker-compose.yml`.
+- Spins up an isolated virtual network with one seed node and multiple dynamic peers to simulate real-world cluster topologies.
 
 ---
 
-## 🏗️ Architecture (Early Stage)
+## 🛠️ How to Run the Cluster
+
+### Prerequisites
+
+- [Docker Desktop](https://www.docker.com/products/docker-desktop/) installed and running.
+- Go 1.25+ (if running the test client locally).
+
+### Booting the Network
+
+You can spin up a 3-node cluster with a single command. Open your terminal in the project root and run:
 
+`bash
+docker-compose up --build
+`
+_Node 1 will act as the seed node. Nodes 2 and 3 will automatically boot, read their Environment Variables, and dial Node 1 to join the Hash Ring._

docker-compose.yml

@@ -0,0 +1,48 @@
+version: '3.8'
+
+services:
+  node1:
+    build: .
+    container_name: fs-node1
+    environment:
+      - LISTEN_ADDR=:3000
+      - STORAGE_ROOT=/data
+      - BOOTSTRAP_NODES=
+    ports:
+      - "3000:3000" # Expose to host machine on 3000
+    volumes:
+      - node1_data:/data
+
+  node2:
+    build: .
+    container_name: fs-node2
+    environment:
+      - LISTEN_ADDR=:3000
+      - STORAGE_ROOT=/data
+      - BOOTSTRAP_NODES=node1:3000 # Dial Node 1
+    ports:
+      - "4000:3000" # Expose to host machine on 4000
+    volumes:
+      - node2_data:/data
+    depends_on:
+      - node1
+
+  node3:
+    build: .
+    container_name: fs-node3
+    environment:
+      - LISTEN_ADDR=:3000
+      - STORAGE_ROOT=/data
+      - BOOTSTRAP_NODES=node1:3000 # Dial Node 1
+    ports:
+      - "5000:3000" # Expose to host machine on 5000
+    volumes:
+      - node3_data:/data
+    depends_on:
+      - node1
+
+volumes:
+  # Persistent virtual hard drives for each node
+  node1_data:
+  node2_data:
+  node3_data:

main.go

@@ -2,33 +2,58 @@ package main
 
 import (
 	"log"
+	"os"
+	"strings"
 
 	"github.com/girish/storage/p2p"
 	"github.com/girish/storage/store"
 )
 
-func main() {
+func makeServer(listenAddr string, storageroot string, bootstrapNodes []string) *FileServer {
 	tcpOpts := p2p.TCPTransportOps{
-		ListenAdder:   ":3000",
+		ListenAdder:   listenAddr,
 		Handshakefunc: p2p.NOPHandshakeFunc,
 		Decode:        p2p.LengthPrefixDecoder{},
 	}
 	tr := p2p.NewTCPTransport(tcpOpts)
 
 	storeOpts := store.StoreOpts{
-		Root:              "my_network_data",
+		Root:              storageroot,
 		PathTransformFunc: store.CASPathTransfromFunc,
 	}
 	localStore := store.NewStore(storeOpts)
 
 	serverOpts := FileServerOpts{
-		StorageRoot: "my_network_data", // This is where files will be saved later
-		Transport:   tr,
-		Store:       localStore,
+		ListenAddr:     listenAddr,
+		StorageRoot:    storageroot,
+		Transport:      tr,
+		Store:          localStore,
+		BootStrapNoeds: bootstrapNodes,
+	}
+	return NewFileServer(serverOpts)
+}
+
+func main() {
+	// 1. Read the network port from Docker (default to :3000)
+	listenAddr := os.Getenv("LISTEN_ADDR")
+	if listenAddr == "" {
+		listenAddr = ":3000"
+	}
+
+	// 2. Read the storage folder from Docker
+	storageRoot := os.Getenv("STORAGE_ROOT")
+	if storageRoot == "" {
+		storageRoot = "network_data"
 	}
-	server := NewFileServer(serverOpts)
 
-	if err := server.Start(); err != nil {
-		log.Fatal(err)
+	// 3. Read the comma-separated list of peers to dial on startup
+	bootstrapNodesStr := os.Getenv("BOOTSTRAP_NODES")
+	var bootstrapNodes []string
+	if bootstrapNodesStr != "" {
+		bootstrapNodes = strings.Split(bootstrapNodesStr, ",")
 	}
+
+	// 4. Spin up the single node
+	server := makeServer(listenAddr, storageRoot, bootstrapNodes)
+	log.Fatal(server.Start())
 }

p2p/TCP_transport.go

@@ -61,6 +61,21 @@ func (t *TCPTransport) ListenAndAccept() error {
 	return nil
 }
 
+// Dial implement the Transport interface. It connects to a remote node
+// and treats that connection exactly like an incomming one
+func (t *TCPTransport) Dial(addr string) error {
+	conn, err := net.Dial("tcp", addr)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("Successfully dialed and connected to peer: %s\n", addr)
+
+	go t.handleConn(conn)
+
+	return nil
+}
+
 func (t *TCPTransport) startAcceptLoop() {
 	for {
 		conn, err := t.listener.Accept()
@@ -105,8 +120,9 @@ func (t *TCPTransport) handleConn(conn net.Conn) {
 		}
 
 		t.rpcCh <- RPC{
-			From: conn.RemoteAddr(),
+			From:    conn.RemoteAddr(),
 			Payload: msg.Payload,
+			Peer:    peer,
 		}
 	}
 
@@ -131,3 +147,10 @@ func (p *TCPPeer) Send(b []byte) error {
 func (t *TCPTransport) Consume() <-chan RPC {
 	return t.rpcCh
 }
+
+func (t *TCPTransport) Close() error {
+	if t.listener != nil {
+		return t.listener.Close()
+	}
+	return nil
+}

p2p/hashring.go

@@ -0,0 +1,80 @@
+package p2p
+
+import (
+	"crypto/sha1"
+	"encoding/binary"
+	"sort"
+	"sync"
+)
+
+type HashRing struct {
+	mu      sync.RWMutex
+	nodes   []uint32
+	nodeMap map[uint32]string
+}
+
+func NewHashring() *HashRing {
+	return &HashRing{
+		nodes:   []uint32{},
+		nodeMap: make(map[uint32]string),
+	}
+}
+
+func (h *HashRing) hash(key string) uint32 {
+	hasher := sha1.New()
+	hasher.Write([]byte(key))
+	sum := hasher.Sum(nil)
+
+	return binary.BigEndian.Uint32(sum[:4])
+}
+
+func (h *HashRing) AddNode(addr string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+
+	hash := h.hash(addr)
+	if _, existed := h.nodeMap[hash]; !existed {
+		h.nodes = append(h.nodes, hash)
+
+		sort.Slice(h.nodes, func(i, j int) bool {
+			return h.nodes[i] < h.nodes[j]
+		})
+		h.nodeMap[hash] = addr
+	}
+}
+
+func (h *HashRing) RemoveNode(addr string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	hash := h.hash(addr)
+	if _, exists := h.nodeMap[hash]; exists {
+		delete(h.nodeMap, hash)
+
+		var newNode []uint32
+		for _, n := range h.nodes {
+			if n != hash {
+				newNode = append(newNode, n)
+			}
+		}
+		h.nodes = newNode
+	}
+}
+
+func (h *HashRing) GetNode(key string) string {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	if len(h.nodes) == 0 {
+		return ""
+	}
+	hash := h.hash(key)
+
+	idx := sort.Search(len(h.nodes), func(i int) bool {
+		return h.nodes[i] >= hash
+	})
+
+	if idx == len(h.nodes) {
+		idx = 0
+	}
+	return h.nodeMap[h.nodes[idx]]
+}

p2p/hashring_test.go

@@ -0,0 +1,56 @@
+package p2p
+
+import (
+	"testing"
+)
+
+func TestHashRing_AddAndGet(t *testing.T) {
+	ring := NewHashring()
+
+	nodes := []string{"192.168.1.1:3000", "192.168.1.2:3000", "192.168.1.3:3000"}
+	for _, n := range nodes {
+		ring.AddNode(n)
+	}
+
+	if len(ring.nodes) != 3 {
+		t.Fatalf("expected 3 nodes, got %d", len(ring.nodes))
+	}
+
+	// Verify that a specific key consistently maps to the exact same node
+	key := "my_test_file.txt"
+	expectedNode := ring.GetNode(key)
+
+	if expectedNode == "" {
+		t.Fatal("expected a valid node address, got empty string")
+	}
+
+	// Do it 100 times to ensure the mutexes and mapping remain stable
+	for i := 0; i < 100; i++ {
+		node := ring.GetNode(key)
+		if node != expectedNode {
+			t.Fatalf("consistent hash failed: expected %s, got %s", expectedNode, node)
+		}
+	}
+}
+
+func TestHashRing_RemoveNode(t *testing.T) {
+	ring := NewHashring()
+	ring.AddNode("nodeA")
+	ring.AddNode("nodeB")
+
+	key := "test_file"
+	firstOwner := ring.GetNode(key)
+
+	// Remove the node that owns the key
+	ring.RemoveNode(firstOwner)
+
+	if len(ring.nodes) != 1 {
+		t.Fatalf("expected 1 node, got %d", len(ring.nodes))
+	}
+
+	// The key should now smoothly map to the only remaining node
+	secondOwner := ring.GetNode(key)
+	if secondOwner == firstOwner {
+		t.Fatalf("expected owner to change after removal")
+	}
+}