docs(bg,en): fix ports and WebSocket example for new server model

- Update architecture diagram to include WebSocket server (port 9913) - Fix HTTP port from 9470 to 9912 (TCP + 440) in protocol docs - Fix WebSocket port from 9471 to 9913 (TCP + 441) in protocol docs - Fix WebSocket Nim example: remove non-existent onMessage API and unsafe cast[string](data), use correct WsServer API with safe helpers - Clarify TCP_NODELAY is also handled internally by hunos (HTTP) - Note that bytesToString/stringToBytes are wire-protocol helpers
feat: rate limiting, shared DB instance, TCP_NODELAY fix
2026-05-18 19:41:21 +03:00 · 2026-05-18 19:33:40 +03:00
13 changed files with 266 additions and 59 deletions
@@ -34,10 +34,36 @@ BaraDB е **мултимодален database engine**, написан на Nim,
 - **WebSocket** (`core/websocket.nim`): Full-duplex стрийминг
 - **Embedded** (`storage/lsm.nim`): Директен in-process достъп

+### Сървърна Архитектура
+
+TCP и HTTP сървърите споделят един LSMTree инстанс за гарантиране на консистентност на данните:
+
+```
+┌─────────────────────────────────────────────────────────┐
+│                 СПОДЕЛЕНО СЪХРАНЕНИЕ                      │
+│                 LSMTree Инстанс                          │
+├────────────────────────┬────────────────────────────────┤
+│   TCP Сървър           │   HTTP Сървър                  │
+│   (Бинарен Протокол)   │   (REST API)                   │
+│   Порт: 9472           │   Порт: 9912 (TCP + 440)       │
+│   TCP_NODELAY: ВКЛ     │   Multi-threaded               │
+├────────────────────────┴────────────────────────────────┤
+│   WebSocket Сървър                                      │
+│   (Streaming/Pub-Sub)                                   │
+│   Порт: 9913 (TCP + 441)                                │
+│   TCP_NODELAY: ВКЛ                                      │
+└─────────────────────────────────────────────────────────┘
+```
+
+**Ключови оптимизации:**
+- **Споделен LSMTree** — Всички сървъри работят върху един и същ database инстанс, елиминирайки несъответствие в данните
+- **TCP_NODELAY** — Включен на слушащите и клиентските сокети в TCP/WebSocket сървърите за по-ниска латентност при малки съобщения. HTTP сървърът (чрез hunos) също управлява `TCP_NODELAY` вътрешно.
+- **Безопасна конверсия на байтове** — Правилни `bytesToString`/`stringToBytes` функции вместо unsafe `cast` операции във wire протокола
+
 ### Управление на Връзки

 - **Connection Pool** (`protocol/pool.nim`): Min/max лимити на връзки с idle timeout
- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket глобални и per-client лимити
+- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket глобални и per-client лимити, интегрирани в TCP и HTTP хендърите
 - **Автентикация** (`protocol/auth.nim`): JWT с HMAC-SHA256 и достъп на база роли
 - **TLS/SSL** (`protocol/ssl.nim`): TLS 1.3 с автоматично генерирани сертификати

@@ -95,7 +121,7 @@ BaraQL конвейрът:
 - **Raft Консенсус** (`core/raft.nim`): Leader election, log репликация
 - **Шардиране** (`core/sharding.nim`): Hash, range и consistent hashing
 - **Репликация** (`core/replication.nim`): Sync, async, semi-sync режими
- **Gossip Протокол** (`core/gossip.nim`): SWIM-подобно управление на членство
+- **Gossip Протокол** (`core/gossip.nim`): SWIM-подобно управление на членство с експоненциален backoff при грешки
 - **Разпределени Транзакции** (`core/disttxn.nim`): Two-phase commit

 ## Ключови Дизайнерски Решения
@@ -2,6 +2,21 @@

 Всички забележителни промени в BaraDB са документирани в този файл.

