From f8c9af45e847881c30cbca6f1c3315dbdf20d668 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Fri, 6 Feb 2026 11:58:01 -0500 Subject: [PATCH] chore: audit fixes, graceful shutdown, doc cleanup - add graceful shutdown on SIGINT/SIGTERM with connection draining - fix all rustdoc warnings (escape angle brackets and square brackets in command doc comments across 56 doc strings) - fix clippy range check warnings in test assertions - add fsync after AOF truncate header rewrite for durability - update all crate READMEs to reflect current feature set (hashes, sets, cluster commands, graceful shutdown) - remove stale "WIP" labels from related crate tables - update CLAUDE.md command count from 49 to 62 --- README.md | 1 + crates/ember-cli/README.md | 4 +- crates/ember-cluster/README.md | 2 +- crates/ember-core/README.md | 8 +- crates/ember-core/src/keyspace.rs | 8 +- crates/ember-core/src/shard.rs | 2 +- crates/ember-persistence/README.md | 8 +- crates/ember-persistence/src/aof.rs | 4 +- crates/ember-persistence/src/snapshot.rs | 2 +- crates/ember-protocol/README.md | 22 +++-- crates/ember-protocol/src/command.rs | 114 +++++++++++------------ crates/ember-server/README.md | 8 +- crates/ember-server/src/server.rs | 59 ++++++++---- 13 files changed, 141 insertions(+), 101 deletions(-) diff --git a/README.md b/README.md index f5c16d9b..73ab210d 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ a low-latency, memory-efficient, distributed cache written in Rust. designed to - **lru eviction** — approximate LRU via random sampling when memory pressure hits - **persistence** — append-only file (AOF) and point-in-time snapshots - **pipelining** — multiple commands per read for high throughput +- **graceful shutdown** — drains active connections on SIGINT/SIGTERM before exiting ## quickstart diff --git a/crates/ember-cli/README.md b/crates/ember-cli/README.md index 66351485..4c7e09c2 100644 --- a/crates/ember-cli/README.md +++ b/crates/ember-cli/README.md @@ -2,7 +2,7 @@ interactive command-line client for [ember](https://github.com/kacy/ember). -> this crate is a stub — the CLI is planned for a future phase. +> this crate is a stub — the CLI is planned for phase 5. ## planned features @@ -20,4 +20,4 @@ interactive command-line client for [ember](https://github.com/kacy/ember). | [ember-protocol](../ember-protocol) | RESP3 parsing and command dispatch | | [ember-persistence](../ember-persistence) | AOF, snapshots, and crash recovery | | [ember-server](../ember-server) | TCP server and connection handling | -| [ember-cluster](../ember-cluster) | distributed coordination (WIP) | +| [ember-cluster](../ember-cluster) | distributed coordination | diff --git a/crates/ember-cluster/README.md b/crates/ember-cluster/README.md index 4ba80e2f..e887b57b 100644 --- a/crates/ember-cluster/README.md +++ b/crates/ember-cluster/README.md @@ -74,4 +74,4 @@ the following CLUSTER commands are supported at the protocol layer: | [ember-protocol](../ember-protocol) | RESP3 parsing and command dispatch | | [ember-persistence](../ember-persistence) | AOF, snapshots, and crash recovery | | [ember-server](../ember-server) | TCP server and connection handling | -| [ember-cli](../ember-cli) | interactive command-line client | +| [ember-cli](../ember-cli) | interactive command-line client (planned) | diff --git a/crates/ember-core/README.md b/crates/ember-core/README.md index e0412013..c5949fc7 100644 --- a/crates/ember-core/README.md +++ b/crates/ember-core/README.md @@ -8,8 +8,8 @@ this is the heart of ember — it implements the shared-nothing, shard-per-core - **engine** — routes requests to shards by key hash, supports single-key, multi-key, and broadcast operations - **shard** — the single-threaded event loop per partition: dispatch, AOF recording, expiration ticks, fsync ticks -- **keyspace** — the key-value store itself: strings, lists, sorted sets, TTL, LRU eviction -- **types** — `Value` enum with `String(Bytes)`, `List(VecDeque)`, `SortedSet` (BTreeMap + HashMap dual-index) +- **keyspace** — the key-value store itself: strings, lists, sorted sets, hashes, sets, TTL, LRU eviction +- **types** — `Value` enum with `String(Bytes)`, `List(VecDeque)`, `SortedSet` (BTreeMap + HashMap dual-index), `Hash(HashMap)`, `Set(HashSet)` - **memory** — per-shard memory tracking and entry size estimation - **expiry** — lazy (on access) and active (background sampling) TTL expiration @@ -34,5 +34,5 @@ let response = engine.route("mykey", ShardRequest::Get { | [ember-protocol](../ember-protocol) | RESP3 parsing and command dispatch | | [ember-persistence](../ember-persistence) | AOF, snapshots, and crash recovery | | [ember-server](../ember-server) | TCP server and connection handling | -| [ember-cluster](../ember-cluster) | distributed coordination (WIP) | -| [ember-cli](../ember-cli) | interactive command-line client (WIP) | +| [ember-cluster](../ember-cluster) | distributed coordination | +| [ember-cli](../ember-cli) | interactive command-line client (planned) | diff --git a/crates/ember-core/src/keyspace.rs b/crates/ember-core/src/keyspace.rs index 7988558d..4b3f189e 100644 --- a/crates/ember-core/src/keyspace.rs +++ b/crates/ember-core/src/keyspace.rs @@ -558,7 +558,7 @@ impl Keyspace { /// Scans keys starting from a cursor position. /// /// Returns the next cursor (0 if scan complete) and a batch of keys. - /// The `pattern` argument supports glob-style matching (*, ?, [abc]). + /// The `pattern` argument supports glob-style matching (`*`, `?`, `[abc]`). pub fn scan_keys( &self, cursor: u64, @@ -1777,7 +1777,7 @@ mod tests { Some(Duration::from_secs(100)), ); match ks.ttl("key") { - TtlResult::Seconds(s) => assert!(s >= 98 && s <= 100), + TtlResult::Seconds(s) => assert!((98..=100).contains(&s)), other => panic!("expected Seconds, got {other:?}"), } } @@ -1800,7 +1800,7 @@ mod tests { ks.set("key".into(), Bytes::from("val"), None); assert!(ks.expire("key", 60)); match ks.ttl("key") { - TtlResult::Seconds(s) => assert!(s >= 58 && s <= 60), + TtlResult::Seconds(s) => assert!((58..=60).contains(&s)), other => panic!("expected Seconds, got {other:?}"), } } @@ -2584,7 +2584,7 @@ mod tests { ks.set("n".into(), Bytes::from("5"), Some(Duration::from_secs(60))); ks.incr("n").unwrap(); match ks.ttl("n") { - TtlResult::Seconds(s) => assert!(s >= 58 && s <= 60), + TtlResult::Seconds(s) => assert!((58..=60).contains(&s)), other => panic!("expected TTL preserved, got {other:?}"), } } diff --git a/crates/ember-core/src/shard.rs b/crates/ember-core/src/shard.rs index 300e0975..2b3a090b 100644 --- a/crates/ember-core/src/shard.rs +++ b/crates/ember-core/src/shard.rs @@ -967,7 +967,7 @@ mod tests { let resp = dispatch(&mut ks, &ShardRequest::Ttl { key: "key".into() }); match resp { - ShardResponse::Ttl(TtlResult::Seconds(s)) => assert!(s >= 58 && s <= 60), + ShardResponse::Ttl(TtlResult::Seconds(s)) => assert!((58..=60).contains(&s)), other => panic!("expected Ttl(Seconds), got {other:?}"), } } diff --git a/crates/ember-persistence/README.md b/crates/ember-persistence/README.md index a34f6816..ebf81f93 100644 --- a/crates/ember-persistence/README.md +++ b/crates/ember-persistence/README.md @@ -15,7 +15,9 @@ each shard gets its own persistence files (`shard-{id}.aof` and `shard-{id}.snap **AOF** — `[EAOF magic][version][record...]` where each record is `[tag][payload][crc32]` -**snapshot (v2)** — `[ESNP magic][version][shard_id][entry_count][entries...][footer_crc32]` where entries are type-tagged (string=0, list=1, sorted set=2). v1 snapshots (no type tags) are still readable. +supported record types: SET, DEL, EXPIRE, LPUSH, RPUSH, LPOP, RPOP, ZADD, ZREM, HSET, HDEL, HINCRBY, SADD, SREM + +**snapshot (v2)** — `[ESNP magic][version][shard_id][entry_count][entries...][footer_crc32]` where entries are type-tagged (string=0, list=1, sorted set=2, hash=3, set=4). v1 snapshots (no type tags) are still readable. ## usage @@ -38,5 +40,5 @@ for entry in result.entries { | [emberkv-core](../ember-core) | storage engine, keyspace, sharding | | [ember-protocol](../ember-protocol) | RESP3 parsing and command dispatch | | [ember-server](../ember-server) | TCP server and connection handling | -| [ember-cluster](../ember-cluster) | distributed coordination (WIP) | -| [ember-cli](../ember-cli) | interactive command-line client (WIP) | +| [ember-cluster](../ember-cluster) | distributed coordination | +| [ember-cli](../ember-cli) | interactive command-line client (planned) | diff --git a/crates/ember-persistence/src/aof.rs b/crates/ember-persistence/src/aof.rs index 11aeb62e..4fd0d4b2 100644 --- a/crates/ember-persistence/src/aof.rs +++ b/crates/ember-persistence/src/aof.rs @@ -58,7 +58,7 @@ const TAG_SREM: u8 = 18; /// A single mutation record stored in the AOF. #[derive(Debug, Clone, PartialEq)] pub enum AofRecord { - /// SET key value [expire_ms]. expire_ms is -1 for no expiration. + /// SET key value \[expire_ms\]. expire_ms is -1 for no expiration. Set { key: String, value: Bytes, @@ -444,6 +444,8 @@ impl AofWriter { let mut writer = BufWriter::new(file); format::write_header(&mut writer, format::AOF_MAGIC)?; writer.flush()?; + // ensure the fresh header is durable before we start appending + writer.get_ref().sync_all()?; self.writer = writer; Ok(()) } diff --git a/crates/ember-persistence/src/snapshot.rs b/crates/ember-persistence/src/snapshot.rs index 197cc16e..1b0f8bda 100644 --- a/crates/ember-persistence/src/snapshot.rs +++ b/crates/ember-persistence/src/snapshot.rs @@ -80,7 +80,7 @@ pub struct SnapshotWriter { impl SnapshotWriter { /// Creates a new snapshot writer. The file won't appear at `path` - /// until [`finish`] is called successfully. + /// until [`Self::finish`] is called successfully. pub fn create(path: impl Into, shard_id: u16) -> Result { let final_path = path.into(); let tmp_path = final_path.with_extension("snap.tmp"); diff --git a/crates/ember-protocol/README.md b/crates/ember-protocol/README.md index 7984944a..6185ee70 100644 --- a/crates/ember-protocol/README.md +++ b/crates/ember-protocol/README.md @@ -7,7 +7,7 @@ RESP3 wire protocol implementation for [ember](https://github.com/kacy/ember). h - **parse** — zero-copy RESP3 frame parser that works directly on byte slices, returning `(Frame, bytes_consumed)` for pipelining support - **serialize** — writes frames directly into `BytesMut` with no intermediate allocations - **command** — converts raw frames into typed `Command` enums with argument validation, arity checks, and flag parsing -- **types** — `Frame` enum: `Simple`, `Error`, `Integer`, `Bulk`, `Null`, `Array` +- **types** — `Frame` enum: `Simple`, `Error`, `Integer`, `Bulk`, `Null`, `Array`, `Map` ## quick start @@ -36,13 +36,21 @@ let cmd = Command::from_frame(frame).unwrap(); ## supported commands -strings: `GET`, `SET`, `DEL`, `EXISTS`, `EXPIRE`, `TTL`, `TYPE` +**strings**: `GET`, `SET` (with NX/XX/EX/PX), `INCR`, `DECR`, `MGET`, `MSET` -lists: `LPUSH`, `RPUSH`, `LPOP`, `RPOP`, `LRANGE`, `LLEN` +**lists**: `LPUSH`, `RPUSH`, `LPOP`, `RPOP`, `LRANGE`, `LLEN` -sorted sets: `ZADD` (with NX/XX/GT/LT/CH flags), `ZREM`, `ZSCORE`, `ZRANK`, `ZRANGE` (with WITHSCORES) +**sorted sets**: `ZADD` (with NX/XX/GT/LT/CH flags), `ZREM`, `ZSCORE`, `ZRANK`, `ZRANGE` (with WITHSCORES), `ZCARD` -server: `PING`, `ECHO`, `DBSIZE`, `INFO`, `BGSAVE`, `BGREWRITEAOF` +**hashes**: `HSET`, `HGET`, `HGETALL`, `HDEL`, `HEXISTS`, `HLEN`, `HINCRBY`, `HKEYS`, `HVALS`, `HMGET` + +**sets**: `SADD`, `SREM`, `SMEMBERS`, `SISMEMBER`, `SCARD` + +**keys**: `DEL`, `EXISTS`, `EXPIRE`, `PEXPIRE`, `TTL`, `PTTL`, `PERSIST`, `TYPE`, `SCAN` + +**server**: `PING`, `ECHO`, `DBSIZE`, `INFO`, `BGSAVE`, `BGREWRITEAOF`, `FLUSHDB` + +**cluster**: `CLUSTER INFO`, `NODES`, `SLOTS`, `KEYSLOT`, `MYID`, `MEET`, `ADDSLOTS`, `DELSLOTS`, `SETSLOT`, `FORGET`, `REPLICATE`, `FAILOVER`, `COUNTKEYSINSLOT`, `GETKEYSINSLOT`, `MIGRATE`, `ASKING` ## related crates @@ -51,5 +59,5 @@ server: `PING`, `ECHO`, `DBSIZE`, `INFO`, `BGSAVE`, `BGREWRITEAOF` | [emberkv-core](../ember-core) | storage engine, keyspace, sharding | | [ember-persistence](../ember-persistence) | AOF, snapshots, and crash recovery | | [ember-server](../ember-server) | TCP server and connection handling | -| [ember-cluster](../ember-cluster) | distributed coordination (WIP) | -| [ember-cli](../ember-cli) | interactive command-line client (WIP) | +| [ember-cluster](../ember-cluster) | distributed coordination | +| [ember-cli](../ember-cli) | interactive command-line client (planned) | diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index d665b1f3..c3c1d4b2 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -24,13 +24,13 @@ pub enum Command { /// PING with an optional message. Returns PONG or echoes the message. Ping(Option), - /// ECHO . Returns the message back to the client. + /// ECHO `message`. Returns the message back to the client. Echo(Bytes), - /// GET . Returns the value or nil. + /// GET `key`. Returns the value or nil. Get { key: String }, - /// SET [EX seconds | PX milliseconds] [NX | XX]. + /// SET `key` `value` \[EX seconds | PX milliseconds\] \[NX | XX\]. Set { key: String, value: Bytes, @@ -41,43 +41,43 @@ pub enum Command { xx: bool, }, - /// INCR . Increments the integer value of a key by 1. + /// INCR `key`. Increments the integer value of a key by 1. Incr { key: String }, - /// DECR . Decrements the integer value of a key by 1. + /// DECR `key`. Decrements the integer value of a key by 1. Decr { key: String }, - /// DEL [key ...]. Returns the number of keys removed. + /// DEL `key` \[key ...\]. Returns the number of keys removed. Del { keys: Vec }, - /// EXISTS [key ...]. Returns the number of keys that exist. + /// EXISTS `key` \[key ...\]. Returns the number of keys that exist. Exists { keys: Vec }, - /// MGET [key ...]. Returns the values for all specified keys. + /// MGET `key` \[key ...\]. Returns the values for all specified keys. MGet { keys: Vec }, - /// MSET [key value ...]. Sets multiple key-value pairs. + /// MSET `key` `value` \[key value ...\]. Sets multiple key-value pairs. MSet { pairs: Vec<(String, Bytes)> }, - /// EXPIRE . Sets a TTL on an existing key. + /// EXPIRE `key` `seconds`. Sets a TTL on an existing key. Expire { key: String, seconds: u64 }, - /// TTL . Returns remaining time-to-live in seconds. + /// TTL `key`. Returns remaining time-to-live in seconds. Ttl { key: String }, - /// PERSIST . Removes the expiration from a key. + /// PERSIST `key`. Removes the expiration from a key. Persist { key: String }, - /// PTTL . Returns remaining time-to-live in milliseconds. + /// PTTL `key`. Returns remaining time-to-live in milliseconds. Pttl { key: String }, - /// PEXPIRE . Sets a TTL in milliseconds on an existing key. + /// PEXPIRE `key` `milliseconds`. Sets a TTL in milliseconds on an existing key. Pexpire { key: String, milliseconds: u64 }, /// DBSIZE. Returns the number of keys in the database. DbSize, - /// INFO [section]. Returns server info. Currently only supports "keyspace". + /// INFO \[section\]. Returns server info. Currently only supports "keyspace". Info { section: Option }, /// BGSAVE. Triggers a background snapshot. @@ -89,54 +89,54 @@ pub enum Command { /// FLUSHDB. Removes all keys from the database. FlushDb, - /// SCAN [MATCH pattern] [COUNT count]. Iterates keys. + /// SCAN `cursor` \[MATCH pattern\] \[COUNT count\]. Iterates keys. Scan { cursor: u64, pattern: Option, count: Option, }, - /// LPUSH [value ...]. Pushes values to the head of a list. + /// LPUSH `key` `value` \[value ...\]. Pushes values to the head of a list. LPush { key: String, values: Vec }, - /// RPUSH [value ...]. Pushes values to the tail of a list. + /// RPUSH `key` `value` \[value ...\]. Pushes values to the tail of a list. RPush { key: String, values: Vec }, - /// LPOP . Pops a value from the head of a list. + /// LPOP `key`. Pops a value from the head of a list. LPop { key: String }, - /// RPOP . Pops a value from the tail of a list. + /// RPOP `key`. Pops a value from the tail of a list. RPop { key: String }, - /// LRANGE . Returns a range of elements by index. + /// LRANGE `key` `start` `stop`. Returns a range of elements by index. LRange { key: String, start: i64, stop: i64 }, - /// LLEN . Returns the length of a list. + /// LLEN `key`. Returns the length of a list. LLen { key: String }, - /// TYPE . Returns the type of the value stored at key. + /// TYPE `key`. Returns the type of the value stored at key. Type { key: String }, - /// ZADD [NX|XX] [GT|LT] [CH] [score member ...]. + /// ZADD `key` \[NX|XX\] \[GT|LT\] \[CH\] `score` `member` \[score member ...\]. ZAdd { key: String, flags: ZAddFlags, members: Vec<(f64, String)>, }, - /// ZREM [member ...]. Removes members from a sorted set. + /// ZREM `key` `member` \[member ...\]. Removes members from a sorted set. ZRem { key: String, members: Vec }, - /// ZSCORE . Returns the score of a member. + /// ZSCORE `key` `member`. Returns the score of a member. ZScore { key: String, member: String }, - /// ZRANK . Returns the rank of a member (0-based). + /// ZRANK `key` `member`. Returns the rank of a member (0-based). ZRank { key: String, member: String }, - /// ZCARD . Returns the cardinality (number of members) of a sorted set. + /// ZCARD `key`. Returns the cardinality (number of members) of a sorted set. ZCard { key: String }, - /// ZRANGE [WITHSCORES]. Returns a range by rank. + /// ZRANGE `key` `start` `stop` \[WITHSCORES\]. Returns a range by rank. ZRange { key: String, start: i64, @@ -144,56 +144,56 @@ pub enum Command { with_scores: bool, }, - /// HSET [field value ...]. Sets field-value pairs in a hash. + /// HSET `key` `field` `value` \[field value ...\]. Sets field-value pairs in a hash. HSet { key: String, fields: Vec<(String, Bytes)>, }, - /// HGET . Gets a field's value from a hash. + /// HGET `key` `field`. Gets a field's value from a hash. HGet { key: String, field: String }, - /// HGETALL . Gets all field-value pairs from a hash. + /// HGETALL `key`. Gets all field-value pairs from a hash. HGetAll { key: String }, - /// HDEL [field ...]. Deletes fields from a hash. + /// HDEL `key` `field` \[field ...\]. Deletes fields from a hash. HDel { key: String, fields: Vec }, - /// HEXISTS . Checks if a field exists in a hash. + /// HEXISTS `key` `field`. Checks if a field exists in a hash. HExists { key: String, field: String }, - /// HLEN . Returns the number of fields in a hash. + /// HLEN `key`. Returns the number of fields in a hash. HLen { key: String }, - /// HINCRBY . Increments a hash field's integer value. + /// HINCRBY `key` `field` `increment`. Increments a hash field's integer value. HIncrBy { key: String, field: String, delta: i64, }, - /// HKEYS . Returns all field names in a hash. + /// HKEYS `key`. Returns all field names in a hash. HKeys { key: String }, - /// HVALS . Returns all values in a hash. + /// HVALS `key`. Returns all values in a hash. HVals { key: String }, - /// HMGET [field ...]. Gets multiple field values from a hash. + /// HMGET `key` `field` \[field ...\]. Gets multiple field values from a hash. HMGet { key: String, fields: Vec }, - /// SADD [member ...]. Adds members to a set. + /// SADD `key` `member` \[member ...\]. Adds members to a set. SAdd { key: String, members: Vec }, - /// SREM [member ...]. Removes members from a set. + /// SREM `key` `member` \[member ...\]. Removes members from a set. SRem { key: String, members: Vec }, - /// SMEMBERS . Returns all members of a set. + /// SMEMBERS `key`. Returns all members of a set. SMembers { key: String }, - /// SISMEMBER . Checks if a member exists in a set. + /// SISMEMBER `key` `member`. Checks if a member exists in a set. SIsMember { key: String, member: String }, - /// SCARD . Returns the cardinality (number of members) of a set. + /// SCARD `key`. Returns the cardinality (number of members) of a set. SCard { key: String }, // --- cluster commands --- @@ -206,49 +206,49 @@ pub enum Command { /// CLUSTER SLOTS. Returns the slot distribution across nodes. ClusterSlots, - /// CLUSTER KEYSLOT . Returns the hash slot for a key. + /// CLUSTER KEYSLOT `key`. Returns the hash slot for a key. ClusterKeySlot { key: String }, /// CLUSTER MYID. Returns the node's ID. ClusterMyId, - /// CLUSTER SETSLOT IMPORTING . Mark slot as importing from node. + /// CLUSTER SETSLOT `slot` IMPORTING `node-id`. Mark slot as importing from node. ClusterSetSlotImporting { slot: u16, node_id: String }, - /// CLUSTER SETSLOT MIGRATING . Mark slot as migrating to node. + /// CLUSTER SETSLOT `slot` MIGRATING `node-id`. Mark slot as migrating to node. ClusterSetSlotMigrating { slot: u16, node_id: String }, - /// CLUSTER SETSLOT NODE . Assign slot to node. + /// CLUSTER SETSLOT `slot` NODE `node-id`. Assign slot to node. ClusterSetSlotNode { slot: u16, node_id: String }, - /// CLUSTER SETSLOT STABLE. Clear importing/migrating state. + /// CLUSTER SETSLOT `slot` STABLE. Clear importing/migrating state. ClusterSetSlotStable { slot: u16 }, - /// CLUSTER MEET . Add a node to the cluster. + /// CLUSTER MEET `ip` `port`. Add a node to the cluster. ClusterMeet { ip: String, port: u16 }, - /// CLUSTER ADDSLOTS [slot...]. Assign slots to the local node. + /// CLUSTER ADDSLOTS `slot` \[slot...\]. Assign slots to the local node. ClusterAddSlots { slots: Vec }, - /// CLUSTER DELSLOTS [slot...]. Remove slots from the local node. + /// CLUSTER DELSLOTS `slot` \[slot...\]. Remove slots from the local node. ClusterDelSlots { slots: Vec }, - /// CLUSTER FORGET . Remove a node from the cluster. + /// CLUSTER FORGET `node-id`. Remove a node from the cluster. ClusterForget { node_id: String }, - /// CLUSTER REPLICATE . Make this node a replica of another. + /// CLUSTER REPLICATE `node-id`. Make this node a replica of another. ClusterReplicate { node_id: String }, /// CLUSTER FAILOVER [FORCE|TAKEOVER]. Trigger a manual failover. ClusterFailover { force: bool, takeover: bool }, - /// CLUSTER COUNTKEYSINSLOT . Return the number of keys in a slot. + /// CLUSTER COUNTKEYSINSLOT `slot`. Return the number of keys in a slot. ClusterCountKeysInSlot { slot: u16 }, - /// CLUSTER GETKEYSINSLOT . Return keys in a slot. + /// CLUSTER GETKEYSINSLOT `slot` `count`. Return keys in a slot. ClusterGetKeysInSlot { slot: u16, count: u32 }, - /// MIGRATE [COPY] [REPLACE] [KEYS key...]. + /// MIGRATE `host` `port` `key` `db` `timeout` \[COPY\] \[REPLACE\] \[KEYS key...\]. /// Migrate a key to another node. Migrate { host: String, diff --git a/crates/ember-server/README.md b/crates/ember-server/README.md index 23661151..37e2f44b 100644 --- a/crates/ember-server/README.md +++ b/crates/ember-server/README.md @@ -1,11 +1,11 @@ # ember-server -the main server binary for [ember](https://github.com/kacy/ember). accepts TCP connections, parses RESP3 commands, routes them through the sharded engine, and writes responses back. supports pipelining. +the main server binary for [ember](https://github.com/kacy/ember). accepts TCP connections, parses RESP3 commands, routes them through the sharded engine, and writes responses back. supports pipelining and graceful shutdown. ## what's in here - **main** — CLI arg parsing (host, port, max-memory, eviction policy, persistence config) -- **server** — TCP accept loop with configurable connection limits, spawns a handler task per client +- **server** — TCP accept loop with configurable connection limits, graceful shutdown on SIGINT/SIGTERM, spawns a handler task per client - **connection** — per-connection event loop: read → parse frames → dispatch commands → write responses. handles idle timeouts, buffer limits, and protocol errors - **config** — configuration helpers for byte sizes, eviction policies, fsync policies @@ -31,5 +31,5 @@ compatible with `redis-cli` and any RESP3 client. | [emberkv-core](../ember-core) | storage engine, keyspace, sharding | | [ember-protocol](../ember-protocol) | RESP3 parsing and command dispatch | | [ember-persistence](../ember-persistence) | AOF, snapshots, and crash recovery | -| [ember-cluster](../ember-cluster) | distributed coordination (WIP) | -| [ember-cli](../ember-cli) | interactive command-line client (WIP) | +| [ember-cluster](../ember-cluster) | distributed coordination | +| [ember-cli](../ember-cli) | interactive command-line client (planned) | diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index 8bf78a5a..42bdee1c 100644 --- a/crates/ember-server/src/server.rs +++ b/crates/ember-server/src/server.rs @@ -1,4 +1,7 @@ //! TCP server that accepts client connections and spawns handler tasks. +//! +//! Handles graceful shutdown on SIGINT/SIGTERM: stops accepting new +//! connections and waits for in-flight requests to drain before exiting. use std::net::SocketAddr; use std::sync::Arc; @@ -19,6 +22,9 @@ const DEFAULT_MAX_CONNECTIONS: usize = 10_000; /// hands each incoming connection a cheap clone of the engine handle. /// Limits concurrent connections to `max_connections` — excess clients /// are dropped immediately. +/// +/// On SIGINT or SIGTERM the server stops accepting new connections, +/// waits for existing handlers to finish, then exits cleanly. pub async fn run( addr: SocketAddr, shard_count: usize, @@ -40,26 +46,47 @@ pub async fn run( engine.shard_count() ); + let shutdown = tokio::signal::ctrl_c(); + tokio::pin!(shutdown); + loop { - let (stream, peer) = listener.accept().await?; - - let permit = match semaphore.clone().try_acquire_owned() { - Ok(permit) => permit, - Err(_) => { - warn!("connection limit reached, dropping connection from {peer}"); - drop(stream); - continue; + tokio::select! { + biased; + + _ = &mut shutdown => { + info!("shutdown signal received, draining connections..."); + break; } - }; - let engine = engine.clone(); + result = listener.accept() => { + let (stream, peer) = result?; + + let permit = match semaphore.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + warn!("connection limit reached, dropping connection from {peer}"); + drop(stream); + continue; + } + }; - tokio::spawn(async move { - if let Err(e) = connection::handle(stream, engine).await { - error!("connection error from {peer}: {e}"); + let engine = engine.clone(); + + tokio::spawn(async move { + if let Err(e) = connection::handle(stream, engine).await { + error!("connection error from {peer}: {e}"); + } + // permit is dropped here, releasing the slot + drop(permit); + }); } - // permit is dropped here, releasing the slot - drop(permit); - }); + } } + + // wait for all connection handlers to finish by acquiring all permits + info!("waiting for active connections to close..."); + let _ = semaphore.acquire_many(max_conn as u32).await; + info!("all connections drained, shutting down"); + + Ok(()) }