diff --git a/.gitignore b/.gitignore index 6a37a738..d6ea3d06 100644 --- a/.gitignore +++ b/.gitignore @@ -12,6 +12,9 @@ AGENTS.md +# Superpowers scratch (plans/specs), not part of the repo +docs/superpowers/ + # Prerequisites *.d diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 00000000..79384f61 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,76 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## What This Repo Is + +EloqKV is a distributed, Redis/Valkey-compatible database with ACID transactions. This repo contains the **Redis API layer** only; the transaction/storage engine lives in the `data_substrate` submodule (GitHub repo `eloqdata/tx_service`), which has its own CLAUDE.md. Most non-protocol work (concurrency control, transactions, storage, WAL) happens in the submodule. + +After cloning: `git submodule update --init --recursive`. + +## Technical Docs — Read These First + +`docs/` contains module-by-module design documentation (index: `docs/README.md`). **Before working on an unfamiliar module, read its doc**: `01` overview/bootstrap, `02` command processing & transactions, `03` data model (objects/commands/catalog), `04` Lua/pub-sub/blocking commands, `05` namespaces, `06` vector search, `07` RDB-AOF interop & tools. Engine internals are documented in `data_substrate/docs/`. + +**Maintenance rule: when a code change alters behavior described in `docs/`, update the corresponding doc in the same change.** Each doc lists the source files it covers. + +## Common Commands + +```bash +# Configure + build (out-of-source; bld/ and install/ are gitignored output dirs) +mkdir -p bld && cd bld +cmake .. -DCMAKE_BUILD_TYPE=Debug \ + -DCMAKE_EXPORT_COMPILE_COMMANDS=ON \ + -DWITH_DATA_STORE=ELOQDSS_ROCKSDB \ + -DWITH_LOG_STATE=ROCKSDB \ + -DOPEN_LOG_SERVICE=OFF \ + -DCMAKE_INSTALL_PREFIX=../install +cmake --build . --parallel 16 +cmake --install . --prefix ../install + +# Run a single-node server (port 6379, foreground) +./install/bin/eloqkv --config=eloqkv.ini + +# Run a single TCL test file (e.g. tests/unit/eloq/hash.tcl) +# against the running server; loop over tests/unit/eloq/*.tcl for the full suite +tclsh tests/test_helper.tcl --host 127.0.0.1 --port 6379 \ + --tags -needs:repl --tags -needs:config-maxmemory --tags -needs:debug \ + --tags -needs:redis_config --tags -needs:redis_expire --tags -needs:slow_test \ + --tags -needs:support_cmd_later --tags -needs:cluster_mode \ + --single /unit/eloq/hash + +# Format changed C/C++ files (clang-format-18; repeat inside data_substrate/) +git diff --name-only --diff-filter=ACMR \ + | grep -E '\.(cpp|cc|cxx|c|hpp|hh|hxx|h)$' | xargs -r clang-format-18 -i +``` + +Verify behavior manually with `redis-cli -h 127.0.0.1 -p 6379`. + +Notes: +- Dependencies: use the `eloqdata/eloq-dev-ci-ubuntu2404` Docker image or run `scripts/install_dependency_ubuntu2404.sh` (ubuntu2404 preferred). +- Debug builds enable fault injection (`WITH_FAULT_INJECT`); add `--tags -needs:fault_inject` to the TCL command on non-Debug builds. +- Key CMake options: `WITH_DATA_STORE` (storage backend: `ELOQDSS_ELOQSTORE` default, `ELOQDSS_ROCKSDB` common for local dev, also DynamoDB/BigTable/RocksDB-Cloud variants), `WITH_LOG_STATE`, `WITH_LOG_SERVICE`, `OPEN_LOG_SERVICE`, `BUILD_ELOQKV_AS_LIBRARY` (build as library for a converged binary instead of the `eloqkv` executable). + +## Architecture + +Command flow, top to bottom: + +1. **Entry**: `src/redis_server.cpp` (`eloqkv` executable) starts a brpc server with `EloqKV::RedisServiceImpl` (`src/redis_service.cpp`, ~the largest file in the repo) as the Redis-protocol service. Config comes from gflags / an ini file (`eloqkv.ini` template at repo root). +2. **Command parsing**: RESP requests are parsed into command objects (`include/redis_command.h`, `src/redis_command.cpp`). Replies are written through `redis_replier.h` / `output_handler.h`. +3. **Data objects**: each Redis type is a `txservice::TxObject` subclass — `redis_string_object`, `redis_hash_object`, `redis_list_object`, `redis_set_object`, `redis_zset_object` (see `include/redis_object.h` for the base). Commands are applied to these objects inside the tx service, not in the protocol layer. +4. **Into the engine**: `RedisServiceImpl` wraps commands in `txservice` requests (`ObjectCommandTxRequest`, `MultiObjectCommandTxRequest`, defined in `data_substrate/tx_service/include/tx_request.h`) and drives them through a `TransactionExecution` state machine. Keys are `EloqKey` (`include/eloqkv_key.h`); the schema/table mapping is `eloqkv_catalog_factory.h`. +5. **Engine**: the `data_substrate` submodule shards data into CcShards, each owned by a TxProcessor pinned to a core, and handles concurrency control, WAL (log service), checkpointing, and cold-data storage. See `data_substrate/CLAUDE.md` and the module design docs in `data_substrate/docs/`. + +Auxiliary subsystems in this layer: +- **Lua/scripting**: `include/lua_interpreter.h`, vendored `lua/`. +- **Pub/Sub**: `include/pub_sub_manager.h`. +- **Namespaces** (multi-tenancy): `include/namespace/`, `src/namespace/`. +- **Vector search**: `include/vector/`, `src/vector/`. +- **Tools**: `src/tools/eloqkv2aof`, `src/tools/eloqkv2rdb` (export to Redis AOF/RDB formats). +- Vendored Redis C sources (`src/redis/`), `crcspeed/`, `fpconv/`. + +Threading model to keep in mind: request handlers run on brpc **bthreads** (coroutines on a custom brpc fork), while CC request `Execute()` runs on shard/TxProcessor context. Do not introduce `bthread::Mutex`/`ConditionVariable` shared between shard-side `Execute()` and bthread waiters — this can permanently deadlock a brpc worker. Use `std::atomic` state + `bthread_usleep` backoff polling instead (see data_substrate's CLAUDE.md for details). + +## Code Style + +Google C++ style, enforced by clang-format-18 (`.clang-format` at repo root). C++20. Naming: functions/classes `MyName`, locals `my_name`, members `my_name_`, enumerators `kEnumName`. Full project conventions: `data_substrate/style_guide.md`. diff --git a/docs/01-architecture-overview.md b/docs/01-architecture-overview.md new file mode 100644 index 00000000..756d6f0f --- /dev/null +++ b/docs/01-architecture-overview.md @@ -0,0 +1,69 @@ +# Architecture Overview + +EloqKV is a Redis/Valkey-compatible distributed database. This repo implements the **Redis protocol layer**: a brpc-based RESP server (`RedisServiceImpl`) that parses commands into command objects, applies them to Redis-type data objects (`TxObject` subclasses), and drives everything through the transaction engine in the `data_substrate/` submodule (documented separately under `data_substrate/docs/`). The layer registers itself with the engine as `TableEngine::EloqKv`, supplying a `CatalogFactory` so the engine can create EloqKV's concrete keys, records, and cc maps without knowing anything about Redis. + +## Component Map + +| Area | Key files | Doc | +|---|---|---| +| Bootstrap / main | `src/redis_server.cpp` | this file | +| Service & dispatch | `src/redis_service.cpp` (~6.5k lines), `include/redis_service.h`, `redis_handler.*`, `redis_connection_context.*` | [02](02-command-processing.md) | +| Output / errors / stats | `redis_replier.*`, `output_handler.h`, `redis_errors.*`, `redis_stats.*` | [02](02-command-processing.md) | +| Commands | `include/redis_command.h` (~8k lines), `src/redis_command.cpp` (~21k lines) | [03](03-data-model.md) | +| Data objects | `redis_object.h`, `redis_{string,hash,list,set,zset}_object.*` | [03](03-data-model.md) | +| Engine plug-in | `eloqkv_key.*` (EloqKey = TxKey impl), `eloqkv_catalog_factory.*` | [03](03-data-model.md) | +| Lua scripting | `lua_interpreter.*`, `lua_output_handler.h`, vendored `lua/` | [04](04-scripting-pubsub-blocking.md) | +| Pub/Sub | `pub_sub_manager.*` | [04](04-scripting-pubsub-blocking.md) | +| Namespaces (multi-tenancy) | `include/namespace/`, `src/namespace/` | [05](05-namespaces.md) | +| Vector search | `include/vector/`, `src/vector/` | [06](06-vector-search.md) | +| RDB/AOF interop & tools | `redis_rdb_restore.*`, `src/tools/eloqkv2{rdb,aof}/`, `crcspeed/`, `fpconv/` | [07](07-persistence-and-tools.md) | +| Vendored Redis C utils | `src/redis/` (dict, sha1, siphash, zmalloc, commands…) | — | +| Engine | `data_substrate/` submodule | `data_substrate/docs/` | + +## Process Bootstrap (`src/redis_server.cpp`) + +`main()` wires the protocol layer into the engine in four ordered steps (mirroring `DataSubstrate`'s lifecycle, see `data_substrate/docs/01-architecture-overview.md`): + +1. **`DataSubstrate::Instance().Init(config_file)`** — after `ConvertEloqkvFlagsToTxFlags()` translates eloqkv-named flags (`ip`/`port`/`ip_port_list`/standby/voter lists) into the engine's `tx_*` flags. gflags override ini values throughout (`CheckCommandLineFlagIsDefault` pattern). +2. **`RedisServiceImpl::Init(server)`** (`src/redis_service.cpp:219`) — builds the prebuilt table list and calls `DataSubstrate::RegisterEngine(TableEngine::EloqKv, &catalog_factory, nullptr, prebuilt_tables, engine_metrics, eloqkv_publish_func)`: + - one engine table per Redis database: `data_table_0` … `data_table_` (`databases` from config, default 16), each `TableType::Primary` + `TableEngine::EloqKv`; + - two namespace system tables: `__ns_0` (namespace registry) and `ns_data_0` (shared namespace data) — see [05-namespaces.md](05-namespaces.md); + - per-command duration/total metrics and read/write aggregates; + - `eloqkv_publish_func` so the engine can deliver cross-node PUBLISH messages back into this layer ([04](04-scripting-pubsub-blocking.md)). +3. **`DataSubstrate::Instance().Start()`** — boots log service, storage handler, and the tx service (TxProcessors, Sharder, checkpointer…). +4. **`RedisServiceImpl::Start(server)`** (`src/redis_service.cpp:635`) then brpc server start — the service object is installed as `brpc::ServerOptions::redis_service` (the brpc Redis protocol entry point; brpc owns and deletes it), `num_threads` is pinned to the engine's `bthread_concurrency` (= `core_number`), builtin brpc services are disabled, and TLS is configured when enabled. The server listens on `eloqkv_port` (default 6379). + +Shutdown is the reverse: stop accepting, `RedisServiceImpl::Stop()`, `DataSubstrate::Shutdown()`. + +## How a command becomes engine work (one paragraph; details in [02](02-command-processing.md)/[03](03-data-model.md)) + +brpc parses RESP on a bthread and calls into `RedisServiceImpl`; the command name is looked up in the dispatch table; a command object is constructed (parse/validate args); the service obtains a `TransactionExecution` from the engine (`NewTxm`, pinned to the current core's shard) and submits an `ObjectCommandTxRequest` / `MultiObjectCommandTxRequest`; the engine routes the command to the owner shard (local or remote) where it executes against the in-memory `TxObject`; results flow back through the request's `TxResult`, and the bthread renders the RESP reply via the replier. Single commands auto-commit; `MULTI/EXEC` and the session-style `BEGIN/COMMIT/ROLLBACK` map onto one engine transaction. + +Because connection bthreads both *submit* transactions and *drive* the engine (brpc workers double as tx processors — `data_substrate/docs/02-threading-model.md`), this layer must follow the engine's threading contract: never block a bthread on a synchronization primitive shared with shard-side `Execute()` code; use the yield/resume functors on `TxRequest` or atomic + `bthread_usleep` patterns. + +## Configuration Surface (layer-specific) + +Defined in `src/redis_server.cpp` / `src/redis_service.cpp`; engine flags are listed in `data_substrate/docs/01-architecture-overview.md`. + +| Flag | Default | Meaning | +|---|---|---| +| `config` | "" | ini file path (same file is handed to the engine) | +| `eloqkv_port` | 6379 | RESP listen port | +| `ip` / `port` / `ip_port_list` / `standby_ip_port_list` / `voter_ip_port_list` | — | translated to engine `tx_*` flags at startup | +| `databases` (ini `[local]`) | 16 | number of Redis databases = number of `data_table_` tables | +| `requirepass` (ini) | "" | AUTH password | +| `cluster_mode` | — | strict Redis Cluster compatibility behavior | +| `txn_isolation_level` / `protocol` / `isolation_level` | — | engine isolation/cc-protocol selection for txs ([02](02-command-processing.md)) | +| `retry_on_occ_error` | — | auto-retry policy for OCC conflicts | +| `enable_tls` / `tls_cert_file` / `tls_key_file` | off | TLS on the RESP port | +| `slow_log_threshold` / `slow_log_max_length` | — | SLOWLOG | +| `enable_redis_stats`, `enable_cmd_sort` | — | INFO/stats behavior | +| `cc_notify` | — | notify-based (vs polling) wakeup between layer and engine | +| `vector_index_worker_num` | — | vector index worker threads ([06](06-vector-search.md)) | +| `maxclients`, `enable_io_uring`, … | — | engine flags commonly set from eloqkv configs | + +## Build shapes + +- Default: `eloqkv` executable (`src/redis_server.cpp` + the `RESELOQ` library). +- `BUILD_ELOQKV_AS_LIBRARY=ON`: builds `eloqkv_lib` for converged multi-engine binaries (the engine's `EnableEngine`/`WaitForEnabledEnginesRegistered` flow exists for this). +- Offline tools `eloqkv_to_rdb` / `eloqkv_to_aof` ([07](07-persistence-and-tools.md)); storage backend selected by `WITH_DATA_STORE` (see repo `CLAUDE.md`). diff --git a/docs/02-command-processing.md b/docs/02-command-processing.md new file mode 100644 index 00000000..bc85c332 --- /dev/null +++ b/docs/02-command-processing.md @@ -0,0 +1,344 @@ +# 02 — Command Processing & the Service Layer + +EloqKV's service layer is a brpc-based RESP server (`RedisServiceImpl`, a subclass of the custom +brpc fork's `brpc::RedisService`) that turns Redis commands into Data Substrate transactions. +Every command is parsed on the brpc worker bthread that read it from the socket, looked up in a +flat name→handler table, parsed into a command object (`RedisCommand` / `RedisMultiObjectCommand` / +`DirectCommand` / `CustomCommand`), wrapped in a `TxRequest`, and executed through a +`TransactionExecution` (txm) obtained from the engine's `TxService` — one auto-committed txm per +simple command, or a long-lived txm for MULTI/EXEC, Lua scripts, and the EloqKV-specific +interactive `BEGIN`/`COMMIT`/`ROLLBACK` sessions. Replies are rendered through an `OutputHandler` +abstraction (RESP2 via `RedisReplier`, or Lua tables via `LuaOutputHandler`). This doc covers the +service layer only; engine internals are in `data_substrate/docs/` (esp. `02-threading-model.md`, +`04-transaction-execution.md`). Command/object semantics are in `docs/03-data-model.md`. + +## File map + +| Area | Files | +|---|---| +| Service impl, dispatch, tx execution | `include/redis_service.h`, `src/redis_service.cpp` | +| Per-command handlers (the command table) | `include/redis_handler.h`, `src/redis_handler.cpp` | +| Per-connection state | `include/redis_connection_context.h`, `src/redis_connection_context.cpp` | +| Command objects + parsers | `include/redis_command.h`, `src/redis_command.cpp` (see docs/03) | +| Process entry point | `src/redis_server.cpp` | +| Reply rendering | `include/output_handler.h`, `include/redis_replier.h`, `src/redis_replier.cpp`, `include/lua_output_handler.h` | +| Errors / stats | `include/redis_errors.h`, `include/redis_stats.h`, `src/redis_stats.cpp` | + +## 1. Process bootstrap + +`main()` (`src/redis_server.cpp:425`) runs a fixed sequence: + +1. Parse gflags + INI config (`--config`). User-facing flags (`ip`, `port`, `ip_port_list`, + `standby_ip_port_list`, `voter_ip_port_list`) are translated to engine flags (`tx_ip`, + `tx_port`, `tx_ip_port_list`, ...) with **tx ports = redis port + 10000** + (`src/redis_server.cpp:152-423`; reverse mapping `TxPortToRedisPort`, + `include/redis_service.h:185`). +2. `DataSubstrate::Instance().Init(config_file)` then `EnableEngine(TableEngine::EloqKv)`. +3. `RedisServiceImpl::Init()` (`src/redis_service.cpp:219`) — *before* the engine starts: + registers the EloqKv engine with `DataSubstrate::RegisterEngine` along with prebuilt tables + (`data_table_0..N-1` for `databases` logical DBs, default 16, plus the two namespace tables + `__ns_0` and `ns_data_0`; `src/redis_service.cpp:232-281`), the catalog factory, per-command + metric definitions, and the pub/sub publish callback. It also parses isolation/protocol flags + (§3), TLS settings (`enable_tls`, `tls_cert_file`, `tls_key_file`; TLS forces io_uring off, + `src/redis_service.cpp:528-531`), and builds the command table via `AddHandlers()`. +4. `DataSubstrate::Instance().Start()` — engine up (TxService, store handler, log service). +5. `RedisServiceImpl::Start()` (`src/redis_service.cpp:635`) — second phase: grabs + `TxService`/`DataStoreHandler` pointers, sizes per-core slow-log structures, computes the + listen address, starts the metrics collector thread and namespace GC daemon. In `bootstrap` + mode it exits the process right here after table creation (`src/redis_service.cpp:704-721`). +6. brpc `Server::Start()` with `server_options.redis_service = redis_service_impl` (ownership + transfers to the server) and optional SSL options; then `RunUntilAskedToQuit()` + (`src/redis_server.cpp:502-548`). + +## 2. Request path end-to-end + +```text +client TCP/RESP + │ + ▼ brpc worker bthread (EloqData brpc fork) +ParseRedisMessage / ConsumeCommand brpc fork: src/brpc/policy/redis_protocol.cpp + │ - RESP parse into args[]; args[0] lowercased by the parser + │ - pins the task to its TaskGroup (SetBoundGroup) when + │ brpc_worker_as_ext_processor is on + ▼ +RedisServiceImpl::DispatchCommand src/redis_service.cpp:5924 + │ - refresh namespace binding (ctx->ns_meta/ns_id), NamespaceGuard + │ - AuthRequired() → "NOAUTH" (5877); shutdown check + │ - in MULTI? → MultiTransactionHandler::Run (queue) + │ - else FindCommandHandler(args[0]) → handler->Run(...) + ▼ +CommandHandler::Run src/redis_handler.cpp (e.g. GET at 747) + │ - ParseCommand(args) → (EloqKey, Command) or error reply + │ - txm = ctx->txm (BEGIN session) or NewTxm(iso_level_, cc_protocol_) + ▼ +RedisServiceImpl::ExecuteCommand src/redis_service.cpp:2657 (single-key) + │ - key-size guard; builds ObjectCommandTxRequest + │ (MultiObjectCommandTxRequest for multi-key, 2716) + │ - optionally attaches yield/resume functors (§5) + ▼ +ExecuteTxRequest / ExecuteMultiObjTxRequest src/redis_service.cpp:4471 / 4519 + │ - SendTxRequestAndWaitResult → txm->Execute(req); req->Wait() + │ - error mapping: MOVED / READONLY / engine error text (1220-1331) + ▼ +engine: TransactionExecution → CcRequest see data_substrate/docs/03, 04 + ▼ +cmd->OutputResult(output) command object renders its result + ▼ +RedisReplier → brpc::RedisReply → RESP2 bytes back on the socket +``` + +Commands are executed **inline in the socket-parse path** — there is no per-command bthread +spawn; pipelined commands in one read buffer are consumed in a loop and their replies batched +into a single socket write (brpc fork `redis_protocol.cpp`, `ConsumeCommand` loop). + +Command objects own argument validation: each `ParseCommand` in `src/redis_command.cpp` +writes protocol errors (wrong arity, bad integer, ...) straight to the `OutputHandler` and +returns `success=false`, in which case no txm work happens. Limits enforced at this layer: +key ≤ 32 MB (2 KB for the EloqStore backend) and object ≤ 256 MB +(`src/redis_service.cpp:193-199`, checked at 2665 and in write-command parsers). + +## 3. Connection state machine + +One `RedisConnectionContext` per socket, created by `NewConnectionContext` +(`src/redis_service.cpp:5917`) and owned by brpc's per-socket parsing context. Key fields +(`include/redis_connection_context.h:61-170`): + +| Field | Meaning | +|---|---| +| `db_id` | logical DB selected by `SELECT` (0..`databases`-1); maps to table `data_table_` via `RedisTableName` (`src/redis_service.cpp:5863`). `SELECT` is rejected inside a non-default namespace (`RD_ERR_SELECT_FORBIDDEN_IN_NS`, `src/redis_command.cpp:1736`) | +| `authenticated` | set by `AUTH`; password compared to `requirepass`, **or** interpreted as a namespace token, which also binds `ns`/`ns_meta`/`ns_id` (`src/redis_command.cpp:1495-1516`; see docs/05) | +| `in_multi_transaction` + `multi_transaction_handler` | MULTI queueing state machine (§4) | +| `txm` | non-null while a `BEGIN` session transaction is open | +| `subscribed_channels` / `subscribed_patterns` | pub/sub state (docs/04); destructor unsubscribes all | +| `scan_cursors`, `bucket_scan_cursors` | server-side SCAN cursor caches (LRU of 100 / per-db bucket save points) | +| `connection_name`, `lib_name`, `lib_ver` | `CLIENT SETNAME/SETINFO` | +| `output`/`arena` | connection-owned `RedisReply` used for out-of-band pushes (pub/sub `FlushOutput`, `src/redis_connection_context.cpp:90`) | + +Auth gating: when `requirepass` is set, every command except `auth`, `hello`, `quit`, `reset` +and `NAMESPACE CURRENT` returns `NOAUTH` until authenticated (`src/redis_service.cpp:5895-5914`). +Note there is **no HELLO handler registered**, so `HELLO` (and thus RESP3 negotiation) returns +"unknown command" despite being auth-exempt — the server speaks RESP2 only (§7). + +Disconnect: the context destructor aborts a dangling `BEGIN` txm with `AbortTx` and bumps the +closed-connection stat (`src/redis_connection_context.cpp:37-54`). A pending MULTI handler's +destructor likewise aborts its txm (`src/redis_handler.cpp:1680`). + +Subscribe mode: subscriptions are tracked, but `DispatchCommand` performs **no +"subscribe-mode only" command restriction** the way vanilla Redis does — regular commands still +execute on a subscribed connection (verified absence in `src/redis_service.cpp:5924-6191`). + +## 4. Transaction semantics + +Two independent (isolation, protocol) pairs are configured at startup +(`src/redis_service.cpp:129-141, 391-487`): + +| Mode | Flags (defaults) | Used by | +|---|---|---| +| Simple commands | `--isolation_level=ReadCommitted`, `--protocol=OccRead` → statics `RedisCommandHandler::iso_level_`/`cc_protocol_` (`include/redis_handler.h:45-50, 92-96`) | every auto-commit command | +| Transactions | `--txn_isolation_level=RepeatableRead`, `--txn_protocol=OCC` → `txn_isolation_level_`/`txn_protocol_` | MULTI/EXEC, WATCH, Lua scripts, BEGIN sessions | + +Accepted values: `ReadCommitted`/`RepeatableRead` and `OCC`/`OccRead`/`Locking`. + +**Auto-commit (default).** Handler calls `NewTxm()` and `ExecuteCommand(..., auto_commit=true)`; +the engine commits/aborts the tx as part of processing the `ObjectCommandTxRequest` itself +(`auto_commit_` flag on the request). Multi-step commands (`MultiObjectCommandTxRequest`, e.g. +LMPOP/SMOVE/blocking pops) suppress auto-commit on middle steps and the service layer issues the +final `CommitTx`/`AbortTx` based on `IsPassed()` (`src/redis_service.cpp:4538-4613`). + +**MULTI/EXEC.** `MultiHandler::Run` returns `REDIS_CMD_CONTINUE` (`src/redis_handler.cpp:1603`), +which makes `DispatchCommand` create a `MultiTransactionHandler` and route every subsequent +command to it (`src/redis_service.cpp:5999-6086`). Queued commands are parsed eagerly by +`ParseMultiCommand` (`src/redis_command.cpp:8351`) into a +`std::variant`; parse failure marks the tx poisoned +(`RD_ERR_EXEC_ABORT_FOR_PREV_ERROR`) and EXEC aborts. On `EXEC`, `MultiExec` +(`src/redis_service.cpp:2055`) lazily creates **one** txm (`txn_isolation_level_`, +`txn_protocol_`), optionally key-sorts the requests to reduce deadlocks +(`--enable_cmd_sort`, off by default; 2109), executes them sequentially, then `CommitTx`. +Divergence from vanilla Redis: any runtime error **aborts the whole transaction and returns +nil** rather than executing remaining commands (2159-2352). `DISCARD` aborts and drops the +queue. Commands not whitelisted in `ParseMultiCommand` get +"Unsupported command in MULTI" (`src/redis_command.cpp:10499`). + +**WATCH.** Handled before MULTI begins: `watch`/`unwatch` outside a queue create the +`MultiTransactionHandler` early (`src/redis_service.cpp:6014-6041`). `WatchKeys` +(`src/redis_handler.cpp:1849`) creates the txm immediately and reads the keys under it with a +`MultiObjectCommandTxRequest` so RepeatableRead+OCC validation detects modification at commit — +a changed watched key surfaces as an OCC failure and EXEC replies nil. `WATCH` inside MULTI or +inside a BEGIN session is an error. Watching disables OCC retry (`tx_retrieable_ = false`). + +**Retry on OCC conflict.** With `--retry_on_occ_error` (default true), EXEC and Lua scripts are +transparently re-parsed and re-executed when commit fails with +`OCC_BREAK_REPEATABLE_READ`/`WRITE_WRITE_CONFLICT` (`src/redis_handler.cpp:1809-1827`, +`src/redis_service.cpp:2601-2633`) — unless keys were WATCHed. + +**BEGIN/COMMIT/ROLLBACK (interactive sessions, EloqKV-specific).** `BeginHandler` stores a fresh +txm in `ctx->txm` (`src/redis_handler.cpp:1619`). While set, every ordinary handler detects +`in_tx = ctx->txm != nullptr` and runs its command on that txm with `auto_commit=false`, +`always_redirect=true`, and `cmd.SetVolatile()` (results may be buffered on the cc entry until +commit; e.g. `src/redis_handler.cpp:761-771`). Commands execute and reply immediately — +unlike MULTI there is no queueing. `COMMIT`/`ROLLBACK` call `CommitTx`/`AbortTx` and clear +`ctx->txm` (`src/redis_handler.cpp:1637, 1661`); commit failure is reported as +`ERR `. BEGIN inside MULTI or vice versa is rejected. + +**Lua (EVAL/EVALSHA).** One txm per script (`txn_isolation_level_`/`txn_protocol_`); each +`redis.call` goes through `GenericCommand` (`src/redis_service.cpp:3024`), a big switch that +re-parses and executes the command on the script's txm with `auto_commit=false`; the script +commits at the end (`EvalLua`, `src/redis_service.cpp:2483`). Details in docs/04. + +## 5. txm acquisition and the bthread threading contract + +`NewTxm` (`src/redis_service.cpp:2021`) binds the new `TransactionExecution` to the **current +bthread's task group**: with `EXT_TX_PROC_ENABLED` it calls +`tx_service_->NewTx(bthread::tls_task_group->group_id_)`, so the txm lives on the TxProcessor/ +CcShard paired 1:1 with this brpc worker (see `data_substrate/docs/02-threading-model.md`). +The brpc fork pins the parsing task to its group for the duration of command processing +(`SetBoundGroup` in `redis_protocol.cpp`), keeping shard-local accesses thread-safe. + +Waiting for results uses two schemes (`data_substrate/tx_service/include/tx_request.h:94-150`): + +- **cc_notify path** — `ExecuteCommand` attaches `yield_func = bthread_block()` and + `resume_func = resume_group->resume_bound_task(resume_tid)` to the `ObjectCommandTxRequest` + when `FLAGS_cc_notify && (!auto_commit || skip_wal_) && ctx->txm == nullptr` + (`src/redis_service.cpp:2677-2702`). The bthread drives the txm itself + (`ForceExternalForwardOnce` → `txm->ExternalForward()`), parks via `bthread_block`, and the + finishing CC request resumes it **on the original task group** — the bthread never migrates, + so it cannot land on a "wrong" group. This covers Lua-driven commands and auto-commit + commands when WAL is disabled. +- **default path** — no functors; `TxRequest::Wait()` loops `txm_->ExternalForward()` + + `tx_result_.Wait()` until the result is set. Used for auto-commit commands with WAL on, for + queued MULTI requests, and for BEGIN-session commands (excluded from cc_notify by + `ctx->txm == nullptr`). + +`SendTxRequest` (`src/redis_service.cpp:1333`) maps a rejected `txm->Execute()` (committed/ +aborted txm) to `TX_REQUEST_TO_COMMITTED_ABORTED_TX`. + +## 6. Cluster behavior: redirect vs MOVED + +Key → slot uses Redis-compatible CRC16 with `{hash-tag}` support +(`include/eloqkv_key.h:168-221`); slot = `Hash() & 0x3fff` (16384 slots). Slots map onto the +engine's 1024 range buckets as `slot_id = bucket_id + i*1024, i∈[0,16)` +(`GetNodeSlotsInfo`, `src/redis_service.cpp:960-970`), i.e. **bucket = slot mod 1024**; bucket +ownership comes from `LocalCcShards::GetAllBucketOwners` and node-group leadership from +`Sharder` (cf. `data_substrate/docs/06`, `08`). + +Whether a command whose key lives on another node group is forwarded internally or bounced back +is decided in the engine (`data_substrate/tx_service/src/cc/local_cc_handler.cpp:1740`): + +- `txservice_auto_redirect_redis_cmd` — set from the DataSubstrate flag `--auto_redirect` + (default **false**, `data_substrate/core/src/tx_service_init.cpp:52`). When true, every remote + command is transparently shipped via the remote CC handler. +- `always_redirect_` per request — handlers pass `in_tx` for simple commands (so plain + auto-commit commands are *not* force-redirected), while MULTI-queued, Lua, and multi-object + requests pass `true` (a transaction can't be bounced mid-flight without breaking atomicity; + `src/redis_handler.cpp:770-771`, `src/redis_command.cpp:8497`). + +If neither applies, the request fails with `DATA_NOT_ON_LOCAL_NODE` (or +`WRITE_REQUEST_ON_SLAVE_NODE`) and `SendTxRequestAndWaitResult` converts it to a client-visible +`MOVED :` using the first key's slot and the current slot map +(`src/redis_service.cpp:1256-1318`, `GenerateMovedErrorMessage` at 1175). In single-node-group +non-cluster deployments a write on a replica instead returns +`READONLY You can't write against a read only replica.` (1256-1269). There is no ASK/ASKING +migration protocol (no occurrences in the service layer). + +`CLUSTER INFO/NODES/SLOTS/KEYSLOT` are the only supported subcommands (`ParseClusterCommand`, +`src/redis_command.cpp:10823`; anything else gets "ERR unknown subcommand"), all +`DirectCommand`s (`include/redis_command.h:915-987`) +built from `RedisClusterNodes`/`RedisClusterSlots` (`src/redis_service.cpp:993, 1101`), which +fan out `FetchNodeInfo` RPCs to classify each replica `Online`/`Loading`/`Failed` +(`GetReplicaNodesStatus`, 807-905; `HostStatus`, `include/redis_service.h:76`). Node IDs are the +numeric node id left-padded to the 40-char format clients expect (`host_id()`, +`include/redis_service.h:115`). `--cluster_mode` (default false) makes a single node group +present itself with the cluster protocol. + +## 7. Command table organization + +`AddHandlers()` (`src/redis_service.cpp:1372-2019`) instantiates ~175 handler objects (owned by +`hd_vec_`) and registers them by lowercase name in `command_map_` via `AddCommandHandler` +(6347; duplicate registration fails). Lookup is exact-match (`FindCommandHandler`, 6361) — safe +because the brpc parser lowercases `args[0]`. Aliases share one handler (`hmset`→`HSetHandler`, +`unlink`→`DelHandler`, `sort_ro`→`SortHandler`). Unknown names get +``ERR unknown command `x` `` (`src/redis_service.cpp:6049-6057`). + +Handler families, by how their command executes: + +| Family | Examples | Path | +|---|---|---| +| `DirectCommand` — no tx | PING, ECHO, INFO, CLUSTER, CLIENT, CONFIG, TIME, SLOWLOG | `ExecuteCommand(ctx, DirectCommand*, output)` (`src/redis_service.cpp:2648`) | +| Single-key `RedisCommand` | GET/SET/INCR/HSET/ZADD/EXPIRE... | `ObjectCommandTxRequest` (§2) | +| `RedisMultiObjectCommand` | MSET/MGET, SINTERSTORE, LMPOP, BLPOP/BLMOVE... | `MultiObjectCommandTxRequest` with multi-step loop (`src/redis_service.cpp:4519`) | +| `CustomCommand` / bespoke overloads | SORT, SCAN/KEYS, HSCAN/SSCAN/ZSCAN, DUMP/RESTORE, FLUSHDB/FLUSHALL | dedicated `ExecuteCommand` overloads (e.g. 4707 SORT, 4987 SCAN); FLUSH uses `UpsertTableTxRequest` truncates (2757, 2907) | +| Control-flow handlers | MULTI/BEGIN/COMMIT/ROLLBACK/DISCARD, EVAL/EVALSHA/SCRIPT, SUBSCRIBE family | manipulate ctx state / Lua / pub-sub directly | +| Conditional builds | `eloqvec.*` (`VECTOR_INDEX_ENABLED`), `compact` (RocksDB-cloud), `fault_inject` | registered only when compiled in | + +Adding a new command = write the command object + `ParseCommand` (redis_command.h/.cpp), +write a `Handler::Run` (redis_handler.h/.cpp), register it in `AddHandlers()`, add the name +to `command_types` (`src/redis_command.cpp:107`, used for metrics/slowlog classification) and, +if it should work inside MULTI/Lua, to `ParseMultiCommand` and `GenericCommand` — these are +**three separate dispatch tables** that must be kept in sync. Optionally add it to +`cmd_access_types_` for read/write aggregated metrics (`include/redis_service.h:647`). + +## 8. Output handling and error taxonomy + +`OutputHandler` (`include/output_handler.h`) is the visitor every command renders through: +`OnString/OnInt/OnArrayStart/OnArrayEnd/OnNil/OnStatus/OnError/OnBool/OnFormatError`. Two +implementations: + +- `RedisReplier` (`src/redis_replier.cpp`) maps onto `brpc::RedisReply` (RESP2 wire types); + a stack of `(array, next-index)` pairs supports nested arrays (e.g. EXEC results containing + LRANGE arrays). `OnBool` → integer, `OnNil` → null bulk string. **RESP2 only** — stated in + `include/output_handler.h:30` and no RESP3/HELLO support exists. +- `LuaOutputHandler` (`include/lua_output_handler.h`) pushes the same events onto a Lua stack + for `redis.call` results. + +Parse/validation errors use the table in `include/redis_errors.h` (`RD_ERR_*` codes indexing +`redis_error_messages`; e.g. WRONGTYPE, cursor, EXECABORT, key/object-too-big, cluster +shutting down). Engine errors surface as `TxErrorMessage(err_code)` text via +`SendTxRequestAndWaitResult` → `error->OnError(...)` (`src/redis_service.cpp:1320-1327`), plus +the special MOVED/READONLY translations of §6. + +## 9. Stats, INFO, slow log, metrics + +- `RedisStats` (`src/redis_stats.cpp`) exposes brpc bvars when `--enable_redis_stats` (default + on): connections received/rejected/closed, blocked clients, read/write/multi-object command + counters. Incremented in the connection ctor/dtor and in `ExecuteTxRequest`/ + `ExecuteMultiObjTxRequest` (4486-4496, 4528-4536). `INFO` is a `DirectCommand` + (`include/redis_command.h:811`) assembled from these counters plus service fields captured at + Init (OS info, exe path, memory; `src/redis_service.cpp:587-630`). +- Prometheus-style metrics (eloq_metrics): per-command duration histograms and totals plus + read/write aggregates, registered with the engine at Init (289-313) and collected in + `DispatchCommand`/`MultiExec` on sampled rounds (`CheckAndUpdateRedisCmdRound`, 6405). A + dedicated collector thread reports connection count and slow-log length every second + (`CollectConnectionsMetrics`, 6381). +- Slow log is **per task group** (vector indexed by `group_id_`, guarded by per-group + `bthread::Mutex`; `src/redis_service.cpp:201-209`) with `--slow_log_threshold` (µs) and + `--slow_log_max_length`; entries truncate args (32 args / 128 bytes each). `SLOWLOG GET` + merges all groups. `CONFIG SET slowlog-*` adjusts at runtime (4683). + +## 10. Gotchas / verified invariants + +- **Three dispatch tables** (handler map, `ParseMultiCommand`, `GenericCommand`) must agree; a + command registered only in the handler map silently becomes "Unsupported command in MULTI" + and unusable from Lua. +- `FindCommandHandler` relies on the brpc fork lowercasing `args[0]`; don't register + mixed-case names. +- `RedisTableName` consults the thread-local `current_namespace` set by `NamespaceGuard` in + `DispatchCommand` (`src/redis_service.cpp:5863-5870, 5956`) — calling engine paths without the + guard silently targets the wrong table family (docs/05). +- `config_` is guarded by an **atomic spin flag, not a mutex**, because the bthread may resume + on a different task group mid-update and a bthread mutex must unlock on its locking thread + (`include/redis_service.h:605-610`). Same class of constraint as the engine's bthread-mutex + deadlock rules (`data_substrate/CLAUDE.md`, threading gotcha). +- A txm obtained from `NewTxm` is **invalid after `CommitTx`/`AbortTx`** (recycled); EvalLua + nulls its pointer explicitly (`src/redis_service.cpp:2634-2637`). `CommitHandler` likewise + clears `ctx->txm` before inspecting the result. +- Auto-commit failures of `TX_INIT_FAIL` require a manual `AbortTx` because the request never + reached the engine (`src/redis_service.cpp:1235-1255`). +- `GET`-family commands inside BEGIN sessions must be marked `SetVolatile()` so buffered + uncommitted writes on the cc entry are handled correctly (`src/redis_handler.cpp:764-769`). +- The brpc server owns and deletes the `RedisServiceImpl` + (`src/redis_server.cpp:505-506`); `Stop()` only halts collectors/GC — engine shutdown is + `DataSubstrate::Shutdown()` in `main`. +- `--auto_redirect` defaults to **false**: a multi-node-group deployment without it returns + MOVED for simple commands (clients must be cluster-aware) while transactions still forward + internally because they set `always_redirect_`. diff --git a/docs/03-data-model.md b/docs/03-data-model.md new file mode 100644 index 00000000..5277670e --- /dev/null +++ b/docs/03-data-model.md @@ -0,0 +1,131 @@ +# Data Model: Keys, Redis-Type Objects, Commands, and the Catalog Plug-in + +EloqKV plugs into the Data Substrate engine through four artifacts: `EloqKey` (the engine's `TxKey` implementation — a raw byte-string key, CRC16-hashed with Redis hash-tag support), `RedisEloqObject` and its five type subclasses (String/List/Hash/Set/Zset, each with a TTL twin — the `TxObject`s that live in `ObjectCcMap` entries), ~100 `TxCommand` subclasses in `redis_command.h` (mutations execute *on the owner shard* and only small results travel; serialized command images, not values, go to the WAL and to standbys), and `RedisCatalogFactory`/`RedisTableSchema` (registered for `TableEngine::EloqKv`, mapping each Redis database to one prebuilt hash-partitioned table `data_table_`). This doc covers each layer plus the end-to-end TTL design. The generic object/command machinery (ApplyCc, CmdSetEntry, BufferedTxnCmdList) is the engine's: see `data_substrate/docs/05-data-model-and-catalog.md` §3 and `data_substrate/docs/03-concurrency-control.md`. Siblings: [01-architecture-overview.md](01-architecture-overview.md), [02-command-processing.md](02-command-processing.md), [04-scripting-pubsub-blocking.md](04-scripting-pubsub-blocking.md), [05-namespaces.md](05-namespaces.md), [06-vector-search.md](06-vector-search.md), [07-persistence-and-tools.md](07-persistence-and-tools.md). + +## 1. One Redis Database = One Engine Table + +At startup `RedisServiceImpl::Init` builds `databases` (config key `local.databases`, default 16) prebuilt tables named `data_table_0` .. `data_table_`, each a `TableName{name, TableType::Primary, TableEngine::EloqKv}`, plus two system tables: `__ns_0` (namespace registry) and `ns_data_0` (shared data table for all non-default namespaces) — `src/redis_service.cpp:232-281`. All are handed to `DataSubstrate::RegisterEngine(TableEngine::EloqKv, &catalog_factory, ...)` (`src/redis_service.cpp:315`), which seeds the engine catalog so no DDL ever runs for Redis databases. + +- **KV (store) table name**: `"eloqkv_" + table_name` (`include/redis_service.h:574`, must match `txservice::KvTablePrefixOf`), e.g. `eloqkv_data_table_0`. The catalog image passed for each prebuilt table *is* this kv table name (wrapped per store backend, `src/redis_service.cpp:247-257`). +- **SELECT** is a `DirectCommand` that just validates `0 <= db_id < databases` and stores it in the connection context (`src/redis_command.cpp:1736-1753`); it is rejected inside a non-default namespace. Every data command then resolves its table via `RedisServiceImpl::RedisTableName(db_id)`: `redis_table_names_[db_id]` in the default namespace, else always `ns_data_0` (`src/redis_service.cpp:5863-5870`) — namespaces multiplex one table by key prefix (see [05-namespaces.md](05-namespaces.md)). +- **Partitioning**: `TableEngine::EloqKv` tables are hash-partitioned and `IsObjectTable() == true` in the engine (table in `data_substrate/docs/05-data-model-and-catalog.md` §1); the factory's scanner is `HashParitionCcScanner` (`src/eloqkv_catalog_factory.cpp:611-617`). + +## 2. EloqKey (`include/eloqkv_key.h`, `src/eloqkv_key.cpp`) + +`EloqKey` wraps a single `EloqString` (`eloqkv_key.h:448`). It implements the duck-typed `TxKey` contract described in `data_substrate/docs/05-data-model-and-catalog.md` §2; `EloqKey::TxKeyImpl()` (`eloqkv_key.h:440`) instantiates the type-erasure vtable. + +- **Namespace prefixing**: the ordinary constructors (`EloqKey(string_view)`, `EloqKey(buf,len)`, `FromNamespace`) run the key through `CreateEloqStringFromNamespace`, which prepends the thread-local `current_namespace` unless it is `default`/empty (`eloqkv_key.h:49-70`, `include/namespace/context.h:18-50`). `EloqKey::Raw()` (`eloqkv_key.h:72-80`) bypasses prefixing — use it for bytes that are already physical (deserialization, scan cursors). +- **Hashing**: `Hash()` is Redis-cluster-compatible CRC16-XMODEM with hash-tag support — if the key contains `{...}` only the substring inside is hashed (`eloqkv_key.h:170-221`; table in `src/eloqkv_key.cpp:27-66`). The hash is only 16 bits; keys sharing a hash tag co-locate in the same bucket. `HashFromSerializedKey` (`eloqkv_key.h:264`) hashes straight from a serialized buffer (no key materialization; wired to `RedisCatalogFactory::KeyHash`, `src/eloqkv_catalog_factory.cpp:664`). +- **Serialization**: `Serialize` writes a `uint16_t` length prefix + raw bytes (`eloqkv_key.h:223-248`) — used in WAL records and inter-node messages. `KVSerialize`/`KVDeserialize` are the *raw bytes only* (`eloqkv_key.h:275-283`) — the form stored as the KV-store key. Comparison is plain memcmp order on the byte string (`eloqkv_key.h:140-161`). +- **Infinities**: `NegativeInfinity()`/`PositiveInfinity()` are function-local singletons compared **by address** in `operator==`/`<` (`eloqkv_key.h:121-161`, 381-403); `Type()` reports `KeyType` by address too (`eloqkv_key.h:419-433`). `PackedNegativeInfinity()` is a single `0x00` byte (`eloqkv_key.h:405-417`), the serializable stand-in required by range metadata. +- **Limits**: `MAX_KEY_SIZE` = 2 KB for EloqStore builds, 32 MB otherwise (`src/redis_service.cpp:195-198`), enforced per command in `ExecuteCommand` (`src/redis_service.cpp:2665`). +- `MemUsage()` = key byte length (`eloqkv_key.h:321`); `NeedsDefrag` delegates to `EloqString`'s mimalloc page-utilization check (< 0.8 → defrag, `include/eloq_string.h:145-158`). + +## 3. Redis-Type Objects (`include/redis_object.h` + five `redis_*_object.h/.cpp`) + +`RedisEloqObject : txservice::TxObject` is the common base. `RedisObjectType` assigns each concrete class a 1-byte on-disk tag — **the enum values are a persistence format; never reorder** (comment at `redis_object.h:44-45`): `String=0, List=1, Hash=2, Del=3, Zset=4, Set=5, TTLString=6, TTLList=7, TTLHash=8, TTLZset=10, TTLSet=11`. `RedisEloqObject::DeserializeObject` dispatches on that first byte to construct the right subclass (`src/redis_object.cpp:33-77`); `SetEncodedBlob` (store-handler path) just calls `Deserialize` on the blob (`redis_object.h:113-119`). + +| Redis type | Class (file) | Internal representation | TTL twin | Serialized layout (after 1-byte type [+ 8-byte ttl for TTL twin]) | +|---|---|---|---|---| +| string | `RedisStringObject` (`redis_string_object.h:39`) | one `EloqString` | `RedisStringTTLObject` (`:256`) | `uint32` len + bytes | +| list | `RedisListObject` (`redis_list_object.h:38`) | `std::deque` (`:224`) | `RedisListTTLObject` (`:231`) | `uint32` count, then per element `uint32` len + bytes | +| hash | `RedisHashObject` (`redis_hash_object.h:40`) | `absl::flat_hash_map` (`:235`) | `RedisHashTTLObject` (`:241`) | `uint32` count, then per pair `uint32` klen + key + `uint32` vlen + val | +| set | `RedisHashSetObject` (`redis_set_object.h:36`) | `absl::flat_hash_set` (`:99`) | `RedisHashSetTTLObject` (`:105`) | `uint32` count, then per member `uint32` len + bytes | +| zset | `RedisZsetObject` (`redis_zset_object.h:38`) | `std::set` ordered by (score, field) **plus** `absl::flat_hash_map` whose keys *view into* the set's `ZNode` strings (`:341-342`, `:216-234`) | `RedisZsetTTLObject` (`:349`) | `uint32` count, then per member `uint32` len + field + 8-byte `double` score | + +Unlike Redis, there are **no small-object encoding switches** (no listpack/intset/skiplist transitions): each type has exactly one in-memory representation. The only "encoding switch" is TTL/non-TTL, implemented as a *class change*: `AddTTL(ttl)` move-constructs the TTL twin (e.g. `src/redis_string_object.cpp:806`), `RemoveTTL()` moves back (e.g. `redis_string_object.h:392`). TTL twins deliberately report the **base** `ObjectType()` (e.g. `redis_string_object.h:288-293`) so type checks never see TTL-ness; only `Serialize` emits the TTL tag. + +- **Size accounting**: every collection object incrementally maintains `serialized_length_` ("estimated memory size... so the persistent storage flush wouldn't fail", `redis_hash_object.h:236-238`). It is returned by `SerializedLength()`, answers `MEMORY USAGE` (`src/redis_command.cpp:16283-16290`, plus key length at output), and gates writes against `MAX_OBJECT_SIZE` = 256 MB (`src/redis_service.cpp:194`): each mutating `Execute` rejects with `RD_ERR_OBJECT_TOO_BIG` when the post-image would exceed it (e.g. `src/redis_set_object.cpp:71`, `src/redis_list_object.cpp:88`, `src/redis_string_object.cpp:227`). Note `TxRecord::MemUsage()` defaults to 0 and only `RedisStringObject` overrides it (`redis_string_object.h:189-201`, ≤16 bytes counts as inline); shard memory pressure for collections is tracked through the per-shard mimalloc heaps rather than per-object `MemUsage` (inference from the absence of overrides). +- **Where each serialization is used**: `Serialize(std::string&)`/`Serialize(std::vector&,offset)` produce the same format; the engine calls them when flushing dirty objects to the KV store and when shipping whole objects between nodes (e.g. `data_substrate/tx_service/include/cc/template_cc_map.h:2214`, `:10694+`). The WAL normally carries serialized *commands*, not objects (engine doc §3) — full object images enter the log only via `RecoverObjectCommand` (§6) and `RestoreCommand`. `DUMP`/`RESTORE` use a separate Redis-RDB-compatible payload (RDB version 10 + CRC64 footer, `src/redis_command.cpp:15944-15968`); `RESTORE` auto-detects legacy EloqKV-native vs Redis RDB payloads (`RestorePayloadFormat`, `src/redis_command.cpp:16176-16223`). +- `operator==` on hash/list/zset compares pointed-to contents regardless of owned-vs-view `EloqString` storage (`redis_hash_object.h:183-202`). + +## 4. The TxCommand Contract (`include/redis_command.h`, `src/redis_command.cpp`) + +`RedisCommand : txservice::TxCommand` adds one thing to the engine contract: `OutputResult(OutputHandler*)` (`redis_command.h:544-568`) — results render through an abstract handler so the same command serves RESP (brpc replier) and Lua (`include/output_handler.h:33`). Three non-transactional command kinds also exist: `DirectCommand` (no tx service at all: PING, SELECT, CLUSTER, INFO..., `redis_command.h:637`), `CustomCommand` (drives its own multi-request execution: SCAN, ZSCAN-wrapper..., `redis_command.h:677`), and the transactional families `StringCommand`/`ListCommand`/`HashCommand`/`HashSetCommand`/`ZsetCommand` (`redis_command.h:1307/1951/4322/4998/3398`), which fix the result type (`RedisStringResult`, `RedisListResult`, ... all `TxCommandResult` subclasses with `err_code_` + `Serialize/Deserialize` for returning results from a remote owner shard, `redis_command.h:420-542`) and implement `CreateObject(image)` to build/deserialize the right object subclass (`src/redis_command.cpp:1293-1304`). + +Lifecycle of e.g. `SET k v` (cross-ref [02-command-processing.md](02-command-processing.md)): + +1. Parse: `ParseSetCommand` & friends produce `(EloqKey, SetCommand)`; `RedisServiceImpl::ExecuteCommand` wraps them in an `ObjectCommandTxRequest{RedisTableName(db_id), &key, cmd}` (`src/redis_service.cpp:2693-2712`). +2. The tx machine ships the command to the key's owner shard (`ApplyCc`); `ObjectCcMap::Execute` runs `cmd->ExecuteOn(object)` against the (possibly dirty) payload. **`ExecuteOn` never mutates**: object-side `Execute(XCommand&)` methods are `const`, validate types via `CheckTypeMatch`, fill `cmd->result_`, and return either an `ExecResult` (`Fail/Read/Write/Delete/Block/Unlock`) or a `CommandExecuteState` (`NoChange/Modified/ModifiedToEmpty`, `redis_object.h:37-42`) that the command translates. +3. On commit, `cmd->CommitOn(obj)` applies the mutation via the object's `Commit*` helpers (`CommitHset`, `CommitSAdd`, ...) and returns the resulting object pointer — possibly a *different* object (TTL twin added/removed) or `nullptr` for "now deleted" (`DelCommand::CommitOn`, and any `ModifiedToEmpty` path such as `SAddCommand::CommitOn` returning nullptr when the set empties). +4. `OutputResult` renders `result_` to the client. + +Key predicates as EloqKV implements them: + +| Predicate | Meaning here | Examples | +|---|---|---| +| `IsReadOnly()` | No commit step; command needs no `Clone()` (clone returns nullptr by default, `redis_command.h:546`) | GET, LRANGE, ZSCORE, TTL | +| `IsOverwrite()` | Post-image independent of pre-image → engine prunes all earlier buffered/logged commands for the object | SET (`redis_command.h:1379`), DEL (`:3183`), StoreListCommand (`:6534`), RESTORE-with-REPLACE (`:7346`), RecoverObjectCommand (`:7497`) | +| `IgnoreOldValue()` | Skip fetching the old value from the KV store entirely (engine pretends a prior delete, `object_cc_map.h:601-612`) | plain SET without NX/XX/GET/KEEPTTL (`redis_command.h:1384-1392`) | +| `IsDelete()` / `IsLazyDelete()` | Tombstone; UNLINK sets lazy (`redis_command.h:3188-3196`; `DelCommand(true)` for UNLINK, `src/eloqkv_catalog_factory.cpp:197-199`) | DEL/UNLINK | +| `ProceedOnNonExistentObject()` / `ProceedOnExistentObject()` | NX/XX-style existence gating evaluated before execution | SET flags (`redis_command.h:1361-1377`), RESTORE replace (`:7336-7344`) | +| `WillSetTTL()` | Command may change TTL without overwriting — triggers the recover-object WAL protocol (§6) | SET EX (`:1394`), EXPIRE (`:7712`), GETEX (`:7593`), PERSIST (`:7843`) | +| `IsVolatile()` | Command object dies after execution (MULTI/Lua reuse); must be `Clone()`d to survive to commit (`redis_command.h:551-567`) | set via `SetVolatile()` e.g. on `StoreListCommand` (`:6489`) | + +**Serialization for WAL/standby/migration**: every mutating command's `Serialize` writes a 1-byte `RedisCommandType` followed by its arguments (e.g. `SAddCommand::Serialize`, `src/redis_command.cpp`); the engine accumulates these images per object and replays them through `RedisTableSchema::CreateTxCommand(cmd_image)` (§7). Results never enter the log. + +**Expired-object hook**: `RedisCommand::RetireExpiredTTLObjectCommand()` returns a fresh `DelCommand` (`src/redis_command.cpp:1288-1291`); the engine injects it ahead of the real command when it observes an expired TTL during ApplyCc (`object_cc_map.h:928`, `:1042`, `:1094`), so the expired object is deleted-then-recreated transactionally. + +## 5. Multi-Key Commands (`RedisMultiObjectCommand`) + +`RedisMultiObjectCommand : txservice::MultiObjectTxCommand` (`redis_command.h:570-631`) holds parallel vectors `vct_key_ptrs_[step]` / `vct_cmd_ptrs_[step]`; the engine executes one step at a time (all keys in a step in parallel, then `HandleMiddleResult()` decides whether to continue, `IncrSteps()`, repeat). Dispatch goes through `MultiObjectCommandTxRequest` (`src/redis_service.cpp:2723-2727`). **There is no CROSSSLOT restriction** — no such check exists in the codebase (verified by grep); keys may hash to any bucket/node, and the engine runs the whole thing as one transaction (cross-ref [02-command-processing.md](02-command-processing.md)). + +| Shape | Commands | Mechanics | +|---|---|---| +| Single step, N keys | MSET (`redis_command.h:5475`), MGET (`:5507`, repeated keys deduped via `raw_cmds_`; WRONGTYPE → nil), multi-key DEL/EXISTS (`MDelCommand`/`MExistsCommand`), SDIFF/SINTER/SUNION, ZDIFF/ZUNION/ZINTER, MWATCH, MHGET | one step; results aggregated in `OutputResult` | +| Two steps, move | SMOVE = SRem then SAdd (`:5926`), LMOVE/RPOPLPUSH = LMovePop then LMovePush (`:5968-6018`) | `HandleMiddleResult` aborts step 2 if the source op failed; `IsPassed` checks both legs | +| Two steps, read-then-store | ZUNIONSTORE/ZINTERSTORE (`:6158`: step 0 = `SZScanCommand` reads on N keys, step 1 = `ZAddCommand` on the destination), ZRANGESTORE, SDIFFSTORE/SINTERSTORE/SUNIONSTORE (destination via overwrite `StoreListCommand`/`SAddCommand` with `force_remove_add_`), MSETNX (exists-check then set), BITOP | middle handler computes the aggregate and loads it into the store-step command; store commands carry `is_in_the_middle_stage_` so commits **clone instead of move** their strings (a step's command may commit more than once, e.g. `SAddCommand::CommitOn` passing it to `CommitSAdd`) | +| Blocking | BLMOVE (`:2848`, 4 steps: try-pop → block → re-pop → push), BLPOP/BRPOP/BLMPOP/LMPOP (`BLMPopCommand` `:2906`: one try-step per key, then a blocking step, then the pop), ZMPOP (`ZMPopCommand` `:4158`) | `ExecuteOn` returns `ExecResult::Block`; the request parks on the cc entry (engine doc §3). `IsExpired()` checks the deadline only in the blocking step (`:2877-2885`); `ForwardResult()` swaps in a `BlockDiscardCommand` when the wakeup yielded nothing (`src/redis_command.cpp:5280-5299`); `IsBlockCommand()` returns true. See [04-scripting-pubsub-blocking.md](04-scripting-pubsub-blocking.md). | + +## 6. TTL End-to-End + +- **Representation**: absolute expiration in **epoch milliseconds** stored in the TTL twin's `ttl_` (`UINT64_MAX` = never). The comment "micro seconds" at `redis_command.h:7719` is stale — `ExpireCommand` compares against `ClockTsInMillseconds()` (`src/redis_command.cpp:7821`) and parsing computes `now_ms + seconds*1000` (`ParseExpireCommand`, `src/redis_command.cpp`). +- **Setting**: `EXPIRE/PEXPIRE/EXPIREAT/PEXPIREAT` → `ExpireCommand` with NX/XX/GT/LT flags (`redis_command.h:7638-7726`); `SET ... EX/PX/EXAT/KEEPTTL` and `SETEX/PSETEX/GETEX` carry `obj_expire_ts_`/`expire_ts_`; `PERSIST` → `PersistCommand`. `CommitOn` either mutates `ttl_` in place (already a TTL object) or swaps the object for its TTL twin via `AddTTL` / back via `RemoveTTL` (`src/redis_command.cpp:7933+`, `:7747-7777`). +- **Reading**: `TTL/PTTL/EXPIRETIME/PEXPIRETIME` share `TTLCommand` (read-only, `redis_command.h:7728`). +- **Lazy enforcement (engine side)**: during ApplyCc the `ObjectCcMap` reads the payload's TTL; if expired, read-only commands return `RecordStatus::Deleted` immediately, and mutating commands set `ttl_expired_` so the engine first applies `RetireExpiredTTLObjectCommand()` (= `DelCommand`) and then runs the new command on a fresh object (`data_substrate/tx_service/include/cc/object_cc_map.h:657-673`, `:928`). Reads from the KV store likewise treat expired records as deleted, and physical reclamation is compaction-driven (`TTLCompactionFilter`) — `data_substrate/docs/09-store-handler.md` §"Record TTL". In-memory pages track `smallest_ttl_` so page cleaning can drop expired entries wholesale (`cc/cc_page_clean_guard.h:78-123`). +- **WAL correctness for TTL-only writes**: EXPIRE/PERSIST/GETEX mutate only the TTL, are *not* overwrites, yet replay must not depend on fetching the old value from the KV store. So when they will actually reset a TTL, `ExecuteOn` serializes the **whole current object** into an attached `RecoverObjectCommand` (`recover_ttl_obj_cmd_`, `src/redis_command.cpp:7923-7927`, `:7741-7742`); the engine logs that command's image instead (`RecoverTTLObjectCommand()` hook, `object_cc_map.h:1085`). `RecoverObjectCommand` is itself an overwrite whose `CommitOn` rebuilds the TTL object from the embedded blob (`src/redis_command.cpp:7601-7677`). +- The checkpointer hands each record's TTL to the store handler (`BatchWriteRecords` items carry a ttl), keyed off `HasTTL()/GetTTL()` — `object_cc_map.h:640-652`, engine doc 09. + +## 7. The Catalog Factory (`include/eloqkv_catalog_factory.h`, `src/eloqkv_catalog_factory.cpp`) + +`RedisTableSchema` is deliberately minimal: `RedisKeySchema` only carries `schema_ts_` (== table version; `CompareKeys` is a stub, `eloqkv_catalog_factory.h:40-70`), `RedisRecordSchema` is empty, the schema image *is* the kv table name, and `kv_info_` is a per-backend `KVCatalogInfo` (`DynamoCatalogInfo` / `RocksDBCatalogInfo` / plain `txservice::KVCatalogInfo` for DSS builds) holding `kv_table_name_` (`src/eloqkv_catalog_factory.cpp:57-99`). No indexes, no statistics, no auto-increment (`GetIndexes` asserts; `BindStatistics` no-op; `StatisticsObject` nullptr). `CreateSkEncoder` is not overridden — the base returns nullptr (`data_substrate/tx_service/include/catalog_factory.h:380`): EloqKV has no secondary keys. + +`RedisTableSchema::CreateTxCommand(cmd_image)` (`src/eloqkv_catalog_factory.cpp:107-560`) is the replay decoder: first byte → `RedisCommandType` → `make_unique()` → `Deserialize(rest)`. It is used by WAL recovery, standby command application, and bucket migration (engine doc §3). Note the switch also covers read-only commands (GET, HSCAN, ZRANGE...) because command images are also shipped for remote/forwarded execution, and several internal types appear: `RECOVER` (§6), `STORE_LIST`/`SORTABLE_LOAD` (SORT/...STORE plumbing), `BLOCKPOP`/`BLOCKDISCARD`, `SZSCAN` (zset/set union-store scans). + +`RedisCatalogFactory` overrides (`src/eloqkv_catalog_factory.cpp:562-691`): + +| Method | Implementation | +|---|---| +| `CreateTableSchema` | `RedisTableSchema` | +| `CreatePkCcMap` | `ObjectCcMap` (`:571-588`) — the object/command cc map | +| `CreateSkCcMap` / `CreateSkCcmScanner` / `CreateRangeCcmScanner` | `nullptr` (no secondary indexes; no range scans on hash tables) | +| `CreateRangeMap` | `RangeCcMap` (`:599-609`; required plumbing even for hash-partitioned engines) | +| `CreatePkCcmScanner` | `HashParitionCcScanner` (SCAN/KEYS path) | +| `CreateTableStatistics` (both) | `nullptr` — no stats for Redis tables | +| `NegativeInfKey`/`PositiveInfKey`/`PackedNegativeInfinity`/`CreateTxKey`/`KeyHash` | delegate to `EloqKey` singletons/ctors/`HashFromSerializedKey` | +| `CreateTxRecord` | `RedisEloqObject::Create()` — a base-class instance whose `DeserializeObject` re-types on first byte | + +## 8. Supporting Utilities (brief) + +| File | Purpose | +|---|---| +| `include/eloq_string.h` | 24-byte union string: Stack (≤23 bytes inline), Heap (owned, 16-byte-aligned alloc), View (non-owning) — discriminated by 2 flag bits. Copy/move of a View stays a View; call `Clone()` to own. Foundation of all object payloads; the reason commands distinguish move vs clone on commit. | +| `include/redis_string_num.h` | Redis-compatible numeric/string conversions: `string2ll/string2ull/string2double/string2ld`, `d2string/ld2string`, overflow predicates — used by INCR*/ZADD/HINCRBY parsing and execution. | +| `include/redis_string_match.h` | `stringmatchlen` — Redis glob matching (`*?[]\`) for KEYS/SCAN/HSCAN MATCH; `EloqKey::IsMatch` wraps it (`eloqkv_key.h:351-379`). | +| `include/eloq_algorithm.h` | `GenRandMap` — random index selection with/without repetition for SRANDMEMBER/HRANDFIELD/ZRANDMEMBER. | +| `include/b255.h` | Base-255 encoding of namespace IDs that never emits the `:` delimiter (key-prefix safety, see [05-namespaces.md](05-namespaces.md)). | +| `include/base64url.h` | Header-only base64url decode/encode used for tokens (namespace auth). | + +## 9. Gotchas and Verified Invariants + +- **`RedisObjectType` values are on-disk format** — `Del=3` and `TTLZset=10` leave holes (no `9`); never renumber (`redis_object.h:44-60`). +- **ExecuteOn is read-only; CommitOn mutates.** Object `Execute(...)` methods are `const` and must compute size-limit checks against the *would-be* post-image (`serialized_length_ + delta > MAX_OBJECT_SIZE` → `RD_ERR_OBJECT_TOO_BIG`, e.g. `src/redis_set_object.cpp:71`). If a command's `ExecuteOn` succeeded, `CommitOn` may not fail. +- **`serialized_length_` is maintained incrementally** and re-verified by asserts during `Serialize` (`redis_list_object.h:92`, `redis_zset_object.h:97`); `Deserialize` recomputes it. TTL twins exclude the 8 ttl bytes from `serialized_length_` and add them in `SerializedLength()` (`redis_hash_object.h:280-284` vs `:401`). +- **Zset dual structure**: `z_hash_map_` keys are `string_view`s into `ZNode.field_` strings owned by `z_ordered_set_` (`redis_zset_object.h:334-342`); any code path that erases from the set must erase from the map first (and vice versa), and copy construction must rebuild the views. +- **CommitOn returning `nullptr` means deletion** — empty collections are never stored (`CommandExecuteState::ModifiedToEmpty`; `DelCommand::CommitOn`, `src/redis_command.cpp`). Redis semantics "empty key = absent key" fall out of this. +- **Namespace prefix is implicit in `EloqKey` construction** — constructing `EloqKey(sv)` anywhere picks up the bthread-local `current_namespace` (`namespace/context.h:46`). Internal code handling already-prefixed bytes must use `EloqKey::Raw`. +- **TTL twins lie about their type on purpose** (`ObjectType()` returns the base type) so `IsMatchType`/`CheckTypeMatch` and all command code treat them as plain objects; only serialization and `HasTTL()` reveal TTL-ness. +- **`is_in_the_middle_stage_` / `should_not_move_string`**: store-stage commands of multi-step commands clone strings on commit instead of moving them, because the same command instance can commit again (e.g. `RedisHashSetObject::CommitSAdd(paras, should_not_move_string)`, `redis_set_object.h:72`). +- **MGET maps WRONGTYPE to nil** per Redis semantics (`redis_command.h:5536-5553`); `EXISTS` with repeated keys multiplies by occurrence count (`cnt_`, `redis_command.h:3289`). +- **DUMP emits Redis-RDB-version-10 payloads with CRC64** (`redis_command.h:7286-7287`, `src/redis_command.cpp:15944`); RESTORE additionally accepts the older EloqKV-native dump format (version tag 0) for backward compatibility. +- **Defrag**: `NeedsDefrag` walks object/string allocations and reports pages under 80% utilization (`src/redis_string_object.cpp:789`, `eloq_string.h:145`); the shard's defrag pass re-allocates such entries (engine-side mechanism). diff --git a/docs/04-scripting-pubsub-blocking.md b/docs/04-scripting-pubsub-blocking.md new file mode 100644 index 00000000..7c8dbe6b --- /dev/null +++ b/docs/04-scripting-pubsub-blocking.md @@ -0,0 +1,132 @@ +# 04 — Lua Scripting, Pub/Sub, and Blocking Commands + +EloqKV layers three "stateful" Redis features on top of the Data Substrate engine. **Lua scripting** (EVAL/EVALSHA/SCRIPT) runs each script inside a single engine transaction: a pooled Lua 5.1 interpreter executes the script body, and every `redis.call()` re-enters the normal command-parsing path (`RedisServiceImpl::GenericCommand`) bound to that one `TransactionExecution`, so the transaction-scoped operations commit or abort atomically (default RepeatableRead + OCC, with automatic whole-script retry on OCC conflicts); direct commands such as `PUBLISH` run outside that transaction. **Pub/Sub** is a node-local, in-memory subscriber registry (`PubSubManager`) plus a best-effort cross-node fan-out: PUBLISH sends a `PublishTxRequest` that streams a `PublishRequest` CcMessage to every other node-group leader, where the engine calls back into EloqKV's registered `publish_func`. **Blocking commands** (BLPOP/BRPOP/BLMOVE/BLMPOP/BRPOPLPUSH) never block a thread: the engine parks the command's `ApplyCc` request on the key's lock object (`queue_block_cmds_`), a committing writer pops and re-executes it, and the transaction machine is re-enlisted every 10 ms to check the timeout — the client's bthread meanwhile just waits on the `TxRequest` result. Engine internals referenced here are documented in `data_substrate/docs/02-threading-model.md`, `03-concurrency-control.md`, and `04-transaction-execution.md`. + +Siblings: `01-architecture-overview.md`, `02-command-processing.md`, `03-data-model.md`, `05-namespaces.md`, `06-vector-search.md`, `07-persistence-and-tools.md`. + +--- + +## 1. Lua scripting + +Files: `include/lua_interpreter.h`, `src/lua_interpreter.cpp`, `include/lua_output_handler.h`, EVAL/EVALSHA/SCRIPT plumbing in `src/redis_service.cpp` and `src/redis_handler.cpp`. The Lua VM itself is the vendored Redis fork under `lua/` (with `lua_enablereadonlytable`, cjson, struct, cmsgpack, bit). + +### Interpreter instances and sandbox + +- `LuaInterpreter` wraps one `lua_State` (`include/lua_interpreter.h:45`). Instances are pooled in a global `moodycamel::ConcurrentQueue>` (`include/redis_service.h:597`); `GetLuaInterpreter()` dequeues or constructs one, `CleanAndReturnLuaInterpreter()` clears the stack and returns it (`src/redis_service.cpp:1354-1370`). The pool is global, not per-core; any brpc worker bthread may use any interpreter. +- `InitLua` (`src/lua_interpreter.cpp:188-294`) loads base/table/string/math/debug plus cjson, struct, cmsgpack, bit (each made read-only); installs `__redis__err__handler` (adds `source:line` to errors); installs the global-protection metatable (scripts cannot create or read undeclared globals); then nils out `print`, `loadfile`, `dofile` and sets `debug = nil`. There is no `os`/`io` library, so no wall-clock or filesystem access from scripts. +- The `redis` table exposes only `call`, `pcall`, `error_reply`, `status_reply`, `sha1hex` (`src/lua_interpreter.cpp:778-819`). Redis features like `redis.log`, `redis.setresp`, `redis.replicate_commands`, `redis.breakpoint` do not exist. +- Each interpreter owns a private `RedisConnectionContext` (`lua_conn_ctx_`, `include/lua_interpreter.h:92-98`); `SetConnectionContext` copies only `db_id` from the caller's connection, so `SELECT` inside a script cannot leak to the client connection (`src/lua_interpreter.cpp:391-394`). + +### Script cache and SHA + +- `CreateFunction` SHA1-hashes the body and defines a global Lua function `f_` if not already present, so each interpreter instance also memoizes compiled bodies (`src/lua_interpreter.cpp:344-373`). +- The service-level cache is `std::unordered_map scripts_` (sha → body) guarded by `std::shared_mutex script_mutex_` (`include/redis_service.h:600-601`). `SCRIPT LOAD` compiles + inserts (`src/redis_service.cpp:2419-2447`); EVAL also inserts (`src/redis_service.cpp:2588-2593`); `SCRIPT EXISTS`/`SCRIPT FLUSH` read/clear it (`src/redis_service.cpp:2393-2417`). The cache is **node-local and not replicated** — EVALSHA on a node that never saw the body returns `NOSCRIPT` (`src/redis_service.cpp:2468-2473`). EVALSHA resolves sha → body and falls into `EvalLua` (`src/redis_service.cpp:2449-2481`). + +### Execution flow (`EvalLua`, `src/redis_service.cpp:2483-2646`) + +1. `EvalHandler`/`EvalshaHandler` reject EVAL inside `BEGIN` or `MULTI` (`src/redis_handler.cpp:1964-1973`, `2049-2058`), then call `EvalLua` on the connection's bthread. +2. Parse `numkeys`, KEYS, ARGV; enter a `while (true)` retry loop (`src/redis_service.cpp:2549`). +3. Per attempt: `NewTxm(txn_isolation_level_, txn_protocol_)` — the *transaction-class* settings, default **RepeatableRead + OCC** via `--txn_isolation_level`/`--txn_protocol` (`src/redis_service.cpp:135-141`, `440-485`), distinct from the simple-command defaults (ReadCommitted/OccRead). +4. Acquire an interpreter, set `KEYS`/`ARGV` globals, and install the redis hook: a lambda capturing the txm that forwards to `GenericCommand(ctx, txm, args, reply)` (`src/redis_service.cpp:2565-2572`). +5. `CallFunction` runs `f_` under `lua_pcall` with the error handler, with `_G` temporarily read-only (`src/lua_interpreter.cpp:396-430`). +6. On Lua error: `AbortTx(txm)`; if the message indicates an OCC conflict (`OCC break repeatable read` / `write-write conflicts`) and `--retry_on_occ_error` (default true, `src/redis_service.cpp:143`), the **whole script re-runs** in a fresh txm (`src/redis_service.cpp:2596-2614`). Otherwise the error is returned. +7. On success: `CommitTx(txm)`; commit-time OCC conflicts also re-run the script (`src/redis_service.cpp:2616-2638`). Finally `LuaReplyToRedisReply` converts the script's return value. + +So a script is exactly one engine transaction: every `redis.call` issues `ObjectCommandTxRequest`s with `auto_commit=false` against the shared txm (e.g. `src/redis_service.cpp:2657-2714`), and nothing is visible to other transactions until the final commit. + +### The redis.call bridge and value conversion + +- `redis.call`/`redis.pcall` land in `RedisGenericCommand` (`src/lua_interpreter.cpp:599-709`): arguments are popped from the Lua stack (numbers formatted with `%.17g`), the command name lower-cased, and the hook invoked with a `LuaOutputHandler`. `call` raises a Lua error on command failure (aborting the script); `pcall` returns the `{err=...}` table to the script. Recursive invocation via debug hooks is detected and refused (`src/lua_interpreter.cpp:609-621`). +- Command result → Lua (`LuaOutputHandler`, `src/lua_interpreter.cpp:67-173`): string→string, int→integer, nil→`false`, status→`{ok=...}`, error→`{err=...}`, arrays→tables (nested via `array_index_` stack). +- Lua return → RESP (`LuaReplyToRedisReply`, `src/lua_interpreter.cpp:435-545`): string→bulk, number→**integer (truncated)**, `false`→nil, `true`→`:1`, table→array up to the first nil (with `{err=}`/`{ok=}` recognized), other types→`ERR Unsupported return type`. These match Redis semantics. + +### Limits / unsupported + +- Commands not in `GenericCommand`'s switch fail with `Unknown Redis command called from script` (`src/redis_service.cpp:4362-4364`). Notably absent: SUBSCRIBE/PSUBSCRIBE, blocking commands (BLPOP etc.), MULTI/EXEC, EVAL (no nesting). PUBLISH **is** supported inside scripts (`src/redis_service.cpp:3112-3120`). +- No script kill / max-execution-time mechanism was found (no `SCRIPT KILL`, no `lua_sethook` budget); a long-running script occupies its bthread until it finishes. +- Determinism is not required for replication correctness: scripts execute once on the receiving node and only their *effects* (individual `TxCommand`s) are WAL-logged / forwarded to standbys; the script text never replicates. (Inference from the architecture; the per-command logging path is the same as for ordinary writes.) +- Keys are not validated for slot-locality: each `redis.call` runs with `always_redirect=true` (`include/redis_service.h:317-323`), so a script may touch keys on remote node groups; the engine fetches/locks them remotely rather than returning CROSSSLOT errors. + +--- + +## 2. Pub/Sub + +Files: `include/pub_sub_manager.h`, `src/pub_sub_manager.cpp`, glue in `src/redis_service.cpp`, engine path in `data_substrate/tx_service`. + +### Data structures + +A single process-global `PubSubManager eloqkv_pub_sub_mgr` (`src/redis_service.cpp:184`) holds, under one `bthread::Mutex pub_sub_mu_`: + +- `pub_sub_channels_`: `flat_hash_map>` — exact channels (`include/pub_sub_manager.h:69-72`). +- `pattern_subs_`: same shape for glob patterns (`include/pub_sub_manager.h:75-77`). + +Each connection mirrors its own membership in `subscribed_channels` / `subscribed_patterns` sets (`include/redis_connection_context.h:158-159`) so `SUBSCRIBE` replies can report the count and teardown can be quick. Channel/pattern entries are erased when their subscriber set empties. + +### Subscribe / unsubscribe + +`SubscribeHandler` etc. (`src/redis_handler.cpp:4818+`) call straight into the manager (`src/redis_service.cpp:6426-6449`). Confirmations (`subscribe`/`unsubscribe`/`psubscribe`/`punsubscribe`, channel, count) are 3-element RESP2 arrays built in the connection's reusable `output` reply and written **directly to the socket** by `RedisConnectionContext::FlushOutput()` → `socket->Write` (`src/pub_sub_manager.cpp:39-45`, `src/redis_connection_context.cpp:90-105`), bypassing the normal brpc reply pipeline. There is no RESP3 push-frame support; subscribers receive classic RESP2 arrays. + +### PUBLISH flow + +`PublishCommand` is a `DirectCommand` — it does not touch any key/cc-map (`include/redis_command.h:1292-1305`, `src/redis_command.cpp:3668-3677`). `RedisServiceImpl::Publish` (`src/redis_service.cpp:6451-6462`): + +1. **Cross-node**: create a throwaway txm, send `PublishTxRequest{chan, msg}` (`data_substrate/tx_service/include/tx_request.h:1166-1179`). `TransactionExecution::ProcessTxRequest(PublishTxRequest&)` iterates **all node groups** and calls `cc_handler_->PublishMessage(ng_id, ...)`, then finishes the request immediately — fire-and-forget (`data_substrate/tx_service/src/tx_execution.cpp:1074-1083`). +2. `LocalCcHandler::PublishMessage` skips the local node and sends a `CcMessage::PublishRequest` over the cc stream to each **node-group leader** (`data_substrate/tx_service/src/cc/local_cc_handler.cpp:1819-1830`, `remote/remote_cc_handler.cpp:1096-1112`). +3. On the receiving node, `CcStreamReceiver` dispatches the message to `LocalCcShards::PublishMessage` (`remote/cc_stream_receiver.cpp:1896-1905`), which spawns a background bthread that invokes the engine-registered `publish_func_` (`src/cc/local_cc_shards.cpp:1250-1268`, `include/cc/local_cc_shards.h:1838-1846`). +4. `publish_func` is EloqKV's `eloqkv_publish_func = [](chan, msg){ eloqkv_pub_sub_mgr.Publish(chan, msg); }`, wired in at startup via `DataSubstrate::RegisterEngine(..., eloqkv_publish_func)` (`src/redis_service.cpp:185-187`, `315-320`; plumbed through `core/src/data_substrate.cpp:410` → `core/src/tx_service_init.cpp:319` → `LocalCcShards`). +5. **Local**: the publisher then runs `eloqkv_pub_sub_mgr.Publish` itself (`src/pub_sub_manager.cpp:395-449`): exact-channel matches get `["message", chan, msg]`; every pattern is tested with `stringmatchlen` (Redis glob, `src/redis_string_match.cpp`) and matches get `["pmessage", pattern, chan, msg]`. Each delivery is serialized and written to the subscriber's socket inline, under `pub_sub_mu_`, so a slow subscriber stalls unrelated Pub/Sub traffic. + +The integer PUBLISH returns counts **local subscribers only** (comment at `src/redis_service.cpp:6459-6461`). + +### Lifecycle and guarantees + +- `~RedisConnectionContext` calls `UnsubscribeAll(this)` if any subscriptions remain (`src/redis_connection_context.cpp:37-53`), removing the raw pointer from both maps. The destructor carries a `Fixme(zkl): risk of data race for subscribed_channels` comment — teardown vs. concurrent publish is a known thin spot. +- Delivery is **at-most-once, best-effort**: no persistence, no acks (the `PublishTxRequest` completes before remote delivery), per-message background bthreads, and remote fan-out targets only node-group leaders. A failed `socket->Write` just logs a warning (`src/redis_connection_context.cpp:99-103`). Subscribers connected to a node that is not currently a node-group leader will not see remote publishes (mechanism-implied; verified that sends go to `Sharder::Instance().LeaderNodeId(ng)` only). +- Namespaces/db ids are **not** part of the channel key: `Publish(chan, msg)` carries the raw channel name, so pub/sub is global across SELECT databases and namespaces (no `db_id`/`ns_id` qualification anywhere in `PubSubManager`). + +--- + +## 3. Blocking commands (BLPOP / BRPOP / BLMOVE / BLMPOP / BRPOPLPUSH) + +Files: handlers `src/redis_handler.cpp:4008-4146`, command objects `include/redis_command.h:2746-2976` + `src/redis_command.cpp:5067-5575`, engine machinery in `data_substrate/tx_service` (see `data_substrate/docs/03-concurrency-control.md` §5 and `04-transaction-execution.md`, which notes the `tx_progress_block_` / 10 ms re-enlist design). + +### Command model + +All five commands compile to a `RedisMultiObjectCommand` (`BLMPopCommand` or `BLMoveCommand`) — a multi-*step* state machine where each step is a vector of (key, `TxCommand`) pairs sent as one `MultiObjectCommandTxRequest`. The shared child command is `BlockLPopCommand`, whose behavior is selected by `txservice::BlockOperation` (`data_substrate/tx_service/include/tx_command.h:61-71`): `PopNoBlock` (try-pop), `BlockLock` (wait until non-empty, then hold the lock), `PopElement` (actual pop after a successful BlockLock), `Discard` (cancel a parked request), `NoBlock` (non-blocking variant for MULTI). Its `ExecuteOn` maps these to `ExecResult::{Read,Write,Delete,Block,Unlock}` (`src/redis_command.cpp:5067-5122`). Timeouts are absolute microsecond deadlines: `ts_expired_ = ClockTs() + timeout`; timeout 0 becomes one year (`src/redis_command.cpp:19847-19850`); `IsExpired()` compares against `LocalCcShards::ClockTs()` and only at the blocking step (`include/redis_command.h:2877-2885`). + +### End-to-end flow (BLPOP on an empty list) + +1. **Dispatch.** `BLPopHandler::Run` parses, creates/joins a txm, and calls the multi-object `ExecuteCommand` (`src/redis_handler.cpp:4064-4090`) → `ExecuteMultiObjTxRequest` (`src/redis_service.cpp:4519-4620`), which counts the client in `blocked_clients` stats and drives the step loop: `SendTxRequestAndWaitResult` → `HandleMiddleResult()` → `IncrSteps()` → repeat. +2. **Client wait.** The connection bthread parks in `tx_req->Wait()` on a `bthread::Mutex`/`ConditionVariable` (or yield/resume under `cc_notify`) (`data_substrate/tx_service/include/tx_request.h:112-150`, `tx_req_result.h:75-76`). A bthread is a coroutine; the brpc worker thread stays free. +3. **Step 0..n-1 (try-pop).** Each key is probed with `PopNoBlock`; a non-empty list returns elements immediately (`ExecResult::Write/Delete`) and the command skips to completion. An empty list returns `ExecResult::Unlock` — the shard releases the lock and finishes the request with a "nil" result (`data_substrate/tx_service/include/cc/object_cc_map.h:1168-1186`). +4. **Enter blocking step.** `HandleMiddleResult` flips all child commands to `BlockLock` (`src/redis_command.cpp:5541-5552`) and the next step sends them again to every key's owner shard. +5. **Park on the cc entry.** `ApplyCc` executes `BlockLock` on the object; if still empty, `ExecuteOn` returns `ExecResult::Block` and the shard pushes the *request itself* onto the key lock's blocked-command queue: `cce->PushBlockCmdRequest(&req)` → `NonBlockingLock::queue_block_cmds_`, then releases the just-acquired lock (`object_cc_map.h:1149-1166`, `cc_entry.h:360-366`, `non_blocking_lock.h:261-264`). The shard thread moves on — nothing blocks. +6. **Wake on mutation.** When any transaction that write-locked the key commits/clears (e.g. an LPUSH), `ReleaseWriteLock`/`ClearTx` first try `PopBlockCmdRequest(ccs, object)`: it scans the queue, asks each parked command `AblePopBlockRequest(object)` (list non-empty? `src/redis_command.cpp:5161-5173`), and for the first match upgrades that tx to the WriteLock and re-enqueues the parked `ApplyCc` onto the shard queue (`data_substrate/tx_service/src/cc/non_blocking_lock.cpp:385-391`, `570-580`, `675-705`). Re-execution now sees elements → `ExecResult::Read` → the handler-result completes, decrementing `atm_block_cnt_` in `MultiObjectCommandOp` (`tx_operation.cpp:5660-5677`). +7. **Re-enlist & forward.** Blocking txms are tracked in the TxProcessor's `tx_progress_block_` map (populated by `EnlistWaitingTx` when the top operation `IsBlockCommand()`, `tx_service.h:816-841`, via `StartTiming`, `tx_execution.cpp:700-711`). The processor loop's `CheckWaitingTxs()` re-forwards these txms every **10 ms** (`check_progress_block_period = 10000` µs vs 2 s for ordinary txs, `tx_service.h:719-749`). `MultiObjectCommandOp::Forward` at the blocking step returns early while `!IsExpired() && atm_block_cnt_ > 0` (`tx_operation.cpp:5785-5794`); once unblocked it calls `ForwardResult()` (picks the winning key, marks the rest for discard) and the service loop advances to the `PopElement` step — the real pop, this time written to the WAL — then (for BLMOVE) the push to the destination key, then auto-commit (`src/redis_service.cpp:4599-4613`). +8. **Timeout.** When the 10 ms re-enlist finds `IsExpired()`, `ForwardResult()` substitutes `BlockDiscardCommand` for every still-parked child; Forward sends these `Discard` commands to the shards (`tx_operation.cpp:5796-5864`). On the shard, `Discard` aborts the parked request out of `queue_block_cmds_`/`blocking_queue_` with `TASK_EXPIRED` (`object_cc_map.h:405-433`, `non_blocking_lock.cpp:707-735`); `TASK_EXPIRED` is deliberately not treated as an error in the post-lambdas (`tx_operation.cpp:5667-5672`). The command then reports nil to the client (`src/redis_command.cpp:5497-5500`). If a discard races and finds nothing, the txn is remembered in the shard's `active_blocking_txs_` so a late wake can still be cleaned up (`object_cc_map.h:420-422`, `cc_shard.h:1455-1459`). + +### Why nothing blocks + +Three parking lots, all non-blocking: (a) the *client bthread* parks on the TxRequest result (coroutine yield); (b) the *txm* is a heap state machine that simply isn't forwarded while waiting (re-polled at 10 ms); (c) the *cc request* is parked on the key's `NonBlockingLock`. Shard threads (TxProcessors) never wait on anything — consistent with the threading rules in `data_substrate/docs/02-threading-model.md`. + +### Interaction with MULTI / Lua / failover + +- Inside MULTI, the same commands are parsed with `multi=true` → `BlockOperation::NoBlock` and `ts_expired_ = 0`, i.e. they degrade to non-blocking try-pops (`src/redis_command.cpp:19719-19724`, `19877-19880`; dispatched from `ParseMultiCommand`, `src/redis_command.cpp:10073-10121`). Inside Lua they are simply unsupported (Section 1). +- Remote keys: a parked remote `ApplyCc` acks with the cce's term; on timeout the txm probes the remote node's liveness via `BlockCcReqCheck` (`tx_operation.cpp:5890-5904`). If the owner fails over, the request errors out rather than hanging (and the engine recovers orphaned blocking txs via the `active_blocking_txs_` map and `RemoveExpiredActiveBlockingTxs`, `cc_shard.cpp:2649-2680`). + +### SCAN + +Cursor iteration does **not** use this machinery: SCAN cursors are cached per connection (`BucketScanCursor` / `scan_cursors` in `include/redis_connection_context.h:48-59`, `135-151`) and each SCAN call is an ordinary bounded request. + +--- + +## 4. Gotchas and invariants (verified) + +1. **Script atomicity is transaction-scoped; retries are whole-script.** A script aborted by an OCC conflict re-executes from the top (`src/redis_service.cpp:2596-2614`), and transactional effects roll back. `PUBLISH` is a `DirectCommand` that runs outside the transaction, so it is not part of that atomic unit. +2. **Script cache is per node.** `scripts_` is a plain member of `RedisServiceImpl`; EVALSHA against a different node than the SCRIPT LOAD target yields NOSCRIPT. Also, `SCRIPT FLUSH` clears only the sha→body map; compiled `f_` globals linger in pooled interpreters (harmless, since EVALSHA checks `scripts_` first). +3. **EVAL is barred from MULTI/BEGIN** (`src/redis_handler.cpp:1964-1973`) — no nested transactions. +4. **Pub/sub is at-most-once and node-local in bookkeeping.** No durability, no cross-node subscriber registry; remote fan-out goes only to node-group leaders, fire-and-forget. PUBLISH's return value counts local deliveries only. +5. **Pub/sub writes bypass the reply pipeline** and share the connection's single `output`/arena; teardown vs. publish has an acknowledged race (`src/redis_connection_context.cpp:38` Fixme). +6. **Blocking pops are two-phase on purpose**: the `BlockLock` wait step is read-only (not logged); only the post-wake `PopElement` step takes the write lock and hits the WAL (`src/redis_command.cpp:5353-5355` comments, `object_cc_map.h` Block branch). The wake handoff upgrades the parked tx to WriteLock *before* re-execution, so no competing pop can steal the element (`non_blocking_lock.cpp:695-700`). +7. **Timeout granularity is ~10 ms** (the `CheckWaitingTxs` cadence), and a "0 = forever" timeout is actually one year (`src/redis_command.cpp:19847-19850`). +8. **Blocked clients are observable**: `INFO clients` reports `blocked_clients` maintained around `ExecuteMultiObjTxRequest` (`src/redis_service.cpp:4528-4535`, `4615-4618`). diff --git a/docs/05-namespaces.md b/docs/05-namespaces.md new file mode 100644 index 00000000..9fa0d219 --- /dev/null +++ b/docs/05-namespaces.md @@ -0,0 +1,209 @@ +# 05 — Namespace Isolation & Management + +EloqKV namespaces (PR #479) are token-authenticated, multi-tenant key spaces layered on top of the +Data Substrate engine. A namespace is created by an administrator with `NAMESPACE ADD `, which +returns a random 128-bit token; a client enters the namespace by sending that token as the password +in `AUTH`. Isolation is implemented purely by **key prefixing inside one shared engine table** +(`ns_data_0`), not by separate tables: every user key is transparently prepended with +`b255(ns_id) ':' b255(epoch) ':'` at `EloqKey` construction time, driven by a bthread-local +"current namespace" that is set per dispatched command from the connection context. Namespace +metadata lives in a dedicated engine table (`__ns_0`) and all metadata mutations are multi-key +engine transactions. `FLUSHDB`/`FLUSHALL` in a namespace is an O(1) epoch bump; a background GC +bthread later range-deletes the orphaned prefix through normal engine transactions. Related docs: +[02-command-processing.md](02-command-processing.md) for the dispatch pipeline, +[03-data-model.md](03-data-model.md) for `EloqKey`, and +`data_substrate/docs/04-transaction-execution.md` / `05-data-model-and-catalog.md` for the engine +transaction and table machinery used here. + +## 1. Purpose & model + +- A **namespace** is a named, isolated key space. The implicit namespace is `default` + (`kDefaultNamespace`, `include/namespace/context.h:13`); it is the entire pre-existing keyspace + and cannot be added, refreshed, or deleted (`src/redis_command.cpp:1606-1610,1638,1677-1681`). +- Names are arbitrary byte strings of 1–255 bytes (`src/redis_command.cpp:1612-1618`); the only + reserved name is `default`. Names are *not* embedded in data keys — a monotonically increasing + `uint64` **namespace id** is allocated from a `next_id` counter and used instead + (`src/namespace/storage.cpp:211-243`). There is no explicit limit on namespace count. +- Relationship to the 16 Redis databases: the default namespace keeps the classic per-DB engine + tables `data_table_0..15`. **All non-default namespaces share one engine table** `ns_data_0`, and + `SELECT` is rejected inside a namespace with `RD_ERR_SELECT_FORBIDDEN_IN_NS` + (`src/redis_command.cpp:1736-1743`) — a namespace has exactly one logical DB. +- Three prebuilt tables are registered at startup (`src/redis_service.cpp:236-281`): + `data_table_<0..15>` (default namespace), `__ns_0` (namespace metadata, + `RedisServiceImpl::NamespaceTableName()`), and `ns_data_0` (all namespace data, + `RedisServiceImpl::NsDataTableName()`), all `TableEngine::EloqKv` primary tables. + +## 2. Isolation mechanism: key prefixing, baked in at key construction + +Verified: isolation is **prefix-based within a shared table**, not table-per-namespace. + +- The data-key prefix is `encoded_ns_id ':' b255(epoch) ':'` + (`NamespacePrefix::MakePrefix`, `include/namespace/prefix.h:16-25`). `b255e()` encodes integers + base-255 while skipping the byte `0x3A` (`:`), so encoded ids/epochs can never contain the + delimiter (`src/b255.cpp:11-28`) — parsing is unambiguous and prefixes are dense (≤9 bytes per + component). +- A bthread-local variable holds the *current namespace prefix* (`GetCurrentNamespace()`, + `src/namespace/context.cpp:32-50`; falls back to a thread-local for non-bthread callers). The + RAII `NamespaceGuard` swaps it for a scope (`include/namespace/context.h:52-69`). +- `RedisServiceImpl::DispatchCommand` installs `NamespaceGuard ns_guard(ctx->ns_id)` before any + command logic runs (`src/redis_service.cpp:5935-5956`), after re-validating the connection's + cached namespace metadata (see §5). +- The scoping hook is the `EloqKey` constructor itself: `EloqKey(std::string_view)` routes through + `CreateEloqStringFromNamespace`, which prepends the current prefix via `ComposeNamespaceKey` + (`include/eloqkv_key.h:55-58`, `include/namespace/context.h:18-50`). For `default`/empty the key + is unchanged. Since every command parser builds keys with this constructor (333 call sites in + `src/redis_command.cpp` vs. 2 deliberate `EloqKey::Raw` uses), every command is scoped without + per-command code. `EloqKey::Raw` (`include/eloqkv_key.h:72-80`) is the explicit bypass used only + for already-composed or metadata keys (namespace module, GC, DBSIZE range bounds). +- Table routing: `RedisServiceImpl::RedisTableName(db_id)` returns `data_table_` when the + current prefix is empty/`default`, otherwise `ns_data_0` (`src/redis_service.cpp:5863-5870`). +- Cross-namespace access is prevented because clients never supply the prefix — it is derived from + the token they authenticated with, and `:`-free b255 encoding means no crafted user key can + collide with another tenant's prefix. Range operations (SCAN/KEYS/DBSIZE) are clamped to + `[prefix, MakePrefixNext(prefix))` (`include/namespace/prefix.h:56-74`). + +An important invariant: **the prefix is baked into the key bytes when the command is parsed**, so +queued `MULTI` commands, blocking commands, and engine-side retries keep their scope even if they +later execute on a different bthread. + +## 3. Lifecycle: create / authenticate / use / drop + +Management is exposed as a `NAMESPACE` command (handler registered at +`src/redis_service.cpp:1394-1396`; parsing at `src/redis_command.cpp:10600-10644`): +`NAMESPACE CURRENT | GET | ADD | REFRESH | DEL `. + +Management preconditions (`NamespaceCommand::Execute`, `src/redis_command.cpp:1540-1572`): all +subcommands except `CURRENT` require (a) cluster mode disabled, (b) `requirepass` configured, and +(c) the caller authenticated with `requirepass` in the `default` namespace — i.e. only the admin +manages namespaces. + +- **Create** (`ADD`): generates a random token (re-rolled while it equals `requirepass`, + `src/redis_command.cpp:1619-1623`) and calls `NamespaceManager::Add` → + `NamespaceStorage::Add` (`src/namespace/storage.cpp:150-326`). Inside **one engine transaction** + (`RepeatableRead` + `Locking`, committed with `txservice::CommitTx`) it checks name and token + uniqueness, allocates `next_id`, and writes five records to `__ns_0`: + + | key in `__ns_0` | value | purpose | + |-----------------------------|----------------------|----------------------------------| + | `n:` | raw 16-byte token | name → token | + | `t:` | `b255(id)` | token → encoded id | + | `i:` | name | id → name | + | `e:` | epoch (decimal text) | current epoch, starts at `"1"` | + | `next_id` | next counter value | id allocator | + + (prefix constants at `include/namespace/storage.h:13-20`). The reply is the token in base64url + (22 chars). These ops run through `RedisServiceImpl::ExecuteNamespaceTxRequest` + (`src/redis_service.cpp:1201-1217`), a thin typed dispatcher over the normal engine + `TxRequest` path — so namespace metadata gets the engine's atomicity, replication, and + persistence for free (see `data_substrate/docs/04-transaction-execution.md`). +- **Authenticate**: `AuthCommand::Execute` (`src/redis_command.cpp:1495-1525`) first interprets the + password as a token (`NamespaceToken::FromBase64Url`); if + `NamespaceManager::GetMetadataByToken` resolves it, the connection binds to that namespace + (`ctx->ns`, `ctx->ns_id`, `ctx->ns_meta`). Otherwise the password is compared with `requirepass` + for default-namespace access. `GetMetadataByToken` consults an RCU-cached + `token → NamespaceMetadata` map and falls back to a `__ns_0` lookup + (`src/namespace/manager.cpp:221-266`; RCU container in `include/rcu.h`). +- **Use**: every dispatched command recomputes `ctx->ns_id` from the live metadata epoch and + installs the guard (§5). +- **Rotate** (`REFRESH`): issues a new token for the namespace inside one engine transaction, + deleting the old `t:` record and keeping the id/epoch (`src/namespace/storage.cpp:328-567`); + the manager cache entry is invalidated (`src/namespace/manager.cpp:153-165`). +- **Drop** (`DEL`): in one engine transaction, reads token/id/epoch, writes a GC record + `g::` = `"1"`, and deletes the `n:`, `t:`, `i:`, `e:` records + (`src/namespace/storage.cpp:569-745`). Data keys in `ns_data_0` are *not* touched here — the + commit makes them unreachable (token gone), and the GC record schedules asynchronous physical + deletion. Because record + tombstones commit atomically, a crash either leaves the namespace + intact or leaves a durable GC record. +- **Flush** (`FLUSHDB`/`FLUSHALL` inside a namespace — both take the same path, + `src/redis_service.cpp:2907-2918`): bumps `e:` to `epoch+1` and writes the GC record for the + *old* epoch in the same transaction (`src/redis_service.cpp:2757-2821`), then updates the cached + atomic epoch. All existing keys instantly become invisible (new prefix), making flush O(1) for + the client; the old-epoch keys are GC'd later. + +## 4. Garbage collection of dropped/flushed prefixes + +`NamespaceGc` (`include/namespace/gc.h`, `src/namespace/gc.cpp`) is one background bthread started +after service init and joined on shutdown (`src/redis_service.cpp:762-763,782`; +`gc.cpp:18-33`). + +- **Trigger/scan**: the daemon loops while `!server_->IsStopping()`, range-scanning `__ns_0` over + `["g:", "g;")` for GC records (`ScanGCRecords`, `gc.cpp:107-218`) with a `ReadCommitted` + + `OccRead` transaction that is aborted afterwards (read-only). If none exist it sleeps 30 s in + 100 ms slices so shutdown stays responsive (`gc.cpp:53-64`). +- **Deletion path**: for each record it reconstructs the dead data prefix + `MakePrefix(b255_id, epoch)` (`gc.cpp:73-93`, safe to parse on `:` because b255 output never + contains it) and calls `CleanPrefixKeys` (`gc.cpp:220-411`): repeatedly (1) scan + `[prefix, prefixNext)` in `ns_data_0` with a read-only tx, (2) delete every found key with + `DelCommand`s inside a single `RepeatableRead`+`Locking` engine transaction, (3) sleep and + rescan, until a scan round finds zero keys. Deletion goes through the normal engine write path — + no direct KV-store calls — so checkpointing propagates the deletes to the data store + (see `data_substrate/docs/07-durability-and-recovery.md` / `09-store-handler.md`). +- **Throttling**: 5 ms between successful delete rounds, 50 ms after a failed one + (`gc.cpp:409`), 100 ms backoff on scan/tx-creation errors, 10 ms between GC records + (`gc.cpp:101`), and `IsStopping()` checks at every loop boundary. +- **Crash safety / idempotency**: the GC record is committed atomically with the epoch bump or + metadata deletion, so it survives crashes and is re-discovered on restart; cleanup is pure + range-delete and safe to repeat. The record itself is deleted (own small transaction, + `DeleteGCRecord`, `gc.cpp:413-439`) only after a fully empty scan; malformed records are dropped + defensively (`gc.cpp:74-90`). + +## 5. Connection binding & command interactions + +Per-connection state lives in `RedisConnectionContext` +(`include/redis_connection_context.h:122-124`): `ns` (name, default `"default"`), `ns_id` (the +composed key prefix; empty for default), and `ns_meta` (shared `NamespaceMetadata` with +`encoded_id` and an atomic `epoch`, `include/namespace/manager.h:17-23`). + +- **Entry**: only via `AUTH ` (or `AUTH `; the username is ignored by + `AuthCommand::Execute`). There is no `HELLO AUTH` path — EloqKV has no HELLO handler; `hello`, + `auth`, `quit`, `reset`, plus the special-case `NAMESPACE CURRENT`, are the only commands allowed + pre-auth (`AuthRequired`, `src/redis_service.cpp:5877-5915`). Namespace *management* + (`NAMESPACE ADD`/`REFRESH`/`DEL`) is refused when `requirepass` is empty + (`src/redis_command.cpp:1558-1562`), so namespace tokens can only be created on a + password-protected server. Authentication itself is token-based; note that only the 2-argument + `AUTH` form errors on an empty `requirepass` (`src/redis_command.cpp:10558-10567`) — the + 3-argument `AUTH ` form does not. +- **Per-command re-validation** (`src/redis_service.cpp:5935-5956`): if `ctx->ns_meta` is set, + dispatch re-fetches the live metadata by token and requires pointer equality with the cached + object; on success it recomputes `ctx->ns_id` from the current epoch (so another connection's + `FLUSHDB` takes effect immediately via the shared atomic), otherwise it resets the connection + to the default namespace. Then the `NamespaceGuard` scopes everything the + command does. +- **MULTI / Lua**: both run inside `DispatchCommand` under the guard + (`src/redis_service.cpp:5999-6008`), and keys are prefixed at parse time, so queued transactions + and `redis.call` from scripts are scoped correctly. +- **SCAN / KEYS / DBSIZE**: these manage prefixes explicitly. The scan path disables the ambient + guard (`NamespaceGuard ns_guard("")`, `src/redis_service.cpp:5005`), composes + `[ns_id ⊕ pattern-prefix, next)` bounds itself, pushes `ComposeNamespaceKey(ns, pattern)` down as + the engine-side filter, and strips `ns_id` from returned keys before matching/replying + (`src/redis_service.cpp:5002-5062,5237-5240,5329-5334`). `DBSIZE` counts a prefix-bounded scan + with `EloqKey::Raw` bounds (`src/redis_command.cpp:2224-2270`). +- **SELECT** is forbidden inside a namespace (`src/redis_command.cpp:1736-1743`); `db_id` is + ignored by table routing anyway for non-default namespaces. + +## 6. Gotchas & invariants (verified unless marked inference) + +1. **Tokens are bearer credentials stored in plaintext.** 16 random bytes (OpenSSL `RAND_bytes`, + UUIDv4-formatted; falls back to `std::random_device` if OpenSSL fails, + `src/namespace/token.cpp:11-30`) stored raw in `__ns_0` and listable by the admin via + `NAMESPACE GET *`. A token equal to `requirepass` is rejected/re-rolled at every layer + (`src/namespace/storage.cpp:69-72,156-159,334-337`; `src/redis_command.cpp:1620-1623`) so the + admin password can never resolve as a namespace token. +2. **Cluster mode**: namespace management is refused when `FLAGS_cluster_mode` is set + (`src/redis_command.cpp:1550-1556`). Inference: this is because the `NamespaceManager` RCU cache + is per-process and has no cross-node invalidation. +3. **GC vs. in-flight writes**: a command parsed before a concurrent flush/drop carries the + old-epoch prefix and may commit after the epoch bump; it is invisible to the namespace and is + normally mopped up because GC rescans until empty. Inference: a write landing *after* GC's final + empty scan (record already deleted) would leak an invisible orphan key in `ns_data_0`. +4. **Epoch-in-prefix makes flush O(1) but defers space reclamation** to the GC daemon's paced + range-deletes; large namespaces are deleted in scan-sized batches, each batch one engine + transaction (`src/namespace/gc.cpp:355-403`). +5. **`MemoryNamespaceStorage`** (`src/namespace/manager.cpp:13-133`) is an in-memory + `INamespaceStorage` with no GC; no production call site constructs it (the service always wires + `NamespaceStorage`, `src/redis_service.cpp:213`). Note it stores `b255prefix(id)` (with trailing + `:`) as the id where the persistent storage stores bare `b255e(id)` + (`manager.cpp:32` vs. `storage.cpp:243`) — harmless today but inconsistent if ever swapped in. +6. **Default namespace fast path is zero-cost**: `ComposeNamespaceKey` returns the key untouched + for `default`/empty (`include/namespace/context.h:18-30`), so pre-namespace deployments see no + key-format change and no overhead beyond the per-dispatch guard swap. diff --git a/docs/06-vector-search.md b/docs/06-vector-search.md new file mode 100644 index 00000000..7f07ae83 --- /dev/null +++ b/docs/06-vector-search.md @@ -0,0 +1,115 @@ +# Vector Search (EloqVec) + +EloqKV ships an optional vector-search module ("EloqVec") that adds eight `ELOQVEC.*` commands for creating HNSW indexes, inserting/updating/deleting vectors with typed metadata, and running (optionally filtered) k-NN searches. The ANN structure itself is **not** a `TxObject` and lives outside the engine's cc maps: each node keeps a process-local in-memory [usearch](https://github.com/unum-cloud/usearch) index, while everything that must be durable — index metadata and a sharded delta log of every mutation — is stored as ordinary records in an engine-internal hash table (`__vector_index_meta_table`) via normal transactions, so it inherits the engine's WAL/checkpoint durability (`data_substrate/docs/07-durability-and-recovery.md`). Snapshots of the in-memory index are saved to a local file and optionally uploaded to object storage through an `rclone rcd` sidecar; recovery on any node is lazy: download/load the last snapshot, then replay the delta log. + +Compile-time gated: `VECTOR_INDEX_ENABLED` (default `OFF`, `CMakeLists.txt:145`) — when on, usearch v2.21.0 and nlohmann/json are fetched at build time (`CMakeLists.txt:302-326`) and `src/vector/*.cpp` is compiled into the server (`CMakeLists.txt:383-394`). + +## Key files + +| File | Role | +|---|---| +| `include/vector/vector_handler.h`, `src/vector/vector_handler.cpp` | `VectorHandler` singleton: all index lifecycle + data ops, index cache, persistence | +| `include/vector/vector_index.h`, `hnsw_vector_index.h/.cpp` | abstract index API and the usearch-backed HNSW implementation | +| `include/vector/log_object.h`, `src/vector/log_object.cpp` | sharded delta log stored as engine records | +| `include/vector/vector_type.h`, `src/vector/vector_type.cpp` | `IndexConfig`, `VectorIndexMetadata`, `VectorRecordMetadata` (schema), `VectorId`, serialization | +| `include/vector/predicate.h`, `src/vector/predicate.cpp` | JSON filter → predicate tree, evaluated against per-vector metadata | +| `include/vector/cloud_manager.h`, `src/vector/cloud_manager.cpp` | object-storage upload/download via a spawned `rclone rcd` process | +| `include/vector/vector_util.h` | memcmp-comparable binary encoding of metadata values | +| `src/redis_command.cpp:20520-21356`, `src/redis_service.cpp:5539-5861` | command parsing and execution glue | + +## 1. Command surface + +Commands are registered in `src/redis_service.cpp:1896-1928` and parsed in `src/redis_command.cpp` (`RedisCommandType::ELOQVEC_*`, `src/redis_command.cpp:274-281`): + +| Command | Syntax | Notes | +|---|---|---| +| `ELOQVEC.CREATE` | `index config_json [schema_json]` | `config_json` fields are **order-sensitive** (parsed with `nlohmann::ordered_json`): `dimension` (uint > 0), `metric` (`L2SQ`/`L2`, `IP`, `COSINE`), `algorithm` (`HNSW` only), `persist_strategy` (`EVERY_N` requires `threshold` > 0; `MANUAL` → threshold −1), then optional HNSW params `m`, `ef_construction`, `ef_search` (`src/redis_command.cpp:20520-20767`, allowed-key check `src/vector/hnsw_vector_index.cpp:138-161`). `schema_json` is an ordered object defining the metadata schema; each field maps to a type from `INT32 / INT64 / DOUBLE / BOOL / STRING`. | +| `ELOQVEC.ADD` | `index key vector ["metadata_json"]` | `key` is a uint64; `vector` is whitespace-separated floats; `metadata_json` is a **JSON array** of values in schema order, not an object (`VectorRecordMetadata::Encode`, `src/vector/vector_type.cpp:199-234`). | +| `ELOQVEC.BADD` | `index key_count key1 vec1 [meta1] key2 vec2 [meta2] …` | all-with-meta or all-without; max 10000 entries (`MAX_BATCH_ADD_SIZE`, `include/redis_command.h:7092`). | +| `ELOQVEC.UPDATE` | `index key vector ["metadata_json"]` | fails if `key` is not in the index. | +| `ELOQVEC.DELETE` | `index key` | fails if `key` is not in the index. | +| `ELOQVEC.SEARCH` | `index k vector ["filter_json"]` | replies array of `[id, distance]` pairs (`src/redis_command.cpp:21331-21355`); no hydration of vectors or any Redis keys. | +| `ELOQVEC.INFO` | `index` | dumps config/params/timestamps; `status` is hard-coded `"ready"` (`src/redis_command.cpp:20838-20839`). | +| `ELOQVEC.DROP` | `index` | removes metadata, log shards, cache entry, local + cloud snapshot files. | + +Filter JSON (Mongo-style, `src/vector/predicate.cpp:101-347`): leaf operators `$eq $ne $gt $gte $lt $lte $in`, combinators `$and $or $not`, implicit AND for multiple fields/ops at one level. Fields must exist in the schema except the pseudo-field `id` (compared as Int64 against `VectorId::id_`, `predicate.cpp:212-215, 440-443`). All comparisons are `memcmp` over the order-preserving binary encoding produced at write time (`include/vector/vector_util.h:105-195` — sign-flipped big-endian ints, MySQL-style sortable doubles). + +Vector IDs are bare uint64s chosen by the client; there is **no linkage to Redis keys, databases, or namespaces** — the module is a self-contained store keyed by `(index_name, id)`, global to the cluster (not db- or namespace-scoped, cf. [05-namespaces.md](05-namespaces.md)). + +## 2. Architecture + +Three kinds of state per index named `N`: + +1. **Metadata record** — key `vector_index:N:metadata` in the engine-internal table `__vector_index_meta_table` (`TableType::Primary`, `TableEngine::InternalHash`, hash-partitioned; `include/vector/vector_type.h:36-42`). Serialized `VectorIndexMetadata`: name, `IndexConfig`, schema, persist threshold, current snapshot `file_path`, created/last-persist timestamps (`src/vector/vector_type.cpp:393-459`). +2. **Sharded delta log** — 1024 shards (`VECTOR_INDEX_LOG_SHARD_COUNT`, `src/vector/vector_handler.cpp:50`). Each shard is a metadata record `log:meta:vector_index:N:shard_` plus one record per entry `log:item:vector_index:N:shard_:` in the *same* internal table (`src/vector/log_object.cpp:180-220`). A log item is `{INSERT|UPDATE|DELETE, serialized VectorId (id + encoded metadata), serialized float vector, ts, seq}`. Shard chosen by FNV-1a of the decimal id (`log_object.cpp:193-214`). +3. **In-memory usearch index + snapshot file** — `HNSWVectorIndex` wraps `index_dense_gt`; the usearch key type is the whole `VectorId`, so the encoded metadata rides inside the ANN structure itself, while hashing/equality use only `id_` (`src/vector/hnsw_vector_index.cpp:707-763`). Snapshot file: `/N-.index`, with sentinel timestamp `0000000000000000` meaning "never persisted" (`vector_type.cpp:359-381`, `vector_type.h:44`). + +(1) and (2) are replicated/durable engine state shared by the whole cluster; (3) is **per-process**, held in `VectorHandler::vec_indexes_` (name → `{shared_ptr, shared_ptr}`) under a `std::shared_mutex` (`include/vector/vector_handler.h:277-281`). + +### Threading model + +- A dedicated `TxWorkerPool` named `"vindex"` (plain `std::thread`s, default 1 thread, `--vector_index_worker_num`, `src/redis_service.cpp:109-112, 495-506`) executes every vector command: the brpc bthread submits a closure and blocks on a `bthread::ConditionVariable` until it finishes (`src/redis_service.cpp:5550-5575` and siblings). So vector work never runs on TxProcessor threads or brpc workers (cf. `data_substrate/docs/02-threading-model.md`); the worker threads drive their own `TransactionExecution` via `NewTxInit`/`Execute`/`Wait`. +- The worker's `thread_id` is forwarded to usearch as the search-context slot (`vector_handler.cpp:448`, `hnsw_vector_index.cpp:313-337`). +- Inside `HNSWVectorIndex`, add/remove/update/get/search take only a **shared** lock — concurrent mutation safety is delegated to usearch's internal synchronization; the exclusive lock is reserved for `initialize`, `save`, and parameter changes, so a snapshot save blocks all index ops (`hnsw_vector_index.cpp:35, 93, 295, 380, …`). +- Async persistence runs on the same pool via `SubmitWork` (`vector_handler.cpp:552-554`). + +The handler singleton is created in `RedisServiceImpl::Start` once the tx service exists, with `data_path` from the engine core config and an optional `CloudConfig` (`vector_cloud_endpoint`/`vector_cloud_base_path` flags or `[store]` ini keys, `src/vector/vector_type.cpp:31-34, 465-475`; wiring `src/redis_service.cpp:743-760`). If a cloud endpoint is configured but `rclone` can't be started/connected, server startup fails. + +## 3. Write path (ADD; UPDATE/DELETE/BADD analogous) + +`VectorHandler::Add` (`src/vector/vector_handler.cpp:454-563`), all inside **one engine transaction** (`NewTxInit(RepeatableRead, OCC)`): + +1. Read `vector_index:N:metadata` (plain read) — missing → `INDEX_NOT_EXIST`. +2. `GetOrCreateIndex`: return cached entry, or lazily build it (§5). +3. Encode metadata JSON against the schema (array, exact arity) → binary blob inside `VectorId`. +4. `LogObject::append_log_sharded`: reads the shard's `log:meta:…` **for write** (write intent; `is_for_write=true`, `log_object.cpp:424-437`, ctor `data_substrate/tx_service/include/tx_request.h:217-230`), assigns the next sequence id, upserts one `log:item:…` record per entry, updates shard meta. The write intent on the shard meta serializes all writers of that shard (`log_object.cpp:463-465`). +5. Mutate the in-memory usearch index (**before commit**). +6. `CommitTx`. On commit failure, compensate the in-memory index (remove the just-added id; UPDATE restores the previous vector; DELETE re-adds it) (`vector_handler.cpp:529-535, 651-657, 760-766`). +7. If `persist_threshold != −1` and `log_count_of_this_shard × 1024 ≥ threshold` and no persist is already pending for this index, enqueue `PersistIndex(name)` on the worker pool (`vector_handler.cpp:543-556`). + +`BatchAdd` groups entries by shard id (sorted `std::map` traversal to keep shard-lock acquisition ordered and deadlock-free, `vector_handler.cpp:866-880`) and appends per shard within the same transaction. + +## 4. Query path (SEARCH) + +`VectorHandler::Search` (`vector_handler.cpp:382-452`): meta read → `GetOrCreateIndex` → if `filter_json` present, `PredicateExpression::Parse` against the schema (parse failure → `METADATA_OP_FAILED`, including filters on schema-less indexes) → `HNSWVectorIndex::search`. With a filter, usearch's `filtered_search` invokes the predicate callback **during graph traversal** (pre-filtering, not post-filtering): the callback decodes field offsets from the metadata blob embedded in the candidate's `VectorId` and evaluates the tree by memcmp (`vector_handler.cpp:436-444`, `hnsw_vector_index.cpp:312-330`, `predicate.cpp:378-471`). Results are ids + distances only; the `exact` brute-force flag exists in the API but the handler always passes `false` (`vector_handler.cpp:448-449`). The wrapping transaction does no data reads beyond the meta record (plus log replay if the index was lazily initialized) and is committed at the end. + +## 5. Durability, recovery, persistence + +**Durability of every mutation** comes from step 4 of the write path: log items are ordinary records in an engine table, so they hit the engine WAL at commit and checkpoint to the kv store like any write (`data_substrate/docs/07-…`, `10-log-service.md`). The usearch structure is only a serving cache of (snapshot ⊕ log). + +**Lazy rebuild** — `CreateAndInitializeIndex` (`vector_handler.cpp:1016-1089`), triggered the first time any node touches an index after restart/failover/drop-from-cache: + +1. Deserialize metadata from the record just read. +2. If `file_path` does not contain the initial-timestamp sentinel (i.e., a snapshot exists): with a cloud manager, delete any local copy and download `file_path` minus the `data_path` prefix from object storage; without one, require the local file (`vector_handler.cpp:1043-1063`) — so in cloud-less multi-node setups, a node that didn't write the snapshot cannot rebuild (`INDEX_INIT_FAILED`). +3. `initialize` = create usearch index from config + load snapshot file if present (`hnsw_vector_index.cpp:32-89`). +4. Replay the full sharded log via `ApplyLogItems` (scan all 1024 shards, batch consecutive INSERTs, apply UPDATE/DELETE as barriers) (`vector_handler.cpp:1260-1375`). + +**Snapshotting** — `PersistIndex` (`vector_handler.cpp:1091-1258`), its own transaction: + +1. Read metadata **for write** (write intent on the meta record). +2. `truncate_all_sharded_logs`: per shard, read shard meta for write (blocking all concurrent writers of the index for the duration), delete every item record, reset counters (`log_object.cpp:669-793, 1079-1118`). +3. Save usearch index to `/N-.index` (exclusive lock → searches/writes on this node stall). +4. Update metadata (`file_path`, `last_persist_ts`) via upsert; upload the new file to `:/N-.index` through the rclone REST API (`cloud_manager.cpp:273-295`); then `CommitTx`. Any failure aborts the transaction (log truncation and meta update roll back; the in-memory index is untouched). +5. Post-commit: delete the old snapshot locally and in the cloud (best effort). + +The trigger dedupe set `pending_persist_indexes_` guarantees at most one queued persist per index (`vector_handler.cpp:546-551`). + +**Cloud sidecar** — `CloudManager` spawns `rclone rcd --rc-no-auth --rc-addr=127.0.0.1:15572` (logs to `/tmp/vector_cloud_service.log`) and drives it with libcurl JSON POSTs (`operations/copyfile`, `deletefile`, `mkdir`); the bucket (first path segment of `base_path`) is created at startup (`cloud_manager.cpp:119-271`). The remote object name is the snapshot path with the local `data_path` prefix stripped. + +## 6. Consistency model + +- **Per-command, not session-transactional.** Each `ELOQVEC.*` op opens and commits its own internal transaction; the commands do not participate in `MULTI/EXEC`, `BEGIN/COMMIT`, or Lua scripts (the handlers bypass the txm-per-connection machinery of [02-command-processing.md](02-command-processing.md) entirely). +- **Durable state is transactional; the served index is eventually consistent with it.** Log append + meta ops commit atomically, but the in-memory usearch mutation happens *before* commit with post-hoc compensation. A concurrent search on the same node can transiently observe an entry whose transaction later aborts (and briefly miss a deleted-then-restored one). There is no read-your-write guarantee across nodes: another node's cached index instance only learns about writes it didn't apply locally when it rebuilds from snapshot + log. +- **Single-writer-per-shard.** The write intent on each shard's log meta record serializes writers of that shard cluster-wide; persistence takes intents on all 1024 shards, momentarily quiescing all writes to the index. +- Every data/search op re-reads the metadata record first, so a dropped index is detected immediately on any node even if its cache entry survives. + +## 7. Gotchas / verified invariants + +- **Persist trigger is a coarse estimate**: per-shard count × 1024 assumes uniform shard fill (`vector_handler.cpp:545`). +- **`PersistIndex(force)` is dead**: the `force` parameter is never read in the body (`vector_handler.cpp:1091`). +- **Metadata for ADD is a positional JSON array** matching schema order — an object is rejected (`vector_type.cpp:215-219`). Filters, in contrast, are objects keyed by field name. +- **Float metadata fields are strict**: a `DOUBLE` field rejects integer JSON literals (`is_number_float()` check, `vector_util.h:150-156`); `Int32` overflow is rejected. +- **Per-entry engine round-trips.** Log append/truncate issue one `UpsertTxRequest` per item (TODO batch, `log_object.cpp:482`); DROP/persist of a 1024-shard log touches every shard meta plus every live item in one transaction. +- **usearch `update` = remove + add** with the remove error deliberately ignored (`hnsw_vector_index.cpp:557-565`), but `VectorHandler::Update/Delete` first `get()` the id and fail with the index untouched if it doesn't exist (`vector_handler.cpp:639-646, 748-755`). +- **`max_elements` is fixed at 1,000,000** (`IndexConfig` default, `vector_type.h:282`) — it is not exposed in `ELOQVEC.CREATE`'s parser, and usearch capacity is reserved up-front at that size. +- Test expectations (filter semantics, persist/reload round-trips, concurrency of add/search) live in `src/vector/tests/VectorHandler-Test.cpp`, `VectorCache_HNSW-Test.cpp`, `LogObject-Test.cpp`; the tests register `__vector_index_meta_table` as a prebuilt engine table and run with `skip_wal`/`skip_kv` (`VectorHandler-Test.cpp:75-120`). diff --git a/docs/07-persistence-and-tools.md b/docs/07-persistence-and-tools.md new file mode 100644 index 00000000..b40d7db1 --- /dev/null +++ b/docs/07-persistence-and-tools.md @@ -0,0 +1,130 @@ +# 07 — Redis-Format Persistence Interop & Offline Tools + +**Summary.** EloqKV does not persist data through RDB snapshots or AOF rewrite the way Redis does — native durability is the engine's replicated WAL plus background checkpoints into the kv store (see `data_substrate/docs/07-durability-and-recovery.md`). What this layer provides instead is *interop* with the Redis on-disk/wire formats for migration and tooling: (1) online `DUMP`/`RESTORE` commands that serialize/deserialize single values in Redis RDB payload format (`src/redis_rdb_restore.cpp`, `src/redis_command.cpp`), so `redis-cli --migrate`-style key copying works in both directions; and (2) offline exporters `eloqkv_to_rdb` and `eloqkv_to_aof` (`src/tools/`) that read the checkpointed kv store directly — either an embedded RocksDB directory or, since PR #485, a rocksdb-cloud snapshot in S3/GCS — and emit a Redis-loadable `dump.rdb` or AOF command stream. `SAVE`/`BGSAVE`/`BGREWRITEAOF` are not implemented at all; they return `ERR unknown command`. + +Related docs: [03-data-model.md](03-data-model.md) (the `RedisEloqObject` types and their native `Serialize()` format these tools convert from/to), [05-namespaces.md](05-namespaces.md), engine side `data_substrate/docs/07-durability-and-recovery.md` and `09-store-handler.md`. + +## 1. Positioning + +| Mechanism | Online? | Purpose | Source of truth | +|---|---|---|---| +| Engine WAL + checkpoint | yes | actual durability/recovery | `data_substrate` (TxLog + Checkpointer) | +| `DUMP` / `RESTORE` | yes | per-key migration to/from real Redis | `src/redis_command.cpp:15944`, `src/redis_rdb_restore.cpp` | +| `eloqkv_to_rdb` | no (offline) | full-store export to a Redis `dump.rdb` | `src/tools/eloqkv2rdb/eloqkv2rdb.cpp` | +| `eloqkv_to_aof` | no (offline) | full-store export to RESP command files | `src/tools/eloqkv2aof/eloqkv2aof.cpp` | + +There is no RDB *loader* for whole files: imports into EloqKV go through `RESTORE` (or replaying an AOF stream of normal commands). + +Vendored libs used here: `crcspeed/` is the CRC-64 (Jones polynomial, same as Redis) implementation used for DUMP payload and RDB file checksums (`CMakeLists.txt:403`); `fpconv/` is a Grisu2 double-to-shortest-string converter used by `d2string()` for score formatting (`include/redis_string_num.h:256`). + +## 2. DUMP / RESTORE + +### 2.1 DUMP (EloqKV → Redis payload) + +`DumpCommand::ExecuteOn` (`src/redis_command.cpp:15944`) converts the live object via `ConvertEloqObjectToRedisDumpPayload` (`src/redis_rdb_restore.cpp:1424`), then appends the standard 10-byte Redis DUMP footer: 2-byte little-endian RDB version `10` (`redis_dump_version_`, `include/redis_command.h:7287`) + 8-byte CRC-64 over everything before it (`src/redis_command.cpp:15958`). Encodings emitted are deliberately plain (no listpack/ziplist/intset, no LZF): + +| EloqKV type | RDB type byte emitted | +|---|---| +| String | 0 (`RDB_TYPE_STRING`) | +| List | 1 (`RDB_TYPE_LIST`, plain length-prefixed elements) | +| Set | 2 (`RDB_TYPE_SET`) | +| Hash | 4 (`RDB_TYPE_HASH`) | +| Zset | 5 (`RDB_TYPE_ZSET_2`, binary little-endian double scores) | + +Any other object type makes the conversion fail and DUMP returns a syntax error (`src/redis_rdb_restore.cpp:1486`, `src/redis_command.cpp:15949-15952`). Footer version 10 corresponds to Redis 7.0; real Redis only accepts payload versions ≤ its own RDB version, so (inference) Redis ≤ 6.x will reject EloqKV DUMP output while 7.x accepts it. + +### 2.2 RESTORE — version & checksum gate + +`ParseRestoreCommand` (`src/redis_command.cpp:20005`) verifies the payload *at parse time* via `RestoreCommand::VerifyDumpPayload` (`src/redis_command.cpp:16176`): + +- footer version `0` → legacy **EloqKV-native** payload (the object's own `Serialize()` image, produced by DUMP before Redis compatibility was added; `include/redis_command.h:7284-7286`). Checksum verified with `crc64speed_big` (initialized at server start, `src/redis_service.cpp:369`). +- footer version `4` or `10` → **Redis RDB** payload; checksum verified with `crc64` (byte-swapped on big-endian hosts). +- anything else (including version 9 from Redis 5/6 and version 11 from Redis 7.2+) → rejected: "DUMP payload version or checksum are wrong". + +Redis-format payloads are then converted into the native object image by `ConvertRedisDumpPayloadToEloqPayload` (`src/redis_rdb_restore.cpp:1417`). Supported RDB object encodings: + +| RDB type (byte) | Decodes to | Notes | +|---|---|---| +| STRING (0) | String | int8/16/32 and LZF-compressed strings handled (`Reader::ReadString`, `src/redis_rdb_restore.cpp:245`) | +| LIST (1), LIST_ZIPLIST (10), QUICKLIST (14), QUICKLIST2 (18) | List | quicklist2 plain + packed (listpack) node containers (`:1355-1396`) | +| SET (2), SET_INTSET (11), SET_LISTPACK (20) | Set | | +| ZSET (3, string scores incl. nan/±inf), ZSET_2 (5, binary doubles), ZSET_ZIPLIST (12), ZSET_LISTPACK (17) | Zset | | +| HASH (4), HASH_ZIPLIST (13), HASH_LISTPACK (16) | Hash | | +| Streams, modules, hash-with-field-TTL (21+), or any other type byte | **rejected** (`:1411-1412`) | | + +Decode hardening: LZF output capped at 64 MiB, collection counts capped at 2^20 entries and sanity-checked against remaining bytes (`kMaxLzfDecodedLength`/`kMaxCollectionEntries`, `src/redis_rdb_restore.cpp:71-72`, `ValidateCount :140`); ziplist/listpack blobs must self-consistently terminate (back-length, 0xFF terminator, total-bytes header checked). + +### 2.3 RESTORE — semantics + +- `REPLACE`, `ABSTTL` supported; `IDLETIME`/`FREQ` parsed and validated but ignored (`uint64_t idle_time_sec_{0}; // unsupport`, `include/redis_command.h:7374-7375`). +- TTL argument is milliseconds; without `ABSTTL` it is added to the current clock (`src/redis_command.cpp:20083-20087`). `RestoreCommand::CommitOn` builds the object from the native image and, if a TTL was given, swaps it for the TTL-variant object via `AddTTL` (`src/redis_command.cpp:16008-16098`). +- Key-exists handling is done through the TxCommand protocol: `ProceedOnNonExistentObject()=true`, `ProceedOnExistentObject()=replace_` (`include/redis_command.h:7336-7344`); the default result is `RD_ERR_BUSY_KEY_EXIST` ("BUSYKEY") which stands when the object exists and `REPLACE` was not given (`include/redis_command.h:7383`). +- Invalid payload + no `REPLACE`: the command is still routed (so BUSYKEY can win, matching Redis error precedence) carrying `payload_valid_=false`; with `REPLACE` it errors immediately (`src/redis_command.cpp:20096-20133`). +- DUMP/RESTORE resolve tables through the connection's selected DB and namespace like any other command, so they work inside custom namespaces (inference from normal dispatch; see [05-namespaces.md](05-namespaces.md)). + +## 3. eloqkv_to_rdb (offline RDB exporter) + +One source file, two very different builds, selected by `WITH_DATA_STORE` (`CMakeLists.txt:473-482`, install rules `:519-528`): + +| `WITH_DATA_STORE` | Tool built | Reads | +|---|---|---| +| `ROCKSDB` | `eloqkv_to_rdb`, `eloqkv_to_aof` | local embedded RocksDB dir (`--rocksdb_path`) | +| `ELOQDSS_ROCKSDB_CLOUD_S3` / `_GCS` | `eloqkv_to_rdb` only | rocksdb-cloud snapshot in S3/GCS (PR #485, commit 794c6ff) | +| anything else (incl. default `ELOQDSS_ELOQSTORE`) | none | — | + +Both paths only see **checkpointed** data: writes that are committed in the WAL but not yet flushed by the engine checkpointer are *not* in the kv store and will be missing from the export. + +### 3.1 Legacy local-RocksDB path (`Rocksdb2RDB`, `eloqkv2rdb.cpp:2068`) + +- Opens the RocksDB directory read-write (`rocksdb::DB::Open`), so it requires the server to be stopped (RocksDB LOCK file) — run it on a copy otherwise. +- Discovers tables via the hardcoded catalog keys `data_table_{0..15}_catalog` in the default column family, reading the `kv_cf_name` wide column to find each DB's column family (`:2128-2167`). Exactly 16 databases are assumed (`const int databases = 16`, `:2129`); a store created with a different `databases` config crashes on `CHECK(status.ok())`. +- Pipeline: single reader thread batches keys (`--round_batch_size`) into a pool sized `thread_count * pre_read_ratio`; `ParseWorker` threads deserialize the stored value (layout `[deleted i8][version i64][obj_type i8][payload]`, `:734-797`), drop deleted/TTL-expired entries, and append RDB-encoded bytes to pooled buffers; a single `WriteWorker` drains buffers to the file and folds them into the running CRC (`:572-625`). `SELECT db` opcodes are written between per-DB phases (`:2210-2217`). + +### 3.2 RocksDB-Cloud path (`RocksdbCloud2RDB`, `eloqkv2rdb.cpp:1690`) + +- **Point-in-time and safe against a live cluster**: it opens each shard's bucket with `cookie_on_open = ` and an empty `new_cookie_on_open` (`:1912-1913`), i.e. it mounts a named CLOUDMANIFEST branch created by the DSS backup API (`CreateSnapshotForBackup` rolls a branch named `{backup_name}-{shard_id}-{backup_ts}`, `data_substrate/store_handler/eloq_data_store_service/rocksdb_cloud_data_store.cpp:855-871`), with `disable_cloud_file_deletion=true` and no manifest roll (`:1862-1864`). `--snapshot_name` takes one comma-separated cookie per shard and must match `--shard_num` (`:2401-2408`). +- Shard object paths follow the DSS layout `/ds_{shard_id}` (`BuildShardObjectPath`, `:912`); shards are processed sequentially, each appending to the same output file. +- Within a shard, `--thread_count` > 1 splits the keyspace into ranges weighted by live-SST sizes (`BuildShardRangeBoundaries`, `:1469`) and scans them in parallel (`ScanShardRange`, `:1556`). Each flushed buffer re-emits its own `SELECT db` prefix, so out-of-order buffer interleaving from multiple scan threads still yields a semantically correct RDB (`:1648-1654`, `acquire_buffer :1593`). +- Key/value formats differ from the legacy path: keys are DSS-composite `{kv_table_name}/{partition_id}/{key}` (`ParseDssKey`, `:1349`); values are `[version_ts u64 (MSB = has_ttl)][ttl u64?][obj_type i8 + payload]` (`DeserializeDssValue`, `:1325`). Only tables whose kv name starts with `eloqkv_data_table_` are exported, with the DB index parsed from the suffix — FLUSHDB-renamed tables like `eloqkv_data_table_0_2026_...` still map to DB 0 (`ExtractDbNumberFromCatalogKey`, `:1432`). +- S3 specifics: optional static credentials (`--aws_access_key_id/--aws_secret_key`, otherwise instance credentials), MinIO-style endpoints via `--rocksdb_cloud_s3_endpoint_url` with path-style addressing (`:1816-1840`), tunable SST cache (`--rocksdb_cloud_sst_file_cache_size`), local scratch dir `--db_path` (default `/tmp/eloqkv_rdb_export`). A progress line per shard prints keys/bytes/rates (`ShardProgressPrinter`, `:1057`). + +### 3.3 Output RDB structure (both paths) + +Header `REDIS0006` (`RedisRdbUtil::ParseHeader`, `:213`) — RDB file version 6, old enough that only base types are legal — then per key: optional `0xFD` seconds-resolution expiry, type byte 0–4, length-prefixed key, plain (non-compact) value encoding; trailer `0xFF` + little-endian CRC-64 of the whole file (`:2049-2059`, `:2283-2294`). Integer-looking strings are stored with int8/16/32 special encodings (`OutputString`, `:134`); LZF compression is stubbed out (`:113-115`, `:162-174`). Zset scores are written as `std::to_string(score)` strings under RDB type 3, with single-byte nan/±inf markers (`:335-353`). + +## 4. eloqkv_to_aof (offline AOF exporter) + +`src/tools/eloqkv2aof/eloqkv2aof.cpp` — built only for `WITH_DATA_STORE=ROCKSDB`. Same legacy local-RocksDB discovery (16 hardcoded `data_table_N_catalog` entries, `:485-524`) and the same reader/parser pool pipeline, but each `ParseWorker` writes its **own** output file `/.aof` (`:531-533`), so the result is N independent RESP command streams, each self-contained (each tracks `last_db_idx_` and emits its own `SELECT`, `:301-305`). Object → command mapping (`RedisReplyUtil::ParseEloqKV`, `:98-200`): + +| Type | Commands emitted | +|---|---| +| String | one `SET key value` | +| List | one `RPUSH key elem` **per element** | +| Hash | one `HSET key field value` per pair | +| Set | one `SADD key member` per member | +| Zset | one `ZADD key score member` per member (score via `d2string`/fpconv, shortest round-trip) | +| any TTL | trailing `EXPIREAT key ` | + +One-command-per-element makes output large: `test_result.md` records 13 GB of RocksDB exploding to a 188 GB AOF. Deleted and already-expired records are skipped, like the RDB tool. + +## 5. SAVE / BGSAVE / BGREWRITEAOF / LASTSAVE + +Not implemented and not stubbed: none of them appear in the `command_types` map (`src/redis_command.cpp:103+`) or the handler registry, so they fall through to `ERR unknown command` (`src/redis_service.cpp:6049-6057`). The only references are commented-out vendored Redis headers (`include/redis/server.h`). Durability is always-on engine WAL + checkpoint; backups are taken on the kv-store side (DSS snapshot/backup, `data_substrate/docs/07-durability-and-recovery.md`), not via Redis commands. + +## 6. Gotchas & invariants + +- **Exports are checkpoint-lagged.** Both offline tools read the kv store; un-checkpointed committed writes are absent. The cloud path is at least a consistent point-in-time snapshot (manifest branch); the legacy local path is only consistent because the server must be stopped. +- **Encoding round-trips lose representation, not data.** RESTORE flattens every compact encoding (ziplist/listpack/intset/quicklist) into the regular EloqKV deque/flat_hash_map objects; DUMP never re-creates compact encodings. Values survive; memory layout and `OBJECT ENCODING` fidelity do not. +- **Size caps on RESTORE.** Collections > 2^20 entries or LZF strings > 64 MiB inside a Redis-format payload are rejected (`src/redis_rdb_restore.cpp:71-72`) even though such keys can exist in Redis; legacy EloqKV-format (version 0) payloads bypass these decode caps entirely. +- The RDB exporters compute the file CRC with `crc64speed` after calling `crc64speed_init()` (`eloqkv2rdb.cpp:2335`); the server initializes only the big-endian table variant for legacy DUMP verification (`crc64speed_init_big()`, `src/redis_service.cpp:369`) — three CRC entry points (`crc64`, `crc64speed`, `crc64speed_big`) all implement the same Jones CRC-64. + +## 7. Key files + +| File | Role | +|---|---| +| `src/redis_rdb_restore.cpp`, `include/redis_rdb_restore.h` | RDB payload codec (RESTORE decode, DUMP encode) | +| `src/redis_command.cpp:15944-16249, 20005-20140` | DUMP/RESTORE command logic, footer verify, parse | +| `src/tools/eloqkv2rdb/eloqkv2rdb.cpp` | offline RDB exporter (local + rocksdb-cloud) | +| `src/tools/eloqkv2aof/eloqkv2aof.cpp` | offline AOF exporter (local RocksDB only) | +| `crcspeed/`, `fpconv/` | vendored CRC-64 / double-formatting libs | +| `CMakeLists.txt:473-528` | tool build/install wiring per `WITH_DATA_STORE` | diff --git a/docs/README.md b/docs/README.md new file mode 100644 index 00000000..b1c2c9b5 --- /dev/null +++ b/docs/README.md @@ -0,0 +1,25 @@ +# EloqKV Technical Documentation + +Module-by-module design documentation for the EloqKV Redis API layer, written so that an engineer (or an AI model) can understand the design before reading the code. Every doc cites the source files it was derived from — when docs and code disagree, the code wins; please fix the doc in the same change. + +This repo is the **protocol layer only**. The transaction/storage engine lives in the `data_substrate/` submodule and has its own doc set at `data_substrate/docs/` — engine topics (concurrency control, transactions, distribution, durability, storage, log service) are documented there and only cross-referenced here. + +## Reading order + +| Doc | Module | Start here if you're touching… | +|---|---|---| +| [01-architecture-overview.md](01-architecture-overview.md) | Process bootstrap, brpc server, engine registration, config | anything (read first) | +| [02-command-processing.md](02-command-processing.md) | RESP dispatch, connection state, MULTI/EXEC & BEGIN/COMMIT, txm-on-bthread contract, cluster/redirects, repliers | `redis_service.*`, `redis_connection_context.*`, command dispatch | +| [03-data-model.md](03-data-model.md) | EloqKey, the five type objects (string/hash/list/set/zset), TxCommand families, TTL, catalog factory | `redis_*_object.*`, `redis_command.*`, `eloqkv_key.*`, `eloqkv_catalog_factory.*` | +| [04-scripting-pubsub-blocking.md](04-scripting-pubsub-blocking.md) | Lua/EVAL, pub/sub, blocking commands (BLPOP family) | `lua_interpreter.*`, `pub_sub_manager.*` | +| [05-namespaces.md](05-namespaces.md) | Namespace isolation & management, tokens, namespace GC | `include/namespace/`, `src/namespace/` | +| [06-vector-search.md](06-vector-search.md) | Vector indexes (HNSW), vector commands, index durability | `include/vector/`, `src/vector/` | +| [07-persistence-and-tools.md](07-persistence-and-tools.md) | DUMP/RESTORE RDB interop, eloqkv2rdb / eloqkv2aof exporters | `redis_rdb_restore.*`, `src/tools/` | + +Engine-side reading: `data_substrate/docs/README.md` (index), especially `02-threading-model.md` (bthread/TxProcessor contract — required reading before touching anything concurrent) and `05-data-model-and-catalog.md` (the TxObject/TxCommand model EloqKV plugs into). + +## Maintenance rules + +- **Code changes that alter behavior described here must update the corresponding doc in the same PR.** The per-doc "Key files" lists tell you which doc owns which source files. +- Keep docs grounded: cite file paths, prefer invariants and flows over API listings, delete statements you can no longer verify. +- New module → new numbered doc + a row in the table above + a pointer from the repo `CLAUDE.md`.