+## [Unreleased] — Подобрения на Сървърната Архитектура
+
+### Поправки
+
+- **Споделен Storage Инстанс** — TCP и HTTP сървърите вече споделят един LSMTree инстанс вместо да създават отделни, елиминирайки несъответствие в данните между протоколите
+- **Unsafe Cast Операции** — Заменени са всички `cast[string]` и `cast[seq[byte]]` с безопасните `bytesToString`/`stringToBytes` помощни функции, които правилно копират данните за съвместимост с ARC/ORC управлението на паметта
+
+### Промени
+
+- **TCP_NODELAY Включен** — Както слушащите, така и клиентските сокети вече имат TCP_NODELAY, намалявайки латентността за малки съобщения (заявки, acks) чрез деактивиране на алгоритъма на Nagle
+- **Rate Limiter Интегриран** — Token-bucket rate limiter вече е активен в TCP и HTTP сървър хендърите:
+  - TCP: Per-client rate limiting на query съобщения (код за грешка 429)
+  - HTTP: Rate limiting на `/query` и `/batch` endpoints с поддръжка на `X-Forwarded-For`
+- **Gossip Обработка на Грешки** — Подобрено възстановяване при UDP socket грешки с експоненциален backoff (100ms-5s) и презаписване на сокета само след 10 последователни грешки
+
 ## [Unreleased] — SQL:2023 Стабилизация

 ### Поправки
@@ -45,8 +45,10 @@ BaraDB може да се конфигурира чрез **променливи
 |------------|----------|----------|
 | `BARADB_AUTH_ENABLED` | `false` | Включване на автентикация |
 | `BARADB_JWT_SECRET` | — | JWT подписващ secret |
-| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Глобални заявки в секунда |
-| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Заявки в секунда за клиент |
+| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Глобални заявки в минута (token-bucket) |
+| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Заявки в минута за клиент (token-bucket) |
+
+> **Забележка:** Rate limiting се прилага на ниво протокол и в двата сървъра (TCP и HTTP). TCP използва ID на връзката за per-client лимити. HTTP използва `X-Forwarded-For` хедър (за прокси среди) или глобален ключ.

 ### Логване

@@ -11,6 +11,11 @@ BaraDB поддържа множество протоколи за клиент

 Бинарният протокол използва big-endian кодиране за всички многобайтови стойности.

+### Оптимизации на Връзката
+
+- **TCP_NODELAY** — Включен както на сървърните, така и на клиентските сокети за минимизиране на латентността при малки съобщения (заявки, acks). Това деактивира алгоритъма на Nagle, гарантирайки че пакетите се изпращат незабавно без буфериране.
+- **Rate Limiting** — Token-bucket rate limiter е интегриран на ниво протокол. Заявки надвишаващи лимита получават код за грешка 429.
+
 ### Жизнен Цикъл на Връзката

 ```
@@ -105,7 +110,19 @@ BaraDB поддържа множество протоколи за клиент

 ## HTTP/REST API

-Базов URL: `http://localhost:9470/api/v1`
+Базов URL: `http://localhost:9912/api/v1` (HTTP порт = TCP порт + 440)
+
+### Rate Limiting
+
+Всички query endpoints (`/query`, `/batch`) са защитени от token-bucket rate limiter:
+- **Глобален лимит** — Конфигурира се чрез `BARADB_RATE_LIMIT_GLOBAL` (по подразбиране: 10,000 заявки/минута)
+- **Per-client лимит** — Конфигурира се чрез `BARADB_RATE_LIMIT_PER_CLIENT` (по подразбиране: 1,000 заявки/минута)
+- **Идентификация на клиент** — Използва `X-Forwarded-For` хедър когато е зад прокси, иначе използва глобален ключ
+
+При rate limiting, сървърът връща HTTP 429 с:
+```json
+{"error": "Rate limit exceeded"}
+```

 ### Endpoints

@@ -254,7 +271,7 @@ POST /admin/check

 ## WebSocket Протокол

