diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index e536809..33f0f7f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -2,9 +2,9 @@ name: CI Pipeline
on:
push:
- branches: [ "main" ]
+ branches: [ "**" ]
pull_request:
- branches: [ "main" ]
+ branches: [ "**" ]
jobs:
backend-test:
diff --git a/README.md b/README.md
index 13346c4..668ac99 100644
--- a/README.md
+++ b/README.md
@@ -27,7 +27,6 @@ A distributed file storage system built with **Spring Boot** that demonstrates c
| **Gateway** | Entry point — handles uploads, downloads, chunking, replication |
| **Storage Nodes** | Store file chunks on disk with metadata |
| **ZAB Metadata Cluster** | Consensus-based metadata service — tracks which chunks go where |
-| **Membership Service** | Heartbeat-based health monitoring with auto re-replication |
---
@@ -43,8 +42,10 @@ A distributed file storage system built with **Spring Boot** that demonstrates c
| **Conflict Detection & Resolution** | Detects concurrent updates; resolves using version vectors + Lamport tiebreaker |
| **ZAB Consensus** | Leader election + two-phase commit for metadata operations |
| **Membership & Heartbeat** | Periodic heartbeats; nodes marked DOWN after timeout |
-| **Self-Healing (Re-Replication)** | Automatic chunk re-replication when a node goes DOWN |
-| **Lazy Cleanup** | Garbage collection of orphaned chunks after re-upload |
+| **Self-Healing (Repair)** | Automatic chunk re-replication to healthy nodes when one goes DOWN |
+| **Garbage Collection (GC)** | Automatic purging of "stray" chunks found during reconciliation passes |
+| **Auto-Healing & Rebalance** | Proactive scanning for under-replicated chunks (Heal Pass) to restore N=3 |
+| **Chaos UI (Fault-Togglable)** | Dynamic UI for simulating node crashes/recoveries without process restarts |
---
@@ -55,11 +56,13 @@ A distributed file storage system built with **Spring Boot** that demonstrates c
- **gRPC** (Netty + Protobuf) for inter-node communication
- **REST API** (Spring Web) for client-facing operations
- **Maven** for build management
+- **Simulation Layer**: Conditional heartbeats for non-destructive fault testing
### Frontend
- **React 19** + **TypeScript**
- **Vite** for fast, optimized builds
-- **TailwindCSS** for UI styling
+- **Lucide Icons** for modern cluster visualization
+- **TailwindCSS** for premium, dynamic styling
---
@@ -84,14 +87,11 @@ Create the following **Run Configurations** (`Run → Edit Configurations → +
| **Storage-9002** | `-Dspring.profiles.active=storage` | `--server.port=9002 --storage.dir=storage-9002` |
| **Storage-9003** | `-Dspring.profiles.active=storage` | `--server.port=9003 --storage.dir=storage-9003` |
| **Storage-9004** | `-Dspring.profiles.active=storage` | `--server.port=9004 --storage.dir=storage-9004` |
-| **ZAB Metadata Node 1 (Leader)** | `-Dspring.profiles.active=zab-metadata -Dmetadata.base.dir=.` | `--server.port=8081` |
-| **ZAB Metadata Node 2 (Follower)** | `-Dspring.profiles.active=zab-metadata -Dmetadata.base.dir=.` | `--server.port=8082` |
-| **ZAB Metadata Node 3 (Follower)** | `-Dspring.profiles.active=zab-metadata -Dmetadata.base.dir=.` | `--server.port=8083` |
+| **ZAB Node 1 (Leader)** | `-Dspring.profiles.active=zab-metadata -Dmetadata.base.dir=.` | `--server.port=8081` |
+| **ZAB Node 2** | `-Dspring.profiles.active=zab-metadata -Dmetadata.base.dir=.` | `--server.port=8082` |
+| **ZAB Node 3** | `-Dspring.profiles.active=zab-metadata -Dmetadata.base.dir=.` | `--server.port=8083` |
-**Start all configurations** in this order:
-1. Gateway-8080
-2. Storage nodes (9001–9004)
-3. ZAB Metadata nodes (8081–8083)
+**Start order**: Gateway -> Storage nodes -> ZAB nodes.
### Running with Terminal
@@ -125,15 +125,12 @@ Open **separate terminals** for each node:
### Running the Web UI
-Once the backend nodes are started, open a new terminal in the `demo-ui` folder:
-
```bash
cd demo-ui
-npm install
-npm run dev
+npm install && npm run dev
```
-Then visit the URL provided in the terminal (usually `http://localhost:5173`) to use the Google Drive-like interface!
+Visit `http://localhost:5173`.
### Verify It's Running
@@ -169,6 +166,12 @@ GET http://localhost:8080/membership/nodes
| `GET` | `/membership/nodes/down` | List only DOWN nodes |
| `GET` | `/membership/nodes/{nodeId}` | Get a specific node |
+### Chaos & Fault Simulation
+| Method | Endpoint | Description |
+|---|---|---|
+| `POST` | `/chaos/toggle` | Toggles node health (pauses/resumes heartbeat) |
+| `POST` | `/chaos/kill` | Hard crash (System.exit) for testing process loss |
+
### Storage Node (ports 9001–9004)
| Method | Endpoint | Description |
@@ -212,13 +215,14 @@ Tests cover:
- `ConcurrentUpdateConflictTest` — 7 tests
- `VersionedStorageServiceTest` — 8 tests
-### Test Fault Tolerance
+### Testing Fault Tolerance (viva Ready)
-1. Stop one storage node in IntelliJ
-2. Wait 5 seconds → node marked as DOWN
-3. Check `GET /membership/nodes` → status should be `DOWN`
-4. Watch gateway console → "triggering re-replication"
-5. Restart the node → status goes back to `UP`
+1. **Dashboard Toggling**: Go to the **Pulse Monitor** tab.
+2. **Simulate Failure**: Click "Simulate Failure" on Node 3.
+3. **Observation**: Watch the heart pulse turn red. Within 5-10s, the Gateway logs: `Repair Controller: Node 3 is DOWN, triggering re-replication`.
+4. **Verification**: Check the chunk mapping modal for a file; the chunk that was on Node 3 will now be copied to Node 4.
+5. **Node Recovery**: Click "Bring UP" on Node 3.
+6. **Auto-Healing**: Within 60s, watch the Gateway console for: `🗑️ Reconciliation GC: Found stray chunk on Node 3`. The "zombie" data is purged, and the system is perfectly rebalanced.
---
diff --git a/demo-ui/src/App.tsx b/demo-ui/src/App.tsx
index 10fd7ab..b4c0272 100644
--- a/demo-ui/src/App.tsx
+++ b/demo-ui/src/App.tsx
@@ -113,17 +113,20 @@ export default function App() {
}
};
- const handleKillNode = async (url: string) => {
+ const handleToggleNode = async (url: string) => {
const port = url.split(':').pop();
- if (!confirm(`⚠️ WARNING: This will immediately kill the Java process on port ${port} to test fault-tolerance. Proceed?`)) return;
+ const isCurrentlyUp = health?.storage.upNodeUrls.includes(url);
+ const action = isCurrentlyUp ? "Kill/Simulate Failure" : "Bring UP";
+
+ if (!confirm(`Confirm: ${action} for Node on port ${port}?`)) return;
try {
// Direct call to the node's chaos endpoint
- await axios.post(`${url}/chaos/kill`);
- alert(`Chaos sequence initiated for Node on port ${port}. It will shut down in 1 second.`);
+ await axios.post(`${url}/chaos/toggle`);
+ // No alert needed, the UI will poll and update automatically
} catch (e) {
- console.error("Failed to kill node", e);
- alert("Node is already offline or unreachable.");
+ console.error("Failed to toggle node health", e);
+ alert("Error reaching node. If it was a hard crash (System.exit), you must restart it manually in terminal.");
}
};
@@ -425,14 +428,16 @@ export default function App() {
{isUp ? 'OPERATIONAL' : 'OFFLINE'}
- {isUp && (
-
- )}
+
);
})}
diff --git a/src/main/java/com/example/demo/common/ChaosController.java b/src/main/java/com/example/demo/common/ChaosController.java
index a111817..3869b03 100644
--- a/src/main/java/com/example/demo/common/ChaosController.java
+++ b/src/main/java/com/example/demo/common/ChaosController.java
@@ -2,9 +2,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -19,27 +16,20 @@
public class ChaosController {
private static final Logger logger = LoggerFactory.getLogger(ChaosController.class);
- @Autowired
- private ApplicationContext context;
+ // Global flag for simulation
+ public static boolean isHealthy = true;
@PostMapping("/kill")
public String killNode() {
- logger.error("!!! CHAOS: Killing this node in 1 second !!!");
-
- // Use a background thread to shut down so the response can be sent first
- new Thread(() -> {
- try {
- Thread.sleep(1000);
- if (context instanceof ConfigurableApplicationContext) {
- ((ConfigurableApplicationContext) context).close();
- }
- logger.error("!!! SYSTEM EXIT !!!");
- System.exit(0);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }).start();
+ isHealthy = false;
+ logger.error("!!! CHAOS: Simulating node failure (Heartbeats stopped) !!!");
+ return "OK - Node is now SIMULATED DOWN.";
+ }
- return "OK - Killing node on this port.";
+ @PostMapping("/toggle")
+ public String toggleHealth() {
+ isHealthy = !isHealthy;
+ logger.info("!!! CHAOS: Node health toggled. Now healthy: " + isHealthy + " !!!");
+ return isHealthy ? "UP" : "DOWN";
}
}
diff --git a/src/main/java/com/example/demo/membership/HeartbeatSender.java b/src/main/java/com/example/demo/membership/HeartbeatSender.java
index 3eb205b..adb969b 100644
--- a/src/main/java/com/example/demo/membership/HeartbeatSender.java
+++ b/src/main/java/com/example/demo/membership/HeartbeatSender.java
@@ -1,5 +1,7 @@
package com.example.demo.membership;
+import com.example.demo.common.ChaosController;
+
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Profile;
import org.springframework.http.HttpEntity;
@@ -30,6 +32,10 @@ public HeartbeatSender() {
@Scheduled(fixedRate = 1000)
public void sendHeartbeat() {
+ if (!ChaosController.isHealthy) {
+ // Simulation: Node is "dead", so no heartbeats
+ return;
+ }
try {
HeartbeatRequest request = new HeartbeatRequest(nodeId, serverPort);
HttpHeaders headers = new HttpHeaders();
diff --git a/src/main/java/com/example/demo/membership/RepairController.java b/src/main/java/com/example/demo/membership/RepairController.java
index 91ac6a5..059ff10 100644
--- a/src/main/java/com/example/demo/membership/RepairController.java
+++ b/src/main/java/com/example/demo/membership/RepairController.java
@@ -1,7 +1,5 @@
package com.example.demo.membership;
-import com.example.demo.membership.MembershipService;
-import com.example.demo.membership.NodeInfo;
import com.example.demo.metadata.Manifest;
import com.example.demo.metadata.ZabMetadataService;
import org.springframework.context.annotation.Profile;
@@ -148,7 +146,7 @@ private List listChunks(NodeInfo node) {
*/
@Scheduled(fixedRate = 60000) // Every 1 minute
public void reconcile() {
- System.out.println("🧹 Reconciliation: Starting Garbage Collection pass...");
+ System.out.println("🧹 Reconciliation: Starting Cluster Health & GC pass...");
try {
Map allManifests = zabMetadataService.getAllManifests();
List upNodes = membershipService.getUpNodes();
@@ -158,7 +156,18 @@ public void reconcile() {
return;
}
- // Map chunkId -> Set of authorized node URLs
+ // 1. Build Physical Inventory Map (chunkId -> List of Nodes that have it)
+ Map> physicalInventory = new HashMap<>();
+ for (NodeInfo node : upNodes) {
+ List chunks = listChunks(node);
+ if (chunks != null) {
+ for (String chunkId : chunks) {
+ physicalInventory.computeIfAbsent(chunkId, k -> new HashSet<>()).add(node);
+ }
+ }
+ }
+
+ // 2. Build Authorized Map (chunkId -> Set of authorized node URLs from ZAB)
Map> authorizedMap = new HashMap<>();
for (Manifest m : allManifests.values()) {
if (m.getReplicas() != null) {
@@ -169,28 +178,64 @@ public void reconcile() {
}
}
+ // 3. PASS 1: Garbage Collection (Delete strays)
int deletedCount = 0;
for (NodeInfo node : upNodes) {
- List physicalChunks = listChunks(node);
- if (physicalChunks == null) continue;
+ // If a node has a chunk that is not authorized for it, delete it
+ for (String chunkId : listChunks(node)) {
+ Set authorizedUrls = authorizedMap.get(chunkId);
+ if (authorizedUrls == null || !authorizedUrls.contains(node.getUrl())) {
+ System.out.println("🗑️ Reconciliation GC: Found stray chunk " + chunkId + " on Node " + node.getUrl());
+ if (deleteChunkFromNode(node, chunkId)) deletedCount++;
+ }
+ }
+ }
- for (String chunkId : physicalChunks) {
- Set authorizedNodes = authorizedMap.get(chunkId);
+ // 4. PASS 2: Self-Healing (Fill missing replicas)
+ int healedCount = 0;
+ for (Map.Entry> entry : authorizedMap.entrySet()) {
+ String chunkId = entry.getKey();
+ Set authorizedUrls = entry.getValue();
+
+ Set physicalNodes = physicalInventory.getOrDefault(chunkId, new HashSet<>());
+ int currentPhysicalCount = physicalNodes.size();
+
+ // If manifest says we should have X replicas but physics shows < TARGET_REPLICAS healthy ones
+ if (currentPhysicalCount < TARGET_REPLICAS) {
+ System.out.println("🩹 Reconciliation Heal: Chunk " + chunkId + " is under-replicated (" + currentPhysicalCount + "/" + TARGET_REPLICAS + ")");
- // If chunk is not in ANY manifest OR this node is not an authorized replica
- if (authorizedNodes == null || !authorizedNodes.contains(node.getUrl())) {
- System.out.println("🗑️ Reconciliation: Found stray chunk " + chunkId + " on unauthorized Node " + node.getUrl());
- boolean deleted = deleteChunkFromNode(node, chunkId);
- if (deleted) deletedCount++;
+ // Find a source node that actually has the data
+ NodeInfo sourceNode = physicalNodes.stream().findFirst().orElse(null);
+ if (sourceNode == null) {
+ System.err.println("❌ Critical Heal Error: No healthy source node for " + chunkId);
+ continue;
+ }
+
+ // Find a healthy node that DOESN'T have it yet (and didn't already have it in authorized)
+ // We prioritize nodes that are not in the current authorized set (since one of those is likely the one that failed)
+ NodeInfo targetNode = upNodes.stream()
+ .filter(n -> !physicalNodes.contains(n))
+ .findFirst()
+ .orElse(null);
+
+ if (targetNode != null) {
+ // Find the "failed" URL that we are replacing in the metadata.
+ // It's any URL in the authorized set that isn't currently UP or doesn't have the chunk.
+ String failedUrl = authorizedUrls.stream()
+ .filter(url -> physicalNodes.stream().noneMatch(pn -> pn.getUrl().equals(url)))
+ .findFirst()
+ .orElse(null);
+
+ if (failedUrl != null) {
+ System.out.println("🚀 Reconciliation Heal: Auto-repairing " + chunkId + " | Rebalancing to Node " + targetNode.getUrl() + " to replace " + failedUrl);
+ CompletableFuture.runAsync(() -> copyChunk(sourceNode, targetNode, chunkId, failedUrl), pool);
+ healedCount++;
+ }
}
}
}
- if (deletedCount > 0) {
- System.out.println("✅ Reconciliation: Successfully purged " + deletedCount + " stray chunks across cluster.");
- } else {
- System.out.println("✨ Reconciliation: Cluster is clean. No stray chunks found.");
- }
+ System.out.println("✨ Reconciliation finished. Purged: " + deletedCount + " | Healed: " + healedCount);
} catch (Exception e) {
System.out.println("⚠️ Reconciliation Error: " + e.getMessage());
diff --git a/src/main/java/com/example/demo/metadata/ZabMetadataService.java b/src/main/java/com/example/demo/metadata/ZabMetadataService.java
index 1636300..3bada6a 100644
--- a/src/main/java/com/example/demo/metadata/ZabMetadataService.java
+++ b/src/main/java/com/example/demo/metadata/ZabMetadataService.java
@@ -186,7 +186,12 @@ public Map getClusterStatus() {
try {
String leaderUrl = discoverLeader();
String url = leaderUrl + "/zab-meta/cluster/status";
- ResponseEntity