Merge branch 'cxdb-resiliency-1902202601'

Author: jmccarthy

deploy/docker-compose/.env.example

@@ -66,3 +66,13 @@ PUBLIC_BASE_URL=http://localhost:8080
 # Zstd compression level 1-22 (default: 3)
 # Higher = better compression but slower
 # CXDB_COMPRESSION_LEVEL=3
+
+# Maximum concurrent binary protocol connections (default: 512, 0 = unlimited)
+# CXDB_MAX_CONNECTIONS=512
+
+# Read timeout for idle binary protocol connections in seconds (default: 300)
+# Clients that send no data for this long are disconnected
+# CXDB_CONNECTION_READ_TIMEOUT_SECS=300
+
+# Write timeout for binary protocol connections in seconds (default: 30)
+# CXDB_CONNECTION_WRITE_TIMEOUT_SECS=30

deploy/kubernetes/README.md

@@ -101,6 +101,8 @@ Edit `configmap.yaml` and update:
 - `GOOGLE_ALLOWED_DOMAIN` - Email domain for access control (e.g., "example.com")
 - `CXDB_LOG_LEVEL` - Set to "debug" for verbose logging (optional)
 - `CXDB_ENABLE_METRICS` - Set to "true" to enable Prometheus (optional)
+- `CXDB_MAX_CONNECTIONS` - Max concurrent binary protocol connections (default: 512)
+- `CXDB_CONNECTION_READ_TIMEOUT_SECS` - Idle connection timeout in seconds (default: 300)
 
 Apply the ConfigMap:
 

deploy/kubernetes/configmap.yaml

@@ -21,6 +21,11 @@ data:
   # Storage tuning
   CXDB_MAX_BLOB_SIZE: "10485760"  # 10MB
   CXDB_COMPRESSION_LEVEL: "3"     # Zstd level 1-22
+
+  # Connection limits
+  CXDB_MAX_CONNECTIONS: "512"                  # Max concurrent binary protocol connections
+  CXDB_CONNECTION_READ_TIMEOUT_SECS: "300"     # Idle connection timeout (seconds)
+  CXDB_CONNECTION_WRITE_TIMEOUT_SECS: "30"     # Write timeout (seconds)
 ---
 apiVersion: v1
 kind: ConfigMap

docs/architecture.md