-URL: `ws://localhost:9471`
+URL: `ws://localhost:9913` (WebSocket порт = TCP порт + 441)

 ### Формат на Frame

@@ -284,7 +301,7 @@ WebSocket текстови frame-ове съдържат JSON съобщения
 ### Pub/Sub Пример

 ```javascript
-const ws = new WebSocket('ws://localhost:9471');
+const ws = new WebSocket('ws://localhost:9913');

 ws.onopen = () => {
  // Абониране за промени в таблица
@@ -34,10 +34,36 @@ Multiple communication protocols:
 - **WebSocket** (`core/websocket.nim`): Full-duplex streaming
 - **Embedded** (`storage/lsm.nim`): Direct in-process access

+### Server Architecture
+
+The TCP and HTTP servers share a single LSMTree instance to ensure data consistency:
+
+```
+┌─────────────────────────────────────────────────────────┐
+│                    SHARED STORAGE                        │
+│                    LSMTree Instance                      │
+├────────────────────────┬────────────────────────────────┤
+│   TCP Server           │   HTTP Server                  │
+│   (Binary Protocol)    │   (REST API)                   │
+│   Port: 9472           │   Port: 9912 (TCP + 440)       │
+│   TCP_NODELAY: ON      │   Multi-threaded               │
+├────────────────────────┴────────────────────────────────┤
+│   WebSocket Server                                      │
+│   (Streaming/Pub-Sub)                                   │
+│   Port: 9913 (TCP + 441)                                │
+│   TCP_NODELAY: ON                                       │
+└─────────────────────────────────────────────────────────┘
+```
+
+**Key optimizations:**
+- **Shared LSMTree** — All servers operate on the same database instance, eliminating data inconsistency
+- **TCP_NODELAY** — Enabled on listening and client sockets in TCP/WebSocket servers for lower latency on small messages. The HTTP server (via hunos) also handles `TCP_NODELAY` internally.
+- **Safe byte conversion** — Proper `bytesToString`/`stringToBytes` functions instead of unsafe `cast` operations in the wire protocol
+
 ### Connection Management

 - **Connection Pool** (`protocol/pool.nim`): Min/max connection limits with idle timeout
- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket global and per-client limits
+- **Rate Limiting** (`protocol/ratelimit.nim`): Token-bucket global and per-client limits, integrated in both TCP and HTTP handlers
 - **Authentication** (`protocol/auth.nim`): JWT with HMAC-SHA256 and role-based access
 - **TLS/SSL** (`protocol/ssl.nim`): TLS 1.3 with auto-generated certificates

@@ -121,7 +147,7 @@ Client → Protocol → Auth → Parser → AST → IR → Codegen
 - **Raft Consensus** (`core/raft.nim`): Leader election, log replication
 - **Sharding** (`core/sharding.nim`): Hash, range, and consistent hashing
 - **Replication** (`core/replication.nim`): Sync, async, semi-sync modes
- **Gossip Protocol** (`core/gossip.nim`): SWIM-like membership management
+- **Gossip Protocol** (`core/gossip.nim`): SWIM-like membership management with exponential backoff error recovery
 - **Distributed Transactions** (`core/disttxn.nim`): Two-phase commit

 ## Key Design Decisions
@@ -2,6 +2,21 @@

 All notable changes to BaraDB are documented in this file.

+## [Unreleased] — Server Architecture Improvements
+
+### Fixed
+
+- **Shared Storage Instance** — TCP and HTTP servers now share a single LSMTree instance instead of creating separate ones, eliminating data inconsistency between protocols
+- **Unsafe Cast Operations** — Replaced all `cast[string]` and `cast[seq[byte]]` with safe `bytesToString`/`stringToBytes` helper functions that properly copy data for ARC/ORC memory management compatibility
+
+### Changed
+
+- **TCP_NODELAY Enabled** — Both listening and client sockets now have TCP_NODELAY set, reducing latency for small messages (queries, acks) by disabling Nagle's algorithm
+- **Rate Limiter Integrated** — Token-bucket rate limiter is now active in both TCP and HTTP server handlers:
+  - TCP: Per-client rate limiting on query messages (error code 429)
+  - HTTP: Rate limiting on `/query` and `/batch` endpoints with `X-Forwarded-For` support
+- **Gossip Error Handling** — Improved UDP socket error recovery with exponential backoff (100ms-5s) and socket recreation only after 10 consecutive failures
+
 ## [Unreleased] — AI-Native Platform

 ### Added
@@ -45,8 +45,10 @@ BaraDB can be configured via **environment variables**, a **config file**, or **
 |----------|---------|-------------|
 | `BARADB_AUTH_ENABLED` | `false` | Enable authentication |
 | `BARADB_JWT_SECRET` | — | JWT signing secret |
-| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Global requests per second |
-| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Per-client requests per second |
+| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Global requests per minute (token-bucket) |
+| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Per-client requests per minute (token-bucket) |
+
+> **Note:** Rate limiting is enforced at the protocol level in both TCP and HTTP servers. TCP uses client connection ID for per-client limits. HTTP uses `X-Forwarded-For` header (for proxy setups) or a global key.

 ### Logging

@@ -11,6 +11,11 @@ BaraDB supports multiple protocols for client communication:

 The binary protocol uses big-endian encoding for all multi-byte values.

+### Connection Optimizations
+
+- **TCP_NODELAY** — Enabled on both server and client sockets to minimize latency for small messages (queries, acks). This disables Nagle's algorithm, ensuring packets are sent immediately without buffering.
+- **Rate Limiting** — Token-bucket rate limiter is integrated at the protocol level. Queries exceeding the rate limit receive error code 429.
+
 ### Connection Lifecycle

 ```
