fix response and index consistency

command.zig

@@ -34,7 +34,7 @@ pub fn parse(msg: []const u8, allocator: std.mem.Allocator) ?[]const u8 {
         .read => {
             const key = messageIterator.rest();
 
-            const value = storage.read(key) orelse {
+            const value = storage.readAlloc(key, allocator) orelse {
                 return FAILURE_RESPONSE;
             };
 

index.zig

@@ -282,7 +282,7 @@ pub fn getValuesByPrefix(prefix: []const u8, allocator: std.mem.Allocator) []con
     // Phase 2: Read values without tree_mutex to avoid deadlock with write/delete
     const values = allocator.alloc([]const u8, keys.len) catch return "";
     for (keys, 0..) |key, i| {
-        const value = storage.read(key) orelse "";
+        const value = storage.readAlloc(key, allocator) orelse "";
         values[i] = value;
     }
 

socket.zig

@@ -76,20 +76,17 @@ pub fn read(conn: posix.socket_t, buf: []u8) !usize {
 }
 
 pub fn write(conn: posix.socket_t, msg: []const u8) !void {
-    const written = try posix.write(conn, msg);
-    if (written != msg.len) {
-        return error.PartialWrite;
+    var offset: usize = 0;
+    while (offset < msg.len) {
+        const written = try posix.write(conn, msg[offset..]);
+        if (written == 0) return error.ConnectionClosed;
+        offset += written;
     }
 }
 
 pub fn writev(conn: posix.socket_t, iovecs: []const posix.iovec_const) !void {
-    var total: usize = 0;
     for (iovecs) |iov| {
-        total += iov.len;
-    }
-
-    const written = try posix.writev(conn, iovecs);
-    if (written != total) {
-        return error.PartialWrite;
+        const bytes: [*]const u8 = @ptrCast(iov.base);
+        try write(conn, bytes[0..iov.len]);
     }
 }

storage.zig

@@ -116,19 +116,16 @@ pub fn write(key: []const u8, value: []const u8) bool {
     const hash = hashing.hashKey(key);
     const shard_idx = getShardIndex(hash);
 
-    var entry_key: []const u8 = undefined;
-    var entry_value: []const u8 = undefined;
     {
         shards[shard_idx].rwlock.lock();
         defer shards[shard_idx].rwlock.unlock();
 
         const entry = writeVolatile(hash, key, value) orelse return false;
-        entry_key = entry.key;
-        entry_value = entry.value;
+        // Prefix scans must see the index entry before this write becomes
+        // visible to another connection.
+        index.insert(entry.key);
+        persistence.persist('W', entry.key, entry.value);
     }
-
-    index.insert(entry_key);
-    persistence.persist('W', entry_key, entry_value);
     return true;
 }
 
@@ -152,6 +149,24 @@ pub fn read(key: []const u8) ?[]const u8 {
     return null;
 }
 
+pub fn readAlloc(key: []const u8, allocator: std.mem.Allocator) ?[]const u8 {
+    if (!shards_initialized) return null;
+
+    const hash = hashing.hashKey(key);
+    const shard_idx = getShardIndex(hash);
+    shards[shard_idx].rwlock.lockShared();
+    defer shards[shard_idx].rwlock.unlockShared();
+
+    var current = shards[shard_idx].buckets[hash % shards[shard_idx].buckets.len];
+    while (current) |entry| {
+        if (entry.hash == hash and std.mem.eql(u8, entry.key, key)) {
+            return allocator.dupe(u8, entry.value) catch null;
+        }
+        current = entry.next;
+    }
+    return null;
+}
+
 pub fn deleteVolatile(hash: u32, key: []const u8) bool {
     if (!shards_initialized) return false;
     const shard_idx = getShardIndex(hash);