Added integration testings

.github/workflows/ci.yml

@@ -42,3 +42,42 @@ jobs:
 
       - name: Test
         run: ctest --test-dir build --output-on-failure
+
+  integration-test:
+    name: Integration Tests (Docker)
+    runs-on: ubuntu-24.04
+    needs: build-linux
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install test dependencies
+        run: pip install -r src/Tests/integration/requirements.txt
+
+      - name: Build Docker image
+        run: docker build -t byteforge:ci .
+
+      - name: Run ByteForge container
+        run: docker run -d --name byteforge -p 6625:6625 byteforge:ci
+
+      - name: Wait for server to be ready
+        run: |
+          for i in {1..30}; do
+            (echo > /dev/tcp/127.0.0.1/6625) 2>/dev/null && break
+            sleep 1
+          done
+
+      - name: Run integration tests
+        run: pytest src/Tests/integration -v
+
+      - name: Dump container logs on failure
+        if: failure()
+        run: docker logs byteforge
+
+      - name: Stop container
+        if: always()
+        run: docker rm -f byteforge

.gitignore

@@ -8,6 +8,8 @@ CMakeFiles/
 CMakeScripts/
 cmake_install.cmake
 Makefile
+.venv/
+__pycache__/
 
 # ==========================================
 # Compiled Executables and Libraries and Logs

CMakeLists.txt

@@ -35,11 +35,11 @@ target_include_directories(KV_Core PUBLIC
 )
 
 # Conditional library management depending on Target Environment OS
+find_package(Threads REQUIRED)
+target_link_libraries(KV_Core PRIVATE Threads::Threads)
+
 if(WIN32)
     target_link_libraries(KV_Core PRIVATE ws2_32)
-else()
-    find_package(Threads REQUIRED)
-    target_link_libraries(KV_Core PRIVATE Threads::Threads)
 endif()
 
 # Build the main executable

Dockerfile

@@ -19,8 +19,7 @@ COPY . .
 
 # Run CMake and build the database
 RUN cmake -S . -B build -G Ninja -DCMAKE_BUILD_TYPE=Release && \
-    cmake --build build && \
-    cd build && ctest --output-on-failure
+    cmake --build build
 
 # ==========================================
 # Stage 2: Minimal Runtime (Production-grade)

src/Networking/IocpEventLoop.cpp

@@ -14,42 +14,39 @@ IocpEventLoop::~IocpEventLoop()
 
 void IocpEventLoop::add(SocketType sock)
 {
-    // Associate this socket with the completion port. The socket itself
-    // is used as the "completion key" so wait() can identify it later.
     CreateIoCompletionPort((HANDLE)sock, iocpHandle, (ULONG_PTR)sock, 0);
 
     auto context = std::make_unique<IocpContext>();
     context->socket = sock;
     ZeroMemory(&context->overlapped, sizeof(OVERLAPPED));
     context->buffer.buf = &context->dummy;
-    context->buffer.len = 0; // zero-byte: completes on arrival, consumes nothing
+    context->buffer.len = 0;
 
-    contexts[sock] = std::move(context);
+    {
+        std::lock_guard<std::mutex> lock(contextsMutex);
+        contexts[sock] = std::move(context);
+    }
     armRead(sock);
 }
 
+
 void IocpEventLoop::armRead(SocketType sock)
 {
+    std::lock_guard<std::mutex> lock(contextsMutex);
     auto it = contexts.find(sock);
     if (it == contexts.end())
         return;
 
     IocpContext *ctx = it->second.get();
     DWORD flags = 0;
     DWORD bytesRecvd = 0;
-
     WSARecv(sock, &ctx->buffer, 1, &bytesRecvd, &flags, &ctx->overlapped, nullptr);
-    // WSA_IO_PENDING is the expected outcome here - it means "queued,
-    // will complete later," not an error. Its eventual completion (or
-    // failure) surfaces later through wait().
 }
 
 void IocpEventLoop::remove(SocketType sock)
 {
+    std::lock_guard<std::mutex> lock(contextsMutex);
     contexts.erase(sock);
-    // IOCP has no direct "unregister" call. Once the caller closes the
-    // socket itself (via CloseSocket), the OS tears down its association
-    // with the port automatically.
 }
 
 int IocpEventLoop::wait(std::vector<EventLoopEntry> &out)