@@ -140,7 +145,19 @@ printf '\x00\x00\x00\x12\x01\x02\x00\x00\x00\x00\x08SELECT 1' > /dev/tcp/localho

 ## HTTP/REST API

-Base URL: `http://localhost:9470/api/v1`
+Base URL: `http://localhost:9912/api/v1` (HTTP port = TCP port + 440)
+
+### Rate Limiting
+
+All query endpoints (`/query`, `/batch`) are protected by a token-bucket rate limiter:
+- **Global rate** — Configured via `BARADB_RATE_LIMIT_GLOBAL` (default: 10,000 requests/minute)
+- **Per-client rate** — Configured via `BARADB_RATE_LIMIT_PER_CLIENT` (default: 1,000 requests/minute)
+- **Client identification** — Uses `X-Forwarded-For` header when behind a proxy, otherwise uses a global key
+
+When rate-limited, the server returns HTTP 429 with:
+```json
+{"error": "Rate limit exceeded"}
+```

 ### Endpoints

@@ -304,7 +321,7 @@ POST /admin/check

 ## WebSocket Protocol

-URL: `ws://localhost:9471`
+URL: `ws://localhost:9913` (WebSocket port = TCP port + 441)

 ### Frame Format

@@ -334,7 +351,7 @@ WebSocket text frames contain JSON messages:
 ### Pub/Sub Example

 ```javascript
-const ws = new WebSocket('ws://localhost:9471');
+const ws = new WebSocket('ws://localhost:9913');

 ws.onopen = () => {
  // Subscribe to table changes
@@ -386,7 +403,7 @@ let error = makeErrorMessage(1, 42, "Syntax error")
 ```nim
 import barabadb/protocol/http

