Compare commits

..

2 Commits

Author SHA1 Message Date
dimgigov 343f479127 docs(bg,en): fix ports and WebSocket example for new server model
CI / test (push) Has been cancelled
CI / verify (push) Has been cancelled
Clients CI / build-server (push) Has been cancelled
Clients CI / test-python (push) Has been cancelled
Clients CI / test-javascript (push) Has been cancelled
Clients CI / test-nim (push) Has been cancelled
Clients CI / test-rust (push) Has been cancelled
- 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
2026-05-18 19:41:21 +03:00
dimgigov 8afa998516 feat: rate limiting, shared DB instance, TCP_NODELAY fix
- Add rate limiter to TCP and HTTP servers
- Share single LSMTree between TCP and HTTP servers
- Replace unsafe cast[string]/cast[seq[byte]] with safe helpers
- Stabilize gossip UDP socket with exponential backoff
- Fix TCP_NODELAY via low-level setSockOptInt(IPPROTO_TCP)
  (asyncnet.setSockOpt(OptNoDelay) uses SOL_SOCKET and fails)
- Apply TCP_NODELAY to server and websocket sockets
2026-05-18 19:33:40 +03:00
13 changed files with 266 additions and 59 deletions
+28 -2
View File
@@ -34,10 +34,36 @@ BaraDB е **мултимодален database engine**, написан на Nim,
- **WebSocket** (`core/websocket.nim`): Full-duplex стрийминг - **WebSocket** (`core/websocket.nim`): Full-duplex стрийминг
- **Embedded** (`storage/lsm.nim`): Директен in-process достъп - **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 - **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 и достъп на база роли - **Автентикация** (`protocol/auth.nim`): JWT с HMAC-SHA256 и достъп на база роли
- **TLS/SSL** (`protocol/ssl.nim`): TLS 1.3 с автоматично генерирани сертификати - **TLS/SSL** (`protocol/ssl.nim`): TLS 1.3 с автоматично генерирани сертификати
@@ -95,7 +121,7 @@ BaraQL конвейрът:
- **Raft Консенсус** (`core/raft.nim`): Leader election, log репликация - **Raft Консенсус** (`core/raft.nim`): Leader election, log репликация
- **Шардиране** (`core/sharding.nim`): Hash, range и consistent hashing - **Шардиране** (`core/sharding.nim`): Hash, range и consistent hashing
- **Репликация** (`core/replication.nim`): Sync, async, semi-sync режими - **Репликация** (`core/replication.nim`): Sync, async, semi-sync режими
- **Gossip Протокол** (`core/gossip.nim`): SWIM-подобно управление на членство - **Gossip Протокол** (`core/gossip.nim`): SWIM-подобно управление на членство с експоненциален backoff при грешки
- **Разпределени Транзакции** (`core/disttxn.nim`): Two-phase commit - **Разпределени Транзакции** (`core/disttxn.nim`): Two-phase commit
## Ключови Дизайнерски Решения ## Ключови Дизайнерски Решения
+15
View File
@@ -2,6 +2,21 @@
Всички забележителни промени в BaraDB са документирани в този файл. Всички забележителни промени в 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 Стабилизация ## [Unreleased] — SQL:2023 Стабилизация
### Поправки ### Поправки
+4 -2
View File
@@ -45,8 +45,10 @@ BaraDB може да се конфигурира чрез **променливи
|------------|----------|----------| |------------|----------|----------|
| `BARADB_AUTH_ENABLED` | `false` | Включване на автентикация | | `BARADB_AUTH_ENABLED` | `false` | Включване на автентикация |
| `BARADB_JWT_SECRET` | — | JWT подписващ secret | | `BARADB_JWT_SECRET` | — | JWT подписващ secret |
| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Глобални заявки в секунда | | `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Глобални заявки в минута (token-bucket) |
| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Заявки в секунда за клиент | | `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Заявки в минута за клиент (token-bucket) |
> **Забележка:** Rate limiting се прилага на ниво протокол и в двата сървъра (TCP и HTTP). TCP използва ID на връзката за per-client лимити. HTTP използва `X-Forwarded-For` хедър (за прокси среди) или глобален ключ.
### Логване ### Логване
+20 -3
View File
@@ -11,6 +11,11 @@ BaraDB поддържа множество протоколи за клиент
Бинарният протокол използва big-endian кодиране за всички многобайтови стойности. Бинарният протокол използва big-endian кодиране за всички многобайтови стойности.
### Оптимизации на Връзката
- **TCP_NODELAY** — Включен както на сървърните, така и на клиентските сокети за минимизиране на латентността при малки съобщения (заявки, acks). Това деактивира алгоритъма на Nagle, гарантирайки че пакетите се изпращат незабавно без буфериране.
- **Rate Limiting** — Token-bucket rate limiter е интегриран на ниво протокол. Заявки надвишаващи лимита получават код за грешка 429.
### Жизнен Цикъл на Връзката ### Жизнен Цикъл на Връзката
``` ```
@@ -105,7 +110,19 @@ BaraDB поддържа множество протоколи за клиент
## HTTP/REST API ## 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 ### Endpoints
@@ -254,7 +271,7 @@ POST /admin/check
## WebSocket Протокол ## WebSocket Протокол
URL: `ws://localhost:9471` URL: `ws://localhost:9913` (WebSocket порт = TCP порт + 441)
### Формат на Frame ### Формат на Frame
@@ -284,7 +301,7 @@ WebSocket текстови frame-ове съдържат JSON съобщения
### Pub/Sub Пример ### Pub/Sub Пример
```javascript ```javascript
const ws = new WebSocket('ws://localhost:9471'); const ws = new WebSocket('ws://localhost:9913');
ws.onopen = () => { ws.onopen = () => {
// Абониране за промени в таблица // Абониране за промени в таблица
+28 -2
View File
@@ -34,10 +34,36 @@ Multiple communication protocols:
- **WebSocket** (`core/websocket.nim`): Full-duplex streaming - **WebSocket** (`core/websocket.nim`): Full-duplex streaming
- **Embedded** (`storage/lsm.nim`): Direct in-process access - **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 Management
- **Connection Pool** (`protocol/pool.nim`): Min/max connection limits with idle timeout - **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 - **Authentication** (`protocol/auth.nim`): JWT with HMAC-SHA256 and role-based access
- **TLS/SSL** (`protocol/ssl.nim`): TLS 1.3 with auto-generated certificates - **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 - **Raft Consensus** (`core/raft.nim`): Leader election, log replication
- **Sharding** (`core/sharding.nim`): Hash, range, and consistent hashing - **Sharding** (`core/sharding.nim`): Hash, range, and consistent hashing
- **Replication** (`core/replication.nim`): Sync, async, semi-sync modes - **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 - **Distributed Transactions** (`core/disttxn.nim`): Two-phase commit
## Key Design Decisions ## Key Design Decisions
+15
View File
@@ -2,6 +2,21 @@
All notable changes to BaraDB are documented in this file. 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 ## [Unreleased] — AI-Native Platform
### Added ### Added
+4 -2
View File
@@ -45,8 +45,10 @@ BaraDB can be configured via **environment variables**, a **config file**, or **
|----------|---------|-------------| |----------|---------|-------------|
| `BARADB_AUTH_ENABLED` | `false` | Enable authentication | | `BARADB_AUTH_ENABLED` | `false` | Enable authentication |
| `BARADB_JWT_SECRET` | — | JWT signing secret | | `BARADB_JWT_SECRET` | — | JWT signing secret |
| `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Global requests per second | | `BARADB_RATE_LIMIT_GLOBAL` | `10000` | Global requests per minute (token-bucket) |
| `BARADB_RATE_LIMIT_PER_CLIENT` | `1000` | Per-client requests per second | | `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 ### Logging
+30 -9
View File
@@ -11,6 +11,11 @@ BaraDB supports multiple protocols for client communication:
The binary protocol uses big-endian encoding for all multi-byte values. 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 ### 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 ## 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 ### Endpoints
@@ -304,7 +321,7 @@ POST /admin/check
## WebSocket Protocol ## WebSocket Protocol
URL: `ws://localhost:9471` URL: `ws://localhost:9913` (WebSocket port = TCP port + 441)
### Frame Format ### Frame Format
@@ -334,7 +351,7 @@ WebSocket text frames contain JSON messages:
### Pub/Sub Example ### Pub/Sub Example
```javascript ```javascript
const ws = new WebSocket('ws://localhost:9471'); const ws = new WebSocket('ws://localhost:9913');
ws.onopen = () => { ws.onopen = () => {
// Subscribe to table changes // Subscribe to table changes
@@ -386,7 +403,7 @@ let error = makeErrorMessage(1, 42, "Syntax error")
```nim ```nim
import barabadb/protocol/http import barabadb/protocol/http
var router = newHttpRouter(port = 9470) var router = newHttpRouter(port = 9912)
router.get("/api/users", proc(req: Request): Future[JsonNode] {.async.} = router.get("/api/users", proc(req: Request): Future[JsonNode] {.async.} =
return %*[ return %*[
@@ -405,11 +422,15 @@ router.post("/api/users", proc(req: Request): Future[JsonNode] {.async.} =
```nim ```nim
import barabadb/core/websocket import barabadb/core/websocket
var server = newWsServer(port = 9471) proc bytesToString(data: seq[byte]): string =
server.onMessage = proc(ws: WebSocket, data: seq[byte]) {.gcsafe.} = result = newString(data.len)
echo "Received: ", cast[string](data) for i in 0..<data.len:
asyncCheck ws.send(cast[string](data)) # Echo result[i] = char(data[i])
asyncCheck server.run()
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 ### Connection Pool
+18 -3
View File
@@ -320,21 +320,36 @@ proc startGossipListener*(gp: GossipProtocol) {.async.} =
gp.sock.setSockOpt(OptReuseAddr, true) gp.sock.setSockOpt(OptReuseAddr, true)
gp.sock.bindAddr(Port(gp.gossipPort)) gp.sock.bindAddr(Port(gp.gossipPort))
gp.running = true gp.running = true
var consecutiveErrors = 0
const maxConsecutiveErrors = 10
const baseRetryDelayMs = 100
while gp.running: while gp.running:
try: try:
let (data, senderAddr, _) = await gp.sock.recvFrom(65535) let (data, senderAddr, _) = await gp.sock.recvFrom(65535)
consecutiveErrors = 0 # Reset on success
if data.len > 0: if data.len > 0:
gp.handleIncomingGossip(data, senderAddr) gp.handleIncomingGossip(data, senderAddr)
except: except CatchableError as e:
# Small sleep on error to avoid spin if not gp.running:
break
inc consecutiveErrors
if consecutiveErrors >= maxConsecutiveErrors:
# Recreate socket after too many errors
try:
gp.sock.close() gp.sock.close()
# Recreate socket for next iteration except:
discard
try: try:
gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP) gp.sock = newAsyncSocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)
gp.sock.setSockOpt(OptReuseAddr, true) gp.sock.setSockOpt(OptReuseAddr, true)
gp.sock.bindAddr(Port(gp.gossipPort)) gp.sock.bindAddr(Port(gp.gossipPort))
consecutiveErrors = 0
except: except:
break 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, proc startGossip*(gp: GossipProtocol, healthIntervalMs: int = 1000,
gossipIntervalMs: int = 2000) = gossipIntervalMs: int = 2000) =
+17 -3
View File
@@ -19,6 +19,7 @@ import ../protocol/wire
import ../core/websocket import ../core/websocket
import jwt as jwtlib import jwt as jwtlib
import ../protocol/auth import ../protocol/auth
import ../protocol/ratelimit
type type
HttpServer* = ref object HttpServer* = ref object
@@ -29,6 +30,7 @@ type
metrics*: Metrics metrics*: Metrics
secretKey*: string secretKey*: string
authManager*: AuthManager authManager*: AuthManager
rateLimiter*: RateLimiter
ws*: WsServer ws*: WsServer
Metrics* = ref object Metrics* = ref object
@@ -38,13 +40,12 @@ type
selectCount*: int selectCount*: int
activeConnections*: int activeConnections*: int
proc newHttpServer*(config: BaraConfig): HttpServer = proc newHttpServerWithDb*(config: BaraConfig, db: LSMTree): HttpServer =
let dataDir = config.dataDir / "server"
let db = newLSMTree(dataDir)
let ctx = newExecutionContext(db) let ctx = newExecutionContext(db)
ctx.txnManager = newTxnManager() ctx.txnManager = newTxnManager()
let secret = config.getEffectiveJwtSecret() let secret = config.getEffectiveJwtSecret()
let ws = newWsServer(config, secret) let ws = newWsServer(config, secret)
let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient)
ctx.onChange = proc(ev: ChangeEvent) = ctx.onChange = proc(ev: ChangeEvent) =
let msg = $ev.kind & " " & ev.table let msg = $ev.kind & " " & ev.table
asyncCheck ws.broadcastToTable(ev.table, msg) asyncCheck ws.broadcastToTable(ev.table, msg)
@@ -52,8 +53,14 @@ proc newHttpServer*(config: BaraConfig): HttpServer =
HttpServer(config: config, running: false, db: db, ctx: ctx, HttpServer(config: config, running: false, db: db, ctx: ctx,
secretKey: secret, secretKey: secret,
authManager: am, authManager: am,
rateLimiter: rl,
metrics: Metrics(), ws: ws) metrics: Metrics(), ws: ws)
proc newHttpServer*(config: BaraConfig): HttpServer =
let dataDir = config.dataDir / "server"
let db = newLSMTree(dataDir)
return newHttpServerWithDb(config, db)
# ---------------------------------------------------------------------- # ----------------------------------------------------------------------
# JWT helpers # JWT helpers
# ---------------------------------------------------------------------- # ----------------------------------------------------------------------
@@ -124,6 +131,13 @@ proc queryHandler(server: HttpServer): RequestHandler =
userId = uid userId = uid
role = r 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) let body = parseJson(request.body)
if body == nil or "query" notin body: if body == nil or "query" notin body:
ctx.json(%*{"error": "Missing 'query' in request body"}, 400) ctx.json(%*{"error": "Missing 'query' in request body"}, 400)
+61 -20
View File
@@ -8,6 +8,11 @@ import std/os
import std/endians import std/endians
import std/monotimes import std/monotimes
import std/locks 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 config
import logging import logging
import ../protocol/wire import ../protocol/wire
@@ -22,6 +27,7 @@ import ../core/disttxn
import ../core/replication import ../core/replication
import ../core/sharding import ../core/sharding
import ../core/gossip import ../core/gossip
import ../protocol/ratelimit
import jwt as jwtlib import jwt as jwtlib
type type
@@ -37,12 +43,11 @@ type
clusterMembership*: ClusterMembership clusterMembership*: ClusterMembership
gossipProtocol*: GossipProtocol gossipProtocol*: GossipProtocol
tls*: TLSContext tls*: TLSContext
rateLimiter*: RateLimiter
activeConnections*: int activeConnections*: int
activeConnectionsLock*: Lock activeConnectionsLock*: Lock
proc newServer*(config: BaraConfig): Server = proc newServerWithDb*(config: BaraConfig, db: LSMTree): Server =
let dataDir = config.dataDir / "server"
let db = newLSMTree(dataDir)
let ctx = newExecutionContext(db) let ctx = newExecutionContext(db)
ctx.txnManager = newTxnManager() ctx.txnManager = newTxnManager()
var tls: TLSContext = nil var tls: TLSContext = nil
@@ -85,19 +90,40 @@ proc newServer*(config: BaraConfig): Server =
gp.onSuspect = proc(nodeId: string) {.gcsafe.} = gp.onSuspect = proc(nodeId: string) {.gcsafe.} =
cm.onNodeSuspect(nodeId) cm.onNodeSuspect(nodeId)
# Initialize rate limiter
let rl = newRateLimiter(rlaTokenBucket, config.rateLimitGlobal, config.rateLimitPerClient)
result = Server(config: config, running: false, db: db, ctx: ctx, result = Server(config: config, running: false, db: db, ctx: ctx,
txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(), txnManager: ctx.txnManager, distTxnManager: newDistTxnManager(),
replicationManager: newReplicationManager(), replicationManager: newReplicationManager(),
shardRouter: shardRouter, shardRouter: shardRouter,
clusterMembership: cm, clusterMembership: cm,
gossipProtocol: gp, gossipProtocol: gp,
tls: tls) tls: tls,
rateLimiter: rl)
initLock(result.activeConnectionsLock) initLock(result.activeConnectionsLock)
proc newServer*(config: BaraConfig): Server =
let dataDir = config.dataDir / "server"
let db = newLSMTree(dataDir)
return newServerWithDb(config, db)
# ---------------------------------------------------------------------- # ----------------------------------------------------------------------
# Wire Protocol Helpers # 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 = proc readUint32BE(data: string, pos: int): uint32 =
if pos + 4 > data.len: if pos + 4 > data.len:
raise newException(ValueError, "readUint32BE: index out of bounds") 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 key = data[0..<nullPos]
let value = data[nullPos+1..^1] let value = data[nullPos+1..^1]
if value.len > 0: if value.len > 0:
server.db.put(key, cast[seq[byte]](value)) server.db.put(key, stringToBytes(value))
else: else:
server.db.delete(key) server.db.delete(key)
await client.send("ACK " & $lsn & "\n") 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: if not authenticated and header.kind != mkAuth:
let err = makeErrorMessage(header.requestId, 401, "Authentication required") 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 continue
case header.kind case header.kind
of mkAuth: of mkAuth:
let tokenStr = parseAuthMessage(cast[seq[byte]](payload)) let tokenStr = parseAuthMessage(stringToBytes(payload))
let (valid, userId, role) = verifyToken(secret, tokenStr) let (valid, userId, role) = verifyToken(secret, tokenStr)
if valid: if valid:
authenticated = true authenticated = true
connCtx.currentUser = userId connCtx.currentUser = userId
connCtx.currentRole = role connCtx.currentRole = role
let okMsg = makeAuthOkMessage(header.requestId) let okMsg = makeAuthOkMessage(header.requestId)
await client.send(cast[string](okMsg)) await client.send(bytesToString(okMsg))
info("Client " & $clientId & " authenticated as " & userId) info("Client " & $clientId & " authenticated as " & userId)
else: else:
let err = makeErrorMessage(header.requestId, 403, "Invalid token") let err = makeErrorMessage(header.requestId, 403, "Invalid token")
await client.send(cast[string](err)) await client.send(bytesToString(err))
of mkQuery: of mkQuery:
var pos = 0 var pos = 0
let queryStr = readString(cast[seq[byte]](payload), pos) let queryStr = readString(stringToBytes(payload), pos)
info("[" & $clientId & "] Query: " & queryStr) info("[" & $clientId & "] Query: " & queryStr)
# Shard-aware routing: check if this node should handle the write # 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: if localShards.len == 0:
shardCheck = false shardCheck = false
let err = serializeError(3, "Node not assigned to any shard", header.requestId) let err = serializeError(3, "Node not assigned to any shard", header.requestId)
await client.send(cast[string](err)) await client.send(bytesToString(err))
break break
if shardCheck: if shardCheck:
@@ -484,15 +518,15 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
if success: if success:
let dataMsg = serializeResult(result, header.requestId) let dataMsg = serializeResult(result, header.requestId)
await client.send(cast[string](dataMsg)) await client.send(bytesToString(dataMsg))
let completeMsg = serializeComplete(result.affectedRows, header.requestId) let completeMsg = serializeComplete(result.affectedRows, header.requestId)
await client.send(cast[string](completeMsg)) await client.send(bytesToString(completeMsg))
else: else:
let errorMsg = serializeError(1, errorMsg, header.requestId) let errorMsg = serializeError(1, errorMsg, header.requestId)
await client.send(cast[string](errorMsg)) await client.send(bytesToString(errorMsg))
of mkQueryParams: of mkQueryParams:
let (queryStr, params) = readQueryParamsMessage(cast[seq[byte]](payload)) let (queryStr, params) = readQueryParamsMessage(stringToBytes(payload))
info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)") info("[" & $clientId & "] QueryParams: " & queryStr & " (" & $params.len & " params)")
let startTicks = getMonoTime().ticks() let startTicks = getMonoTime().ticks()
@@ -504,26 +538,26 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
if success: if success:
let dataMsg = serializeResult(result, header.requestId) let dataMsg = serializeResult(result, header.requestId)
await client.send(cast[string](dataMsg)) await client.send(bytesToString(dataMsg))
let completeMsg = serializeComplete(result.affectedRows, header.requestId) let completeMsg = serializeComplete(result.affectedRows, header.requestId)
await client.send(cast[string](completeMsg)) await client.send(bytesToString(completeMsg))
else: else:
let errorMsg = serializeError(1, errorMsg, header.requestId) let errorMsg = serializeError(1, errorMsg, header.requestId)
await client.send(cast[string](errorMsg)) await client.send(bytesToString(errorMsg))
of mkPing: of mkPing:
var pongMsg = WireMessage( var pongMsg = WireMessage(
header: MessageHeader(kind: mkPong, length: 0, requestId: header.requestId), header: MessageHeader(kind: mkPong, length: 0, requestId: header.requestId),
payload: @[], payload: @[],
) )
await client.send(cast[string](serializeMessage(pongMsg))) await client.send(bytesToString(serializeMessage(pongMsg)))
of mkClose: of mkClose:
break break
else: else:
let errorMsg = serializeError(2, "Unsupported message kind: " & $header.kind, header.requestId) 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: except Exception as e:
errorMsg("Client " & $clientId & " error: " & e.msg) errorMsg("Client " & $clientId & " error: " & e.msg)
@@ -537,6 +571,11 @@ proc handleClient(server: Server, client: AsyncSocket, clientId: int) {.async.}
info("Client " & $clientId & " disconnected") info("Client " & $clientId & " disconnected")
client.close() 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.} = proc run*(server: Server) {.async.} =
server.running = true server.running = true
var clientId = 0 var clientId = 0
@@ -548,6 +587,7 @@ proc run*(server: Server) {.async.} =
let sock = newAsyncSocket() let sock = newAsyncSocket()
sock.setSockOpt(OptReuseAddr, true) sock.setSockOpt(OptReuseAddr, true)
setTcpNoDelay(sock)
sock.bindAddr(Port(server.config.port), server.config.address) sock.bindAddr(Port(server.config.port), server.config.address)
sock.listen() sock.listen()
if server.config.tlsEnabled: if server.config.tlsEnabled:
@@ -556,6 +596,7 @@ proc run*(server: Server) {.async.} =
info("BaraDB listening on " & server.config.address & ":" & $server.config.port) info("BaraDB listening on " & server.config.address & ":" & $server.config.port)
while server.running: while server.running:
let client = await sock.accept() let client = await sock.accept()
setTcpNoDelay(client)
acquire(server.activeConnectionsLock) acquire(server.activeConnectionsLock)
var shouldAccept = true var shouldAccept = true
try: try:
+11
View File
@@ -5,6 +5,11 @@ import std/strutils
import std/tables import std/tables
import std/base64 import std/base64
import std/sets 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 config
import jwt as jwtlib import jwt as jwtlib
@@ -307,16 +312,22 @@ proc handleConnection(server: WsServer, client: AsyncSocket) {.async.} =
inc server.nextId inc server.nextId
asyncCheck server.handleWsClient(client, 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.} = proc run*(server: WsServer, port: int = 9471) {.async.} =
server.running = true server.running = true
let sock = newAsyncSocket() let sock = newAsyncSocket()
sock.setSockOpt(OptReuseAddr, true) sock.setSockOpt(OptReuseAddr, true)
setTcpNoDelay(sock)
sock.bindAddr(Port(port)) sock.bindAddr(Port(port))
sock.listen() sock.listen()
echo "BaraDB WebSocket listening on port ", port echo "BaraDB WebSocket listening on port ", port
while server.running: while server.running:
let client = await sock.accept() let client = await sock.accept()
setTcpNoDelay(client)
asyncCheck server.handleConnection(client) asyncCheck server.handleConnection(client)
proc stop*(server: WsServer) = proc stop*(server: WsServer) =
+9 -7
View File
@@ -230,17 +230,21 @@ proc main() =
warn("Failed to generate self-signed certificate. TLS disabled.") warn("Failed to generate self-signed certificate. TLS disabled.")
config.tlsEnabled = false 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 # 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 spawn httpServer.run(config.port + 440) # HTTP port = TCP port + 440
# Start background compaction loop # Start background compaction loop
let cm = newCompactionManager(httpServer.db) let cm = newCompactionManager(sharedDb)
asyncCheck cm.startCompactionLoop() asyncCheck cm.startCompactionLoop()
# Create TCP server (initialization is synchronous, run is async) # Create TCP server (initialization is synchronous, run is async)
let localId {.used.} = if config.raftNodeId.len > 0: config.raftNodeId else: "node-" & $config.port 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 # Start Raft cluster if enabled
if config.raftEnabled: if config.raftEnabled:
@@ -251,11 +255,9 @@ proc main() =
if cmd == "put": if cmd == "put":
let parts = cast[string](data).split("\x00") let parts = cast[string](data).split("\x00")
if parts.len >= 2: if parts.len >= 2:
httpServer.db.put(parts[0], cast[seq[byte]](parts[1])) sharedDb.put(parts[0], cast[seq[byte]](parts[1]))
tcpServer.db.put(parts[0], cast[seq[byte]](parts[1]))
elif cmd == "delete": elif cmd == "delete":
httpServer.db.delete(cast[string](data)) sharedDb.delete(cast[string](data))
tcpServer.db.delete(cast[string](data))
# Wire RAFT ↔ DistTxn # Wire RAFT ↔ DistTxn
wireRaftDistTxn(raftNode, tcpServer) wireRaftDistTxn(raftNode, tcpServer)