Implement a control plane

.github/workflows/docker.yml

@@ -46,7 +46,7 @@ jobs:
           tags: |
             type=ref,event=branch
             type=ref,event=pr
-            type=sha,prefix={{branch}}-
+            type=sha,prefix=commit-
             type=raw,value=latest,enable={{is_default_branch}}
 
       - name: Set up Docker Buildx

docker/Dockerfile

@@ -28,6 +28,7 @@ COPY --from=builder /app/target/release/test_client /app/test_client
 
 # Expose bolt port
 EXPOSE 7687
+EXPOSE 7688
 
 # Set logging level
 ENV RUST_LOG=info

server/docker.sh

@@ -1,2 +1,2 @@
 mkdir -p data logs
-docker run -d  -v $(pwd)/logs:/app/logs -v $(pwd)/data:/app/data -p 7687:7687 ghcr.io/ladybugdb/bolt4rs:main
+docker run -d  -v $(pwd)/logs:/app/logs -v $(pwd)/data:/app/data -p 7687:7687 -p 7688:7688 ghcr.io/ladybugdb/bolt4rs:main

server/src/bin/test_client.rs

@@ -1,6 +1,8 @@
 use bolt4rs::*;
 use clap::Parser;
 use std::error::Error;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
 
 #[derive(Parser)]
 #[command(name = "test_client")]
@@ -9,11 +11,32 @@ struct Args {
     /// The URI of the Bolt server
     #[arg(default_value = "127.0.0.1:7687")]
     uri: String,
+
+    /// Shutdown the server
+    #[arg(long)]
+    shutdown: bool,
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     let args = Args::parse();
+
+    if args.shutdown {
+        // Simple host extraction, assuming format "host:port"
+        let host = args.uri.split(':').next().unwrap_or("127.0.0.1");
+        let control_addr = format!("{}:7688", host);
+
+        println!("Connecting to control plane at {}", control_addr);
+        let mut stream = TcpStream::connect(control_addr).await?;
+        stream.write_all(b"SHUTDOWN").await?;
+
+        let mut buf = [0; 1024];
+        let n = stream.read(&mut buf).await?;
+        let response = String::from_utf8_lossy(&buf[..n]);
+        println!("Server response: {}", response);
+        return Ok(());
+    }
+
     let uri = args.uri;
     let user = "bolt";
     let pass = "test";

server/src/main.rs

@@ -28,17 +28,49 @@ async fn main() -> Result<()> {
     let listener = TcpListener::bind("0.0.0.0:7687").await?;
     debug!("Bolt server listening on 0.0.0.0:7687");
 
-    loop {
-        let (socket, addr) = listener.accept().await?;
-        debug!("New connection from {}", addr);
+    let control_listener = TcpListener::bind("0.0.0.0:7688").await?;
+    debug!("Control plane listening on 0.0.0.0:7688");
 
-        let db = system_db.clone();
-        tokio::spawn(async move {
-            if let Err(e) = handle_connection(&db, socket).await {
-                error!("Connection error: {}", e);
+    loop {
+        tokio::select! {
+            res = listener.accept() => {
+                match res {
+                    Ok((socket, addr)) => {
+                        debug!("New connection from {}", addr);
+                        let db = system_db.clone();
+                        tokio::spawn(async move {
+                            if let Err(e) = handle_connection(&db, socket).await {
+                                error!("Connection error: {}", e);
+                            }
+                        });
+                    }
+                    Err(e) => error!("Accept error: {}", e),
+                }
             }
-        });
+            res = control_listener.accept() => {
+                match res {
+                    Ok((mut socket, addr)) => {
+                        debug!("Control connection from {}", addr);
+                        let mut buf = [0; 1024];
+                        match socket.read(&mut buf).await {
+                            Ok(n) if n > 0 => {
+                                let command = String::from_utf8_lossy(&buf[..n]);
+                                if command.trim() == "SHUTDOWN" {
+                                    debug!("Shutdown command received");
+                                    let _ = socket.write_all(b"OK").await;
+                                    break;
+                                }
+                            }
+                            _ => {}
+                        }
+                    }
+                    Err(e) => error!("Control accept error: {}", e),
+                }
+            }
+        }
     }
+    debug!("Server shutting down");
+    Ok(())
 }
 
 async fn lbug_init() -> Result<lbug::Database> {

server/test.sh

@@ -1,7 +1,7 @@
 #!/bin/bash
 set -e
 
-rm -rf system
+rm -rf data/system
 
 # Kill any existing bolt4rs-server process
 echo "Cleaning up any existing bolt4rs-server process..."
@@ -19,6 +19,12 @@ SERVER_PID=$!
 echo "Waiting for server to start..."
 sleep 2
 
+# Check if server is still running
+if ! kill -0 $SERVER_PID 2>/dev/null; then
+    echo "Server failed to start!"
+    exit 1
+fi
+
 # Run the test client
 echo "Running test client..."
 if cargo run --bin test_client; then
@@ -30,8 +36,7 @@ else
 fi
 
 # Cleanup
-echo "Cleaning up..."
-kill $SERVER_PID || true
-wait $SERVER_PID 2>/dev/null || true
+echo "Shutting down server..."
+cargo run --bin test_client -- --shutdown || true
 
 exit $EXIT_CODE