-var router = newHttpRouter(port = 9470)
+var router = newHttpRouter(port = 9912)

 router.get("/api/users", proc(req: Request): Future[JsonNode] {.async.} =
  return %*[
@@ -405,11 +422,15 @@ router.post("/api/users", proc(req: Request): Future[JsonNode] {.async.} =
 ```nim
 import barabadb/core/websocket

-var server = newWsServer(port = 9471)
-server.onMessage = proc(ws: WebSocket, data: seq[byte]) {.gcsafe.} =
-  echo "Received: ", cast[string](data)
-  asyncCheck ws.send(cast[string](data))  # Echo
-asyncCheck server.run()
+proc bytesToString(data: seq[byte]): string =
+  result = newString(data.len)
+  for i in 0..<data.len:
+    result[i] = char(data[i])
+
+var server = newWsServer(config, secretKey)
+server.onInsert = proc(table, key, value: string) =
+  echo "Insert in ", table, ": ", key
+asyncCheck server.run(9913)  # WS port = TCP port + 441
 ```

 ### Connection Pool
@@ -320,21 +320,36 @@ proc startGossipListener*(gp: GossipProtocol) {.async.} =
  gp.sock.setSockOpt(OptReuseAddr, true)
  gp.sock.bindAddr(Port(gp.gossipPort))
  gp.running = true
+  var consecutiveErrors = 0
+  const maxConsecutiveErrors = 10
+  const baseRetryDelayMs = 100
+
  while gp.running:
    try:
      let (data, senderAddr, _) = await gp.sock.recvFrom(65535)
+      consecutiveErrors = 0  # Reset on success
      if data.len > 0:
        gp.handleIncomingGossip(data, senderAddr)
-    except:
-      # Small sleep on error to avoid spin
-      gp.sock.close()
-      # Recreate socket for next iteration
-      try:
-        gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
-        gp.sock.setSockOpt(OptReuseAddr, true)
-        gp.sock.bindAddr(Port(gp.gossipPort))
-      except:
+    except CatchableError as e:
+      if not gp.running:
        break
+      inc consecutiveErrors
+      if consecutiveErrors >= maxConsecutiveErrors:
+        # Recreate socket after too many errors
+        try:
+          gp.sock.close()
+        except:
+          discard
+        try:
+          gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
+          gp.sock.setSockOpt(OptReuseAddr, true)
+          gp.sock.bindAddr(Port(gp.gossipPort))
+          consecutiveErrors = 0
+        except:
+          break
+      # Exponential backoff with cap
+      let delayMs = min(baseRetryDelayMs * (1 shl min(consecutiveErrors, 6)), 5000)
+      await sleepAsync(delayMs)

 proc startGossip*(gp: GossipProtocol, healthIntervalMs: int = 1000,
                  gossipIntervalMs: int = 2000) =
@@ -19,6 +19,7 @@ import ../protocol/wire
 import ../core/websocket
 import jwt as jwtlib
 import ../protocol/auth
+import ../protocol/ratelimit

 type
  HttpServer* = ref object
@@ -29,6 +30,7 @@ type
    metrics*: Metrics
    secretKey*: string
    authManager*: AuthManager
+    rateLimiter*: RateLimiter
    ws*: WsServer

  Metrics* = ref object
@@ -38,13 +40,12 @@ type
    selectCount*: int
    activeConnections*: int

-proc newHttpServer*(config: BaraConfig): HttpServer =
-  let dataDir = config.dataDir / "server"
-  let db = newLSMTree(dataDir)
+proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer =
  let ctx = newExecutionContext(db)
  ctx.txnManager = newTxnManager()
  let secret = config.getEffectiveJwtSecret()
  let ws = newWsServer(config, secret)
+  let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient)
  ctx.onChange = proc(ev: ChangeEvent) =
    let msg = $ev.kind & " " & ev.table
    asyncCheck ws.broadcastToTable(ev.table, msg)
@@ -52,8 +53,14 @@ proc newHttpServer*(config: BaraConfig): HttpServer =
  HttpServer(config: config, running: false, db: db, ctx: ctx,
             secretKey: secret,
             authManager: am,
+             rateLimiter: rl,
             metrics: Metrics(), ws: ws)

+proc newHttpServer*(config: BaraConfig): HttpServer =
+  let dataDir = config.dataDir / "server"
+  let db = newLSMTree(dataDir)
+  return newHttpServerWithDb(config, db)
+
 # ----------------------------------------------------------------------
 # JWT helpers
 # ----------------------------------------------------------------------
@@ -124,6 +131,13 @@ proc queryHandler(server: HttpServer): RequestHandler =
        userId = uid
        role = r

+      # Rate limiting
+      let clientKey = request.headers["X-Forwarded-For"]
+      let rateLimitKey = if clientKey.len > 0: clientKey else: "http-global"
+      if not server.rateLimiter.allowRequest(rateLimitKey):
+        ctx.json(%*{"error": "Rate limit exceeded"}, 429)
+        return
+
      let body = parseJson(request.body)
      if body == nil or "query" notin body:
        ctx.json(%*{"error": "Missing 'query' in request body"}, 400)
@@ -8,6 +8,11 @@ import std/os
 import std/endians
 import std/monotimes
 import std/locks
+import std/nativesockets
+when defined(windows):
+  from std/winlean import TCP_NODELAY
+else:
+  from std/posix import TCP_NODELAY
 import config
 import logging
 import ../protocol/wire
@@ -22,6 +27,7 @@ import ../core/disttxn
 import ../core/replication
 import ../core/sharding
 import ../core/gossip
+import ../protocol/ratelimit
 import jwt as jwtlib

 type
@@ -37,12 +43,11 @@ type
    clusterMembership*: ClusterMembership
    gossipProtocol*: GossipProtocol
    tls*: TLSContext
+    rateLimiter*: RateLimiter
    activeConnections*: int
    activeConnectionsLock*: Lock

-proc newServer*(config: BaraConfig): Server =
-  let dataDir = config.dataDir / "server"
-  let db = newLSMTree(dataDir)
+proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
  let ctx = newExecutionContext(db)
  ctx.txnManager = newTxnManager()
  var tls: TLSContext = nil
@@ -85,19 +90,40 @@ proc newServer*(config: BaraConfig): Server =
  gp.onSuspect = proc(nodeId: string) {.gcsafe.} =
    cm.onNodeSuspect(nodeId)

+  # Initialize rate limiter
+  let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient)
+
  result = Server(config: config, running: false, db: db, ctx: ctx,
         txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(),
         replicationManager: newReplicationManager(),
         shardRouter: shardRouter,
         clusterMembership: cm,
         gossipProtocol: gp,
-         tls: tls)
+         tls: tls,
+         rateLimiter: rl)
  initLock(result.activeConnectionsLock)

