feat: WAIT command and CONFIG SET live-apply (maxmemory, maxmemory-policy)

[kacy]

summary

two features that close important operational gaps before public preview: live memory reconfiguration without a server restart, and the WAIT command for replication-aware write operations.

what changed

config set live-apply

CONFIG SET maxmemory and CONFIG SET maxmemory-policy now take effect immediately across all shards. previously they were stored in the config registry but never applied.

• KeyspaceStats::update_memory_config() — O(1) field update on each shard's keyspace. syncs track_access with the new eviction policy so LRU sampling stays consistent.
• ShardRequest::UpdateMemoryConfig { max_memory, eviction_policy } — new variant in the shard request enum. handled in the shard main loop with no hot-path overhead: only executed when an operator runs CONFIG SET.
• ConfigRegistry::memory_limit() and ConfigRegistry::eviction_policy() — new helpers for reading the live parsed values.
• ServerContext::max_memory_limit: AtomicU64 — replaces the immutable max_memory: Option<usize> for INFO display. reads are lock-free; updated atomically on CONFIG SET.
• CONFIG SET maxmemory validates the memory string (e.g. 256M, 1G) and rejects invalid input before storing.

WAIT command

WAIT numreplicas timeout-ms blocks until numreplicas replicas have acknowledged all writes processed before the WAIT, or until timeout_ms ms elapse. returns the count of replicas that acknowledged in time.

protocol extension:

• MSG_ACK = 6 — new message type in the replication wire protocol. replicas send it after applying each record, carrying their current local offset.

primary side (ReplicationServer):

• ReplicaTracker — new struct tracking write_offset: AtomicU64 (incremented when a record is successfully flushed to the replica's TCP buffer) and per-replica acknowledged offsets in a Mutex<HashMap<u64, u64>>.
• after full-sync, the stream is split via tokio::io::split. a background task reads MSG_ACK frames and calls tracker.update(); the main task drives stream_records().
• replica is registered on connect, removed on disconnect (regardless of error path).

replica side (ReplicationClient):

• after applying each MSG_RECORD, increments a local counter and sends MSG_ACK back to the primary.

WAIT handler:

• fast path: returns immediately when no replicas are connected or the target offset is already reached.
• poll path: 25ms sleep intervals up to the requested timeout. no spin-wait.
• zero hot-path impact: write_offset is only touched inside the replication background task, never by GET/SET processing.

what was tested

• cargo check --workspace and cargo clippy --workspace -- -D warnings both clean
• cargo fmt --check clean
• all 1,404 unit tests pass
• integration suite: 119/120 pass (the 1 failure is pre-existing port-collision flakiness in the cluster test harness — passes in isolation every time)
• updated config_registry_set_immutable_rejected test to reflect that maxmemory is now mutable. added a new test covering the validation path.

design considerations

why AtomicU64 for max_memory_limit instead of updating ctx.max_memory: ctx.max_memory is Option<usize> and is read by multiple places. changing its type would require modifying many call sites. AtomicU64 (0 = unlimited) is a clean, lock-free addition that lets INFO always show the live value.

why 25ms polling for WAIT: tight enough for sub-second replication lag scenarios, not so tight that it burns CPU under heavy load. matches the Redis approach of periodic polling rather than condition variables (which would require more synchronization across the replication boundary).

why Mutex<HashMap> for replica offsets: this lock is only touched by the replication background task (writes) and by WAIT callers (reads). it is never touched on the GET/SET hot path. a DashMap would be heavier for this low-contention use case.