src/Networking/IocpEventLoop.h

@@ -7,6 +7,7 @@
 #include <windows.h>
 #include <unordered_map>
 #include <memory>
+#include <mutex>
 
 /**
  * @file IocpEventLoop.h
@@ -57,6 +58,9 @@ class IocpEventLoop : public IEventLoop
     /// Handle to the completion port itself.
     HANDLE iocpHandle;
 
+    /// Solves windows bug
+    std::mutex contextsMutex;
+
     /// Per-socket context objects, keyed by socket, kept alive while a recv is pending.
     std::unordered_map<SocketType, std::unique_ptr<IocpContext>> contexts;
 };

src/Server/Server.cpp

@@ -125,7 +125,10 @@ void Server::acceptClients()
             CloseSocket(AcceptSocket);
             continue;
         }
-        connections[AcceptSocket] = Connection{AcceptSocket, active_key};
+        {
+            std::lock_guard<std::mutex> lock(connectionsMutex);
+            connections[AcceptSocket] = Connection{AcceptSocket, active_key};
+        }
         eventLoop->add(AcceptSocket);
         std::cout << "[SERVER] Client connected and registered!\n";
     }
@@ -138,9 +141,12 @@ void Server::stop()
     std::cout << "[SERVER] Stop requested...\n";
 
     running = false;
-    for (auto &[sock, conn] : connections)
-        CloseSocket(sock);
-    connections.clear();
+    {
+        std::lock_guard<std::mutex> lock(connectionsMutex);
+        for (auto &[sock, conn] : connections)
+            CloseSocket(sock);
+        connections.clear();
+    }
     userSessionManager.close_all_sessions();
     CloseSocket(serverSocket);
     // Forces accept() loop to unblock by breaking the file descriptor channel
@@ -168,7 +174,21 @@ void Server::runEventLoop()
                 std::lock_guard<std::mutex> lock(expiredMutex);
                 close.swap(expiredSockets);
             }
-            for (auto &sock : close){
+            for (auto &sock : close)
+            {
+                bool stillValid = false;
+                {
+                    std::lock_guard<std::mutex> lock(connectionsMutex);
+                    stillValid = connections.find(sock) != connections.end();
+                } // release lock BEFORE calling closeConnection
+
+                if (!stillValid)
+                {
+                    std::cout << "[SERVER] Skipping expired socket " << sock
+                              << " - already replaced/closed\n";
+                    continue;
+                }
+
                 std::cout << "[SERVER] Closing expired connection, socket " << sock << "\n";
                 closeConnection(sock);
             }
@@ -181,9 +201,14 @@ void Server::runEventLoop()
         {
             if (entry.event == IOEvent::Readable)
             {
-                auto it = connections.find(entry.socket);
-                if (it == connections.end())
-                    continue; // already cleaned up
+                Connection conn;
+                {
+                    std::lock_guard<std::mutex> lock(connectionsMutex);
+                    auto it = connections.find(entry.socket);
+                    if (it == connections.end())
+                        continue;      // already cleaned up
+                    conn = it->second; // small struct, cheap to copy
+                }
 
                 {
                     std::lock_guard<std::mutex> lock(busyMutex);
@@ -195,15 +220,22 @@ void Server::runEventLoop()
                     busySockets.insert(entry.socket);
                 }
 
-                Connection conn = it->second; // small struct, cheap to copy
                 tpool.acceptJob([this, conn]()
                                 {
+                                    try {
                                     messageHandler(conn.socket, conn.sessionKey);
+                                    } catch (const std::exception& e) {
+                                        std::cout << "[SERVER] Exception in job: " << e.what() << "\n";
+                                    } catch (...) {
+                                        std::cout << "[SERVER] Unknown exception in job\n";
+                                    }
+
                                     std::lock_guard<std::mutex> lock(busyMutex);
                                     busySockets.erase(conn.socket); });
             }
             else // HangUp or Error
             {
+                std::cout << "[EPool] HangUp or error, disconnecting " << entry.socket << "\n";
                 closeConnection(entry.socket);
             }
         }
@@ -212,13 +244,19 @@ void Server::runEventLoop()
 
 void Server::closeConnection(SocketType sock)
 {
+    std::cout << "[SERVER] closeConnection(" << sock
+              << ") thread=" << std::this_thread::get_id()
+              << "\n";
     eventLoop->remove(sock);
 
-    auto it = connections.find(sock);
-    if (it != connections.end())
     {
-        userSessionManager.remove_session(it->second.sessionKey);
-        connections.erase(it);
+        std::lock_guard<std::mutex> lock(connectionsMutex);
+        auto it = connections.find(sock);
+        if (it != connections.end())
+        {
+            userSessionManager.remove_session(it->second.sessionKey);
+            connections.erase(it);
+        }
     }
 
     CloseSocket(sock);
@@ -236,7 +274,7 @@ void Server::messageHandler(SocketType clientSocket, const SessionKey &sessionKe
     if (bytesReceived == 0)
     {
         // A graceful close: the client actually hung up.
-        std::cout << "[CLIENT] Client disconnected\n";
+        std::cout << "[CLIENT] recv returned 0 on socket " << clientSocket << "\n";
         closeConnection(clientSocket);
         return;
     }
@@ -249,7 +287,7 @@ void Server::messageHandler(SocketType clientSocket, const SessionKey &sessionKe
             // got scheduled. Not an error, just nothing to do.
             return;
         }
-        std::cout << "[CLIENT] recv error, disconnecting\n";
+        std::cout << "[CLIENT] recv error, disconnecting " << clientSocket << "\n";
         closeConnection(clientSocket);
         return;
     }
@@ -276,15 +314,17 @@ void Server::messageHandler(SocketType clientSocket, const SessionKey &sessionKe
         if (value.empty())
         {
             std::string response = "Empty Value Recieved, Try again\n";
-            send(clientSocket, response.c_str(), response.length(), 0);
+            int sent = send(clientSocket, response.c_str(), response.length(), 0);
+            std::cout << "[SERVER] send() returned " << sent << "\n";
             return;
         }
 
         hashMap.insert(key, value);
         pers.appendToLog(command, key, value);
 
         std::string response = "Insert command was successful\n";
-        send(clientSocket, response.c_str(), response.length(), 0);
+        int sent = send(clientSocket, response.c_str(), response.length(), 0);
+        std::cout << "[SERVER] send() returned " << sent << "\n";
     }
     else if (command == "GET")
     {
@@ -298,7 +338,8 @@ void Server::messageHandler(SocketType clientSocket, const SessionKey &sessionKe
         else
             response = "Get command was successful but key dont exist\n";
 
-        send(clientSocket, response.c_str(), response.length(), 0);
+        int sent = send(clientSocket, response.c_str(), response.length(), 0);
+        std::cout << "[SERVER] send() returned " << sent << "\n";
     }
     else if (command == "DELETE")
     {
@@ -308,14 +349,16 @@ void Server::messageHandler(SocketType clientSocket, const SessionKey &sessionKe
         pers.appendToLog(command, key, "");
 
         std::string response = "Delete command was successful\n";
-        send(clientSocket, response.c_str(), response.length(), 0);
+        int sent = send(clientSocket, response.c_str(), response.length(), 0);
+        std::cout << "[SERVER] send() returned " << sent << "\n";
     }
     else
     {
         std::cout << "[SERVER] Unknown command received\n";
 
         std::string response = "No command was received\n";
-        send(clientSocket, response.c_str(), response.length(), 0);
+        int sent = send(clientSocket, response.c_str(), response.length(), 0);
+        std::cout << "[SERVER] send() returned " << sent << "\n";
     }
 }
 

src/Server/Server.h

@@ -37,6 +37,7 @@ class Server
     RateLimiter rt;                                             /** Server owns an instance of RateLimter class */
     UserSessionManager userSessionManager;                      /** Managing User Sessions */
     UserSessionBackgroundWorker user_session_background_worker; /** Background worker that sweeps expired sessions*/
+    std::mutex connectionsMutex;                                /**< Guards concurrent access to connections from the accept, event-loop, and thread-pool worker threads. */
     std::unordered_map<SocketType, Connection> connections;     /** Client connections, keyed by socket */
     std::unique_ptr<IEventLoop> eventLoop;                      /** Watches all client sockets for readiness */
     std::thread eventLoopThread;                                /** Thread that runs runEventLoop() */