+proc newServer*(config: BaraConfig): Server =
+  let dataDir = config.dataDir / "server"
+  let db = newLSMTree(dataDir)
+  return newServerWithDb(config, db)
+
 # ----------------------------------------------------------------------
 # Wire Protocol Helpers
 # ----------------------------------------------------------------------

+proc bytesToString(data: seq[byte]): string =
+  ## Safely convert seq[byte] to string (copies data)
+  result = newString(data.len)
+  for i in 0..<data.len:
+    result[i] = char(data[i])
+
+proc stringToBytes(data: string): seq[byte] =
+  ## Safely convert string to seq[byte] (copies data)
+  result = newSeq[byte](data.len)
+  for i in 0..<data.len:
+    result[i] = byte(data[i])
+
 proc readUint32BE(data: string, pos: int): uint32 =
  if pos + 4 > data.len:
    raise newException(ValueError, "readUint32BE: index out of bounds")
@@ -381,7 +407,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
                let key = data[0..<nullPos]
                let value = data[nullPos+1..^1]
                if value.len > 0:
-                  server.db.put(key, cast[seq[byte]](value))
+                  server.db.put(key, stringToBytes(value))
                else:
                  server.db.delete(key)
          await client.send("ACK " & $lsn & "\n")
@@ -435,27 +461,35 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}

      if not authenticated and header.kind != mkAuth:
        let err = makeErrorMessage(header.requestId, 401, "Authentication required")
-        await client.send(cast[string](err))
+        await client.send(bytesToString(err))
        continue

