persistence

.gitignore

@@ -1,3 +1,6 @@
 pizzakv
 main
-tools/
+tools/
+.db
+.dbb
+test_nov.js

main.zig

@@ -6,18 +6,20 @@ const fmt = std.fmt;
 
 const socket = @import("socket.zig");
 const command = @import("command.zig");
-const storage = @import("storage.zig");
+const persistence = @import("persistence.zig");
 
-const PORT = 8080;
+const PORT = 8085;
 
 pub fn main() !void {
     const listener = try socket.init(PORT);
     defer posix.close(listener);
 
-    std.debug.print("2025 pizzakv! TCP Listening on port {any}\n<danilo.fragoso@dev>\n---------\n", .{PORT});
+    std.debug.print("2025 pizzakv! TCP Listening on port {any}\n<danilo@fragoso.dev>\n---------\n", .{PORT});
     std.debug.print("Commands:\n\nread key\nwrite key|value\ndelete key\nstatus\n", .{});
     std.debug.print("---------\n", .{});
 
+    try persistence.init();
+
     while (true) {
         var client_address: net.Address = undefined;
         var client_address_len: posix.socklen_t = @sizeOf(net.Address);

makefile

@@ -1,5 +1,8 @@
 default: build
 
+run:
+	zig run main.zig 
+
 build:
 	zig build-exe main.zig -O ReleaseFast --name pizzakv
 

persistence.zig

@@ -0,0 +1,77 @@
+const std = @import("std");
+const storage = @import("storage.zig");
+
+const MAX_PERSISTENCE_SIZE = 10_000_000 * 100;
+var storage_file: ?std.fs.File = null;
+var thread_pool: std.Thread.Pool = undefined;
+var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
+const allocator = arena.allocator();
+
+var mutex: std.Thread.Mutex = .{};
+
+pub fn init() !void {
+    try std.Thread.Pool.init(&thread_pool, .{ .allocator = allocator, .n_jobs = 1024 });
+
+    const cwd = std.fs.cwd();
+    storage_file = cwd.openFile(".db", .{ .mode = .read_write }) catch |err| {
+        if (err == std.fs.File.OpenError.FileNotFound) {
+            std.debug.print("No persisted data found, starting fresh...\n", .{});
+
+            storage_file = cwd.createFile(".db", .{}) catch |ierr| {
+                std.debug.print("Failed to create storage file: {any}\n", .{ierr});
+                return;
+            };
+
+            std.debug.print("Created new storage file .db\n", .{});
+        }
+        return;
+    };
+
+    const storage_data = storage_file.?.readToEndAlloc(allocator, MAX_PERSISTENCE_SIZE) catch |err| {
+        std.debug.print("Failed to read storage file: {any}\n", .{err});
+        return;
+    };
+    defer allocator.free(storage_data);
+    var records = std.mem.splitScalar(u8, storage_data, '\r');
+    var record_count: usize = 0;
+    while (records.next()) |record| {
+        if (record.len == 0) {
+            continue;
+        }
+        record_count += 1;
+        std.debug.print("Restoring record N:{d}\r", .{record_count});
+
+        var kv = std.mem.splitScalar(u8, record, '|');
+        const key = kv.first();
+        const value = kv.rest();
+
+        _ = storage.writeVolatile(key, value);
+    }
+    std.debug.print("Restored {d} records from persistence", .{record_count});
+    return;
+}
+
+pub fn persist(key: []const u8, value: []const u8) void {
+    mutex.lock();
+    const record = std.fmt.allocPrint(allocator, "{s}|{s}\r", .{ key, value }) catch {
+        std.debug.print("Failed to format record for persistence\n", .{});
+        mutex.unlock();
+        return;
+    };
+    mutex.unlock();
+
+    thread_pool.spawn(persistRecord, .{record}) catch |err| {
+        std.debug.print("Failed to spawn persistence job: {any}\n", .{err});
+        return;
+    };
+}
+
+fn persistRecord(record: []const u8) void {
+    mutex.lock();
+    defer mutex.unlock();
+
+    _ = storage_file.?.write(record) catch |err| {
+        std.debug.print("Failed to write record to storage file: {any}\n", .{err});
+        return;
+    };
+}

storage.zig

@@ -1,5 +1,6 @@
 const std = @import("std");
 const hashing = @import("hashing.zig");
+const persistence = @import("persistence.zig");
 
 const MAX_RECORDS = 10_000_000;
 const Entry = struct {
@@ -15,10 +16,7 @@ const allocator = arena.allocator();
 const EMPTY = "";
 var mutex: std.Thread.Mutex = .{};
 
-pub fn write(key: []const u8, value: []const u8) bool {
-    mutex.lock();
-    defer mutex.unlock();
-
+pub fn writeVolatile(key: []const u8, value: []const u8) ?*Entry {
     const hash = hashing.hashKey(key);
     const index = hash % buf.len;
 
@@ -26,22 +24,35 @@ pub fn write(key: []const u8, value: []const u8) bool {
     while (current) |entry| {
         if (std.mem.eql(u8, entry.key, key)) {
             allocator.free(entry.value);
-            entry.value = allocator.dupe(u8, value) catch return false;
-            return true;
+            entry.value = allocator.dupe(u8, value) catch return null;
+            return entry;
         }
 
         current = entry.next;
     }
 
-    const newEntry = allocator.create(Entry) catch return false;
+    const newEntry = allocator.create(Entry) catch return null;
     errdefer allocator.destroy(newEntry);
     newEntry.* = Entry{
-        .key = allocator.dupe(u8, key) catch return false,
-        .value = allocator.dupe(u8, value) catch return false,
+        .key = allocator.dupe(u8, key) catch return null,
+        .value = allocator.dupe(u8, value) catch return null,
         .next = buf[index],
     };
 
     buf[index] = newEntry;
+    return newEntry;
+}
+
+pub fn write(key: []const u8, value: []const u8) bool {
+    mutex.lock();
+    defer mutex.unlock();
+
+    const entry = writeVolatile(key, value);
+    if (entry == null) {
+        return false;
+    }
+
+    persistence.persist(entry.?.key, entry.?.value);
     return true;
 }