diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 59530ee..fd570d5 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/.gitignore b/.gitignore index e1402ff..b5747eb 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,8 @@ CMakeFiles/ CMakeScripts/ cmake_install.cmake Makefile +.venv/ +__pycache__/ # ========================================== # Compiled Executables and Libraries and Logs diff --git a/CMakeLists.txt b/CMakeLists.txt index 4a6ee81..9e226ea 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/Dockerfile b/Dockerfile index d2df26e..d0f4dff 100644 --- a/Dockerfile +++ b/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) diff --git a/src/Networking/IocpEventLoop.cpp b/src/Networking/IocpEventLoop.cpp index 42d6625..63cb0bb 100644 --- a/src/Networking/IocpEventLoop.cpp +++ b/src/Networking/IocpEventLoop.cpp @@ -14,22 +14,25 @@ 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(); 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 lock(contextsMutex); + contexts[sock] = std::move(context); + } armRead(sock); } + void IocpEventLoop::armRead(SocketType sock) { + std::lock_guard lock(contextsMutex); auto it = contexts.find(sock); if (it == contexts.end()) return; @@ -37,19 +40,13 @@ void IocpEventLoop::armRead(SocketType sock) 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 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 &out) diff --git a/src/Networking/IocpEventLoop.h b/src/Networking/IocpEventLoop.h index a5b2b76..ccad331 100644 --- a/src/Networking/IocpEventLoop.h +++ b/src/Networking/IocpEventLoop.h @@ -7,6 +7,7 @@ #include #include #include +#include /** * @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> contexts; }; diff --git a/src/Server/Server.cpp b/src/Server/Server.cpp index fd0f367..bf4e344 100644 --- a/src/Server/Server.cpp +++ b/src/Server/Server.cpp @@ -125,7 +125,10 @@ void Server::acceptClients() CloseSocket(AcceptSocket); continue; } - connections[AcceptSocket] = Connection{AcceptSocket, active_key}; + { + std::lock_guard 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 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 lock(expiredMutex); close.swap(expiredSockets); } - for (auto &sock : close){ + for (auto &sock : close) + { + bool stillValid = false; + { + std::lock_guard 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 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 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 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 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,7 +314,8 @@ 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; } @@ -284,7 +323,8 @@ void Server::messageHandler(SocketType clientSocket, const SessionKey &sessionKe 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"; } } diff --git a/src/Server/Server.h b/src/Server/Server.h index 3b8a163..4c1b0c5 100644 --- a/src/Server/Server.h +++ b/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 connections; /** Client connections, keyed by socket */ std::unique_ptr eventLoop; /** Watches all client sockets for readiness */ std::thread eventLoopThread; /** Thread that runs runEventLoop() */ diff --git a/src/Tests/integration/README.md b/src/Tests/integration/README.md new file mode 100644 index 0000000..ea53a24 --- /dev/null +++ b/src/Tests/integration/README.md @@ -0,0 +1,117 @@ +# Integration Tests + +Integration tests for ByteForge, run against a real, running server instance +over TCP. These are not unit tests — they require ByteForge to actually be +running on `127.0.0.1:6625` before you start pytest. + +## Prerequisites + +- Python 3.11+ +- A running ByteForge server (either built locally or via Docker) + +## 1. Set up a virtual environment + +Create the `.venv` at the **repo root** (same level as `CMakeLists.txt`), not +inside `Tests/`. + +**Linux / macOS:** +```bash +python3 -m venv .venv +source .venv/bin/activate +``` + +**Windows (PowerShell):** +```powershell +python -m venv .venv +.venv\Scripts\Activate.ps1 +``` + +You should see `(.venv)` appear in your prompt once it's active. + +## 2. Install test dependencies + +From the repo root, with the venv activated: + +```bash +pip install -r src/Tests/integration/requirements.txt +``` + +This installs `pytest` and anything else the tests need. The client itself +(`client.py`) only uses the Python standard library (`socket`), so there's +nothing else to install for it. + +## 3. Start the server + +Pick one: + +**Option A — build and run locally:** +```bash +cmake -S . -B build -G Ninja -DCMAKE_BUILD_TYPE=Release +cmake --build build +./build/ByteForge # adjust to your actual binary name/path +``` + +**Option B — run via Docker:** +```bash +docker build -t byteforge:local . +docker run -d --name byteforge -p 6625:6625 byteforge:local +``` + +Either way, confirm the server printed something like: +`[SERVER] Server is now accepting connections!` + +before running tests — the tests don't wait for the server to be ready +themselves, and will fail with `ConnectionRefusedError` if it's not up yet. + +## 4. Run the tests + +From the repo root, with the venv still activated: + +```bash +pytest src/Tests/integration -v +``` + +Run a single file: +```bash +pytest src/Tests/integration/test_basic_commands.py -v +``` + +Run a single test: +```bash +pytest src/Tests/integration/test_advanced.py::test_unknown_command -v +``` + +## 5. Stop the server + +If you ran it via Docker: +```bash +docker rm -f byteforge +``` + +If you ran it locally, type `stop` in the server's console (per `main.cpp`'s +input loop), or Ctrl+C. + +## Test files + +- **`client.py`** — a small TCP client (`ByteForgeClient`) that speaks + ByteForge's newline-delimited protocol (`INSERT key value\n`, `GET key\n`, + `DELETE key\n`). Includes retry/reconnect logic for transient connection + drops (e.g. rate-limit resets), so other test files can just use it + directly without worrying about flakiness at the socket level. +- **`test_basic_commands.py`** — INSERT/GET/DELETE roundtrip, and confirms a + deleted key reads back as not found. +- **`test_advanced.py`** — unknown command handling, concurrent clients on + distinct keys, cross-client visibility of a delete, and rate limiter + behavior (burst capacity + refill). + +## Notes + +- ByteForge's rate limiter has a **burst capacity of 10** connections per IP, + refilling at **5/sec**. If you run tests repeatedly in quick succession + (e.g. re-running the whole suite back-to-back), you may see transient + connection resets while the bucket refills. `ByteForgeClient` retries + through these automatically; if you write new tests with raw sockets + instead of the client, keep this in mind. +- The server persists state (AOF/RDB), so if you're adding new tests, give + them their own unique keys rather than reusing ones from other tests, to + avoid cross-test interference. \ No newline at end of file diff --git a/src/Tests/integration/client.py b/src/Tests/integration/client.py new file mode 100644 index 0000000..364be2b --- /dev/null +++ b/src/Tests/integration/client.py @@ -0,0 +1,66 @@ +import socket +import time + + +class ByteForgeClient: + def __init__(self, host="127.0.0.1", port=6625, timeout=5.0, + max_retries=5, retry_delay=0.25): + self.host = host + self.port = port + self.timeout = timeout + self.max_retries = max_retries + self.retry_delay = retry_delay + self.sock = self._connect_with_retry() + + def _connect_with_retry(self): + last_error = None + for _ in range(self.max_retries): + try: + return socket.create_connection((self.host, self.port), timeout=self.timeout) + except (ConnectionRefusedError, ConnectionResetError, + ConnectionAbortedError, OSError) as e: + last_error = e + time.sleep(self.retry_delay) + raise ConnectionError( + f"Failed to connect to {self.host}:{self.port} after {self.max_retries} attempts" + ) from last_error + + def _send(self, command: str) -> str: + last_error = None + for _ in range(self.max_retries): + try: + self.sock.sendall((command + "\n").encode()) + return self._recv_line() + except (ConnectionResetError, ConnectionAbortedError, + BrokenPipeError, OSError) as e: + last_error = e + try: + self.sock.close() + except OSError: + pass + time.sleep(self.retry_delay) + self.sock = self._connect_with_retry() + raise ConnectionError( + f"Failed to send command after {self.max_retries} attempts" + ) from last_error + + def _recv_line(self) -> str: + data = b"" + while not data.endswith(b"\n"): + chunk = self.sock.recv(1024) + if not chunk: + break + data += chunk + return data.decode().strip() + + def insert(self, key: str, value: str) -> str: + return self._send(f"INSERT {key} {value}") + + def get(self, key: str) -> str: + return self._send(f"GET {key}") + + def delete(self, key: str) -> str: + return self._send(f"DELETE {key}") + + def close(self): + self.sock.close() \ No newline at end of file diff --git a/src/Tests/integration/requirements.txt b/src/Tests/integration/requirements.txt new file mode 100644 index 0000000..1c98737 --- /dev/null +++ b/src/Tests/integration/requirements.txt @@ -0,0 +1 @@ +pytest==8.3.4 \ No newline at end of file diff --git a/src/Tests/integration/test_basic_commands.py b/src/Tests/integration/test_basic_commands.py new file mode 100644 index 0000000..194f643 --- /dev/null +++ b/src/Tests/integration/test_basic_commands.py @@ -0,0 +1,103 @@ +import pytest +from client import ByteForgeClient +import socket +import threading +import time + + +@pytest.fixture +def client(): + c = ByteForgeClient() + yield c + c.close() + + +def test_insert_get_delete(client): + assert client.insert("volt", "3333") == "Insert command was successful" + assert client.get("volt") == "Get command was successful: 3333" + assert client.delete("volt") == "Delete command was successful" + + +def test_get_after_delete(client): + client.insert("volt", "3333") + client.get("volt") + client.delete("volt") + + response = client.get("volt") + assert response == "Get command was successful but key dont exist" + + + +def test_unknown_command(client): + response = client._send("FOOBAR volt 3333") + assert response == "No command was received" + + +def test_concurrent_clients_distinct_keys(): + # Kept comfortably under the rate limiter's burst capacity (10), + # since earlier tests in the session may have already spent tokens + # that haven't fully refilled yet. + num_clients = 5 + results: list[str | None] = [None] * num_clients + errors: list[str | None] = [None] * num_clients + + def worker(i): + try: + c = ByteForgeClient() + key = f"vault{i}" + value = str(1000 + i) + c.insert(key, value) + results[i] = c.get(key) + c.delete(key) + c.close() + except OSError as e: + errors[i] = str(e) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(num_clients)] + for t in threads: + t.start() + for t in threads: + t.join() + + for i in range(num_clients): + assert errors[i] is None, f"client {i} failed: {errors[i]}" + expected_value = str(1000 + i) + assert results[i] == f"Get command was successful: {expected_value}" + + +def test_cross_client_visibility_after_delete(): + client_a = ByteForgeClient() + client_b = ByteForgeClient() + + try: + assert client_a.insert("shared", "42") == "Insert command was successful" + assert client_a.delete("shared") == "Delete command was successful" + + response = client_b.get("shared") + assert response == "Get command was successful but key dont exist" + finally: + client_a.close() + client_b.close() + + +def test_rate_limiter_blocks_excess_connections(): + host, port = "127.0.0.1", 6625 + max_attempts = 50 + blocked = False + + for _ in range(max_attempts): + try: + s = socket.create_connection((host, port), timeout=1.0) + s.sendall(b"GET volt\n") + data = s.recv(1024) + s.shutdown(socket.SHUT_RDWR) + s.close() + + if data == b"": + blocked = True + break + except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError, OSError): + blocked = True + break + time.sleep(0.2) + assert blocked, f"Expected at least one blocked connection within {max_attempts} attempts" \ No newline at end of file