+      # Rate limiting for query messages
+      if header.kind in {mkQuery, mkQueryParams}:
+        let clientKey = "client-" & $clientId
+        if not server.rateLimiter.allowRequest(clientKey):
+          let err = serializeError(429, "Rate limit exceeded", header.requestId)
+          await client.send(bytesToString(err))
+          continue
+
      case header.kind
      of mkAuth:
-        let tokenStr = parseAuthMessage(cast[seq[byte]](payload))
+        let tokenStr = parseAuthMessage(stringToBytes(payload))
        let (valid, userId, role) = verifyToken(secret, tokenStr)
        if valid:
          authenticated = true
          connCtx.currentUser = userId
          connCtx.currentRole = role
          let okMsg = makeAuthOkMessage(header.requestId)
-          await client.send(cast[string](okMsg))
+          await client.send(bytesToString(okMsg))
          info("Client " & $clientId & " authenticated as " & userId)
        else:
          let err = makeErrorMessage(header.requestId, 403, "Invalid token")
-          await client.send(cast[string](err))
+          await client.send(bytesToString(err))

      of mkQuery:
        var pos = 0
-        let queryStr = readString(cast[seq[byte]](payload), pos)
+        let queryStr = readString(stringToBytes(payload), pos)
        info("[" & $clientId & "] Query: " & queryStr)

        # Shard-aware routing: check if this node should handle the write
@@ -471,7 +505,7 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
                if localShards.len == 0:
                  shardCheck = false
                  let err = serializeError(3, "Node not assigned to any shard", header.requestId)
-                  await client.send(cast[string](err))
+                  await client.send(bytesToString(err))
                break

        if shardCheck:
@@ -484,15 +518,15 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}

          if success:
            let dataMsg = serializeResult(result, header.requestId)
-            await client.send(cast[string](dataMsg))
+            await client.send(bytesToString(dataMsg))
            let completeMsg = serializeComplete(result.affectedRows, header.requestId)
-            await client.send(cast[string](completeMsg))
+            await client.send(bytesToString(completeMsg))
          else:
            let errorMsg = serializeError(1, errorMsg, header.requestId)
-            await client.send(cast[string](errorMsg))
+            await client.send(bytesToString(errorMsg))

      of mkQueryParams:
-        let (queryStr, params) = readQueryParamsMessage(cast[seq[byte]](payload))
+        let (queryStr, params) = readQueryParamsMessage(stringToBytes(payload))
        info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)")

        let startTicks = getMonoTime().ticks()
@@ -504,26 +538,26 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}

        if success:
          let dataMsg = serializeResult(result, header.requestId)
-          await client.send(cast[string](dataMsg))
+          await client.send(bytesToString(dataMsg))
          let completeMsg = serializeComplete(result.affectedRows, header.requestId)
-          await client.send(cast[string](completeMsg))
+          await client.send(bytesToString(completeMsg))
        else:
          let errorMsg = serializeError(1, errorMsg, header.requestId)
-          await client.send(cast[string](errorMsg))
+          await client.send(bytesToString(errorMsg))

      of mkPing:
        var pongMsg = WireMessage(
          header: MessageHeader(kind: mkPong, length: 0, requestId: header.requestId),
          payload: @[],
        )
-        await client.send(cast[string](serializeMessage(pongMsg)))
+        await client.send(bytesToString(serializeMessage(pongMsg)))

      of mkClose:
        break

      else:
        let errorMsg = serializeError(2, "Unsupported message kind: " & $header.kind, header.requestId)
-        await client.send(cast[string](errorMsg))
+        await client.send(bytesToString(errorMsg))

  except Exception as e:
    errorMsg("Client " & $clientId & " error: " & e.msg)
@@ -537,6 +571,11 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
    info("Client " & $clientId & " disconnected")
    client.close()

