From 6d5965e91ad32bf47492f828d7a96a1b69f1636b Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 26 Aug 2026 19:15:00 +0200 Subject: [PATCH] feat(txn): add mvcc transactions with conflict detection --- src/txn.zig | 104 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 src/txn.zig diff --git a/src/txn.zig b/src/txn.zig new file mode 100644 index 0000000..dd0152f --- /dev/null +++ b/src/txn.zig @@ -0,0 +1,104 @@ +const std = @import("std"); +const encode = @import("encode.zig"); +const dbmod = @import("db.zig"); +const WriteBatch = @import("batch.zig").WriteBatch; + +const Allocator = std.mem.Allocator; +const Kind = encode.Kind; +const Db = dbmod.Db; + +const BufferedWrite = struct { + kind: Kind, + value: []u8, +}; + +pub const Transaction = struct { + db: *Db, + allocator: Allocator, + snapshotSeq: u64, + reads: std.StringHashMap(void), + writes: std.StringHashMap(BufferedWrite), + finished: bool, + + pub fn begin(db: *Db) Transaction { + return .{ + .db = db, + .allocator = db.allocator, + .snapshotSeq = if (db.nextSeq == 0) 0 else db.nextSeq - 1, + .reads = std.StringHashMap(void).init(db.allocator), + .writes = std.StringHashMap(BufferedWrite).init(db.allocator), + .finished = false, + }; + } + + pub fn deinit(self: *Transaction) void { + var readKeys = self.reads.keyIterator(); + while (readKeys.next()) |key| self.allocator.free(key.*); + self.reads.deinit(); + + var writeEntries = self.writes.iterator(); + while (writeEntries.next()) |entry| { + self.allocator.free(entry.key_ptr.*); + self.allocator.free(entry.value_ptr.value); + } + self.writes.deinit(); + } + + pub fn get(self: *Transaction, allocator: Allocator, key: []const u8) !?[]u8 { + if (self.writes.get(key)) |buffered| { + if (buffered.kind == .tombstone) return null; + return try allocator.dupe(u8, buffered.value); + } + try self.recordRead(key); + return self.db.getAsOf(allocator, key, self.snapshotSeq); + } + + pub fn put(self: *Transaction, key: []const u8, value: []const u8) !void { + try self.stage(.value, key, value); + } + + pub fn delete(self: *Transaction, key: []const u8) !void { + try self.stage(.tombstone, key, &.{}); + } + + fn stage(self: *Transaction, kind: Kind, key: []const u8, value: []const u8) !void { + const valueCopy = try self.allocator.dupe(u8, value); + errdefer self.allocator.free(valueCopy); + const gop = try self.writes.getOrPut(key); + if (gop.found_existing) { + self.allocator.free(gop.value_ptr.value); + gop.value_ptr.* = .{ .kind = kind, .value = valueCopy }; + return; + } + gop.key_ptr.* = try self.allocator.dupe(u8, key); + gop.value_ptr.* = .{ .kind = kind, .value = valueCopy }; + } + + fn recordRead(self: *Transaction, key: []const u8) !void { + if (self.reads.contains(key)) return; + const keyCopy = try self.allocator.dupe(u8, key); + errdefer self.allocator.free(keyCopy); + try self.reads.put(keyCopy, {}); + } + + pub fn commit(self: *Transaction) !void { + var readKeys = self.reads.keyIterator(); + while (readKeys.next()) |key| { + if (self.db.latestSeq(key.*)) |seq| { + if (seq > self.snapshotSeq) return error.Conflict; + } + } + + var batch = WriteBatch.init(self.allocator); + defer batch.deinit(); + var writeEntries = self.writes.iterator(); + while (writeEntries.next()) |entry| { + switch (entry.value_ptr.kind) { + .value => try batch.put(entry.key_ptr.*, entry.value_ptr.value), + .tombstone => try batch.delete(entry.key_ptr.*), + } + } + try self.db.apply(&batch); + self.finished = true; + } +};