@@ -52,7 +52,7 @@ turn_1 (root, depth=1) → turn_2 (depth=2) → turn_3 (depth=3)
                                            ↘ turn_4 (depth=3, branch)
 ```
 
-**Turn Record** (fixed-size, 104 bytes):
+**Turn Record** (fixed-size, 80 bytes):
 
 ```rust
 TurnRecordV1 {
@@ -251,32 +251,50 @@ graph LR
 
 ## Concurrency Model
 
+**Store-Level RwLock:**
+
+- The `Store` is guarded by a `RwLock`, not a `Mutex`
+- Read operations (GetHead, GetLast, GetBlob, search, list contexts) acquire a shared read lock
+- Write operations (AppendTurn, CtxCreate, CtxFork, PutBlob) acquire an exclusive write lock
+- Multiple readers proceed concurrently; writers are serialized
+
 **TurnID Allocation:**
+
 - Single global atomic counter
 - Linearizable: each turn gets a unique, monotonic ID
 
-**Per-Context Head Updates:**
-- Per-context mutex guards head pointer updates
-- Append operations are serialized per context
-- Different contexts can append concurrently
+**Blob Reads (pread):**
+
+- `BlobStore::get()` uses `pread` (positional read) via a separate read-only file handle
+- This avoids seek/read mutation, allowing `&self` access under a shared read lock
+- Multiple threads can read different blobs concurrently without contention
 
 **Blob Deduplication:**
-- Index sharded by hash prefix (16 shards)
-- Double-checked locking: check index, lock shard, check again, write if missing
 
-**Example Concurrency:**
+- `put_if_absent()` checks the in-memory index first (O(1) hash lookup)
+- If missing, compresses and appends under the write lock
 
-```
-Thread 1: Append to context 1 → acquire lock(ctx1) → write turn → update head → release
-Thread 2: Append to context 2 → acquire lock(ctx2) → write turn → update head → release
-Thread 3: Append to context 1 → wait for lock(ctx1) → write turn → update head → release
-```
+**Connection Limits:**
+
+- Binary protocol connections are capped at `CXDB_MAX_CONNECTIONS` (default: 512)
+- Read/write timeouts prevent stalled clients from holding threads indefinitely
+- Connections beyond the limit are rejected immediately
+
+**SSE Event Bus:**
 
-Blobs are deduplicated safely under contention:
+- Event subscribers use bounded channels (4096 events max per subscriber)
+- `publish()` uses `try_send()`: slow subscribers have events dropped rather than accumulating unbounded memory
+- Disconnected subscribers are removed automatically on the next publish
+
+**Example Concurrency:**
 
 ```
-Thread A: hash=abc → check index (miss) → lock shard(abc) → check again → write → unlock
-Thread B: hash=abc → check index (hit from Thread A) → return existing entry
+Thread 1 (read):  get_last(ctx=1)    │ Concurrent (shared read lock)
+Thread 2 (read):  get_last(ctx=2)    │
+Thread 3 (read):  search("tag=foo")  │
+
+Thread 4 (write): append_turn(ctx=1) │ Exclusive (waits for readers to drain)
+Thread 5 (read):  get_blob(hash)     │ Waits for write to finish
 ```
 
 ## Performance Characteristics
@@ -302,7 +320,7 @@ Thread B: hash=abc → check index (hit from Thread A) → return existing entry
 - **Total: ~0.2ms per turn**
 
 **Storage Efficiency:**
-- Turn record: 104 bytes
+- Turn record: 80 bytes
 - Turn metadata: ~50 bytes (type_id + encoding)
 - Blob overhead: ~50 bytes (header + CRC)
 - Typical 10KB payload compresses to ~3KB (70% reduction)
@@ -339,10 +357,10 @@ Thread B: hash=abc → check index (hit from Thread A) → return existing entry
 - **Fast**: No seeks, pure sequential writes (~500 MB/s on SSD)
 - **Simple recovery**: Scan forward, truncate to last valid CRC
 
-### Why Per-Context Locks?
+### Why RwLock?
 
-- **Scalability**: N contexts can append concurrently
-- **Correctness**: Head updates are linearizable per context
+- **Read scalability**: Multiple readers proceed concurrently (reads dominate in practice)
+- **Correctness**: Writers get exclusive access for head updates and appends
 - **Simple**: No complex MVCC or optimistic concurrency
 
 ## Scalability Limits (v1)
@@ -355,7 +373,7 @@ Thread B: hash=abc → check index (hit from Thread A) → return existing entry
 
 **Not supported in v1:**
 - Distributed/replicated storage
-- Horizontal read scaling (single reader process)
+- Horizontal read scaling (concurrent reads within single process via RwLock, but single-node)
 - Sub-blob chunking for huge payloads (>1MB)
 
 See [Roadmap](https://github.com/strongdm/cxdb/blob/main/ROADMAP.md) for v2 features.

docs/deployment.md

@@ -50,6 +50,9 @@ Internet
 | `CXDB_ENABLE_METRICS` | `false` | Enable Prometheus metrics on :9011 |
 | `CXDB_MAX_BLOB_SIZE` | `10485760` | Max blob size (10MB) |
 | `CXDB_COMPRESSION_LEVEL` | `3` | Zstd compression level (1-22) |
+| `CXDB_MAX_CONNECTIONS` | `512` | Max concurrent binary protocol connections (0 = unlimited) |
+| `CXDB_CONNECTION_READ_TIMEOUT_SECS` | `300` | Idle read timeout per connection (seconds) |
+| `CXDB_CONNECTION_WRITE_TIMEOUT_SECS` | `30` | Write timeout per connection (seconds) |
 
 **Gateway (Go):**
 

docs/storage.md

@@ -97,6 +97,21 @@ ContextHeadRecord {
 }
 ```
 
+## Durability
+
+All write operations use `sync_all()` (fsync) after each file write. This ensures
+data is flushed from the OS page cache to stable storage before the server
+acknowledges the write to the client. The sync order for an append is:
+
+1. `blobs.pack` + `blobs.idx` (blob data and index)
+2. `turns.log` (turn record)
+3. `turns.idx` (turn index)
+4. `turns.meta` (type metadata)
+5. `heads.tbl` (context head update)
+
+A crash at any point leaves files in a recoverable state: CRC checks on startup
+detect and truncate partial writes.
+
 ## Recovery
 
 On startup the store scans logs sequentially. If a trailing record fails CRC or is incomplete,

server/src/blob_store/README.md

@@ -102,14 +102,9 @@ Creates or opens:
 let data = b"Hello, world!";
 let hash = blake3::hash(data);
 
-// Put will compress and deduplicate
-let was_new = store.put(&hash, data)?;
-
-if was_new {
-    println!("New blob stored");
-} else {
-    println!("Blob already exists (deduplicated)");
-}
+// put_if_absent will compress and deduplicate
+let entry = store.put_if_absent(*hash.as_bytes(), data)?;
+println!("Stored: offset={}, raw_len={}", entry.offset, entry.raw_len);
 ```
 
 **Compression:**
@@ -121,13 +116,13 @@ if was_new {
 ```rust
 let hash: [u8; 32] = /* ... */;
 
-match store.get(&hash)? {
-    Some(data) => println!("Found: {} bytes", data.len()),
-    None => println!("Blob not found"),
-}
+let data = store.get(&hash)?;  // Returns Result<Vec<u8>>
+println!("Found: {} bytes", data.len());
 ```
 
-Returns decompressed bytes.
+Returns decompressed bytes or a `NotFound` error. Reads use `pread` (positional
+read) via a separate read-only file handle, so `get()` takes `&self` and
+multiple threads can read concurrently without contention.
 
 ### Checking Existence
 
@@ -147,14 +142,15 @@ The blob store automatically deduplicates identical content:
 let data1 = b"foo";
 let hash1 = blake3::hash(data1);
 
-store.put(&hash1, data1)?;  // Writes to pack file
-store.put(&hash1, data1)?;  // No-op (already exists)
+store.put_if_absent(*hash1.as_bytes(), data1)?;  // Writes to pack file
+store.put_if_absent(*hash1.as_bytes(), data1)?;  // No-op (already exists)
 ```
 
 **Thread safety:**
-- Uses sharded locks (16 shards by hash prefix)
-- Double-checked locking: check index, acquire lock, check again, write if missing
-- Safe under concurrent writes
+
+- Reads (`get`, `contains`, `raw_len`) take `&self` and are safe under concurrent access
+- Writes (`put_if_absent`) take `&mut self` and are serialized by the caller's write lock
+- Deduplication check is O(1) in the in-memory hash index
 
 ## Compression
 
@@ -194,15 +190,19 @@ let hash = blake3::hash(data);  // Hash before compression
 let compressed = zstd::encode(data, 3)?;
 
 // Store with original hash
-store.put(&hash, data)?;  // Handles compression internally
+store.put_if_absent(*hash.as_bytes(), data)?;  // Handles compression internally
 ```
 
 **Why BLAKE3?**
 - Fast: 3-4x faster than SHA-256
 - Secure: 256-bit output, collision-resistant
 - Deterministic: Same input always produces same hash
 
-## Crash Recovery
+## Durability and Crash Recovery
+
+All writes (`put_if_absent`) are followed by `sync_all()` (fsync) on both the
+pack file and the index file. This ensures data is on stable storage before the
+caller is notified of success.
 
 On startup, the store:
 
@@ -212,6 +212,7 @@ On startup, the store:
 4. Rebuilds if necessary
 
 **CRC verification:**
+
 - Each blob record has a CRC-32
 - On read, CRC is verified
 - Corrupted blobs return error (caller must handle)
@@ -247,7 +248,7 @@ use blob_store::{BlobStore, BlobCodec};
 use blake3;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // Open store
+    // Open store (mut needed for writes, reads only need &self)
     let mut store = BlobStore::open(Path::new("./data/blobs"))?;
 
     // Prepare data
@@ -260,13 +261,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Compute hash (before compression)
     let hash = blake3::hash(&msgpack_bytes);
 
-    // Store (will compress internally)
-    let was_new = store.put(&hash, &msgpack_bytes)?;
-    println!("Stored: was_new={}", was_new);
+    // Store (will compress internally, requires &mut self)
+    let _entry = store.put_if_absent(*hash.as_bytes(), &msgpack_bytes)?;
+    println!("Stored blob");
 
-    // Retrieve
-    let retrieved = store.get(&hash)?
-        .expect("Blob should exist");
+    // Retrieve (uses pread, only needs &self)
+    let retrieved = store.get(hash.as_bytes())?;
 
     assert_eq!(retrieved, msgpack_bytes);
     println!("Retrieved: {} bytes", retrieved.len());