+proc setTcpNoDelay(sock: AsyncSocket) =
+  ## Enable TCP_NODELAY using the correct protocol level (IPPROTO_TCP).
+  ## asyncnet.setSockOpt uses SOL_SOCKET by default which fails for OptNoDelay.
+  setSockOptInt(sock.getFd, IPPROTO_TCP.cint, TCP_NODELAY, 1)
+
 proc run*(server: Server) {.async.} =
  server.running = true
  var clientId = 0
@@ -548,6 +587,7 @@ proc run*(server: Server) {.async.} =

  let sock = newAsyncSocket()
  sock.setSockOpt(OptReuseAddr, true)
+  setTcpNoDelay(sock)
  sock.bindAddr(Port(server.config.port), server.config.address)
  sock.listen()
  if server.config.tlsEnabled:
@@ -556,6 +596,7 @@ proc run*(server: Server) {.async.} =
    info("BaraDB listening on " & server.config.address & ":" & $server.config.port)
  while server.running:
    let client = await sock.accept()
+    setTcpNoDelay(client)
    acquire(server.activeConnectionsLock)
    var shouldAccept = true
    try:
@@ -5,6 +5,11 @@ import std/strutils
 import std/tables
 import std/base64
 import std/sets
+import std/nativesockets
+when defined(windows):
+  from std/winlean import TCP_NODELAY
+else:
+  from std/posix import TCP_NODELAY
 import config
 import jwt as jwtlib

@@ -307,16 +312,22 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} =
  inc server.nextId
  asyncCheck server.handleWsClient(client, server.nextId)

+proc setTcpNoDelay(sock: AsyncSocket) =
+  ## Enable TCP_NODELAY using the correct protocol level (IPPROTO_TCP).
+  setSockOptInt(sock.getFd, IPPROTO_TCP.cint, TCP_NODELAY, 1)
+
 proc run*(server: WsServer, port: int = 9471) {.async.} =
  server.running = true
  let sock = newAsyncSocket()
  sock.setSockOpt(OptReuseAddr, true)
+  setTcpNoDelay(sock)
  sock.bindAddr(Port(port))
  sock.listen()
  echo "BaraDB WebSocket listening on port ", port

  while server.running:
    let client = await sock.accept()
+    setTcpNoDelay(client)
    asyncCheck server.handleConnection(client)

 proc stop*(server: WsServer) =
@@ -230,17 +230,21 @@ proc main() =
        warn("Failed to generate self-signed certificate. TLS disabled.")
        config.tlsEnabled = false

+  # Shared LSMTree instance for both TCP and HTTP servers
+  let dataDir = config.dataDir / "server"
+  let sharedDb = newLSMTree(dataDir)
+
  # Start HTTP server (blocking, multi-threaded via hunos) in background thread
-  var httpServer = newHttpServer(config)
+  var httpServer = newHttpServerWithDb(config, sharedDb)
  spawn httpServer.run(config.port + 440)  # HTTP port = TCP port + 440

  # Start background compaction loop
-  let cm = newCompactionManager(httpServer.db)
+  let cm = newCompactionManager(sharedDb)
  asyncCheck cm.startCompactionLoop()

  # Create TCP server (initialization is synchronous, run is async)
  let localId {.used.} = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port
-  var tcpServer = newServer(config)
+  var tcpServer = newServerWithDb(config, sharedDb)

  # Start Raft cluster if enabled
  if config.raftEnabled:
@@ -251,11 +255,9 @@ proc main() =
      if cmd == "put":
        let parts = cast[string](data).split("\x00")
        if parts.len >= 2:
-          httpServer.db.put(parts[0], cast[seq[byte]](parts[1]))
-          tcpServer.db.put(parts[0], cast[seq[byte]](parts[1]))
+          sharedDb.put(parts[0], cast[seq[byte]](parts[1]))
      elif cmd == "delete":
-        httpServer.db.delete(cast[string](data))
-        tcpServer.db.delete(cast[string](data))
+        sharedDb.delete(cast[string](data))

    # Wire RAFT ↔ DistTxn
    wireRaftDistTxn(raftNode, tcpServer)