tickets

59.5 KB

1. // mpackdb DB engine — Zig port of mpackdb v1.0.7 (MPackDB.js / IndexManager.js)
2. // semantics. File-level compatible with the JS implementation (D12):
3. //
4. // <base>.mpack append-only records: [4-byte LE total size][msgpack map]
5. // <base>.meta.json { nextId?, deleted:[offsets], version, schema? }
6. // <base>.<field>.txt sorted index lines "key,offset,length\n"
7. // <base>.idxstate.json { coveredBytes }
8. // <base>.lock lock file containing the holder's pid (O_EXCL protocol,
9. // stale takeover after staleLockTimeout ms)
10. //
11. // Divergences from the JS implementation (documented in the 066 report):
12. // - the in-memory index is the COMPLETE entry set (JS keeps on-disk + delta);
13. // persist() writes the full set instead of read-merge-write. Sequential
14. // cross-process use (close one side, open the other) behaves identically.
15. // - string index keys sort bytewise (JS: String.localeCompare, ICU collation).
16. // Equality lookups are unaffected; range ordering can differ for
17. // mixed-case/accented keys.
18. // - null field values are not indexed (JS indexes them under the key "null").
19. // - external data-file replacement while open (inode change) is not detected.
21. const std = @import("std");
22. const linux = std.os.linux;
23. const msgpack = @import("msgpack.zig");
25. pub const PkType = enum(u8) { number = 0, uuid = 1, string = 2 };
26. pub const IdxType = enum(u8) { lexical = 0, numeric = 1 };
28. pub const Error = error{
29. NoDbFile,
30. NoPrimaryKey,
31. /// mission 077 (074 GAP 10): a query named a field with no index. There is
32. /// no index to binary-search and mpackdb does not silently full-scan, so
33. /// the answer used to be an empty result — indistinguishable from "no such
34. /// record". It is a mistake, and it says so now.
35. NoSuchIndex,
36. DuplicateKey,
37. /// ticket #29: update()'s record must carry the primary key. Without it
38. /// the old record was deleted, the new one written without a key, and the
39. /// table left corrupt (fetch null, a unique index still finding it).
40. RecordLacksPrimaryKey,
41. LockTimeout,
42. CorruptRecord,
43. IoError,
44. OutOfMemory,
45. InvalidFormat,
46. Truncated,
47. MapTooLarge,
48. };
50. pub const Key = union(enum) {
51. num: f64,
52. str: []const u8,
53. };
55. pub const Entry = struct {
56. key: Key,
57. off: u64,
58. len: u64,
59. };
61. const FieldIndex = struct {
62. typ: IdxType,
63. entries: std.ArrayList(Entry) = .empty, // sorted by (key, off)
64. };
66. pub const Loc = struct { off: u64, len: u64 };
68. pub const Record = struct {
69. value: msgpack.Value,
70. off: u64,
71. len: u64,
72. };
74. // =========================================================================
75. // Low-level file IO (raw linux syscalls — plugins avoid std.Io plumbing)
76. // =========================================================================
78. const fio = struct {
79. fn openz(alloc: std.mem.Allocator, path: []const u8, flags: linux.O, mode: u32) ?i32 {
80. const path_z = alloc.dupeZ(u8, path) catch return null;
81. defer alloc.free(path_z);
82. const rc = linux.open(path_z, flags, mode);
83. const fd: i32 = @bitCast(@as(u32, @truncate(rc)));
84. if (fd < 0) return null;
85. return fd;
86. }
88. fn openRead(alloc: std.mem.Allocator, path: []const u8) ?i32 {
89. return openz(alloc, path, .{}, 0);
90. }
92. fn openAppend(alloc: std.mem.Allocator, path: []const u8) ?i32 {
93. return openz(alloc, path, .{ .ACCMODE = .WRONLY, .CREAT = true, .APPEND = true }, 0o644);
94. }
96. fn openTrunc(alloc: std.mem.Allocator, path: []const u8) ?i32 {
97. return openz(alloc, path, .{ .ACCMODE = .WRONLY, .CREAT = true, .TRUNC = true }, 0o644);
98. }
100. /// O_CREAT|O_EXCL — returns null when the file already exists.
101. fn openExcl(alloc: std.mem.Allocator, path: []const u8) ?i32 {
102. return openz(alloc, path, .{ .ACCMODE = .WRONLY, .CREAT = true, .EXCL = true }, 0o644);
103. }
105. fn close(fd: i32) void {
106. _ = linux.close(fd);
107. }
109. fn writeAll(fd: i32, bytes: []const u8) bool {
110. var written: usize = 0;
111. while (written < bytes.len) {
112. const rc = linux.write(fd, bytes.ptr + written, bytes.len - written);
113. if (@as(isize, @bitCast(rc)) <= 0) return false;
114. written += rc;
115. }
116. return true;
117. }
119. fn pread(fd: i32, buf: []u8, offset: u64) ?usize {
120. var got: usize = 0;
121. while (got < buf.len) {
122. const rc = linux.pread(fd, buf.ptr + got, buf.len - got, @intCast(offset + got));
123. const n: isize = @bitCast(rc);
124. if (n < 0) return null;
125. if (n == 0) break;
126. got += @intCast(n);
127. }
128. return got;
129. }
131. fn readAll(alloc: std.mem.Allocator, path: []const u8) ?[]u8 {
132. const fd = openRead(alloc, path) orelse return null;
133. defer close(fd);
134. var content: std.ArrayList(u8) = .empty;
135. var buf: [65536]u8 = undefined;
136. while (true) {
137. const rc = linux.read(fd, &buf, buf.len);
138. if (@as(isize, @bitCast(rc)) <= 0) break;
139. content.appendSlice(alloc, buf[0..rc]) catch return null;
140. }
141. return content.toOwnedSlice(alloc) catch null;
142. }
144. fn fileSize(alloc: std.mem.Allocator, path: []const u8) ?u64 {
145. const path_z = alloc.dupeZ(u8, path) catch return null;
146. defer alloc.free(path_z);
147. var stx: linux.Statx = undefined;
148. const rc = linux.statx(linux.AT.FDCWD, path_z, 0, linux.STATX.BASIC_STATS, &stx);
149. if (rc != 0) return null;
150. return stx.size;
151. }
153. fn mtimeMs(alloc: std.mem.Allocator, path: []const u8) ?i64 {
154. const path_z = alloc.dupeZ(u8, path) catch return null;
155. defer alloc.free(path_z);
156. var stx: linux.Statx = undefined;
157. const rc = linux.statx(linux.AT.FDCWD, path_z, 0, linux.STATX.BASIC_STATS, &stx);
158. if (rc != 0) return null;
159. return @as(i64, stx.mtime.sec) * 1000 + @divTrunc(@as(i64, stx.mtime.nsec), 1_000_000);
160. }
162. fn rename(alloc: std.mem.Allocator, old_path: []const u8, new_path: []const u8) bool {
163. const old_z = alloc.dupeZ(u8, old_path) catch return false;
164. defer alloc.free(old_z);
165. const new_z = alloc.dupeZ(u8, new_path) catch return false;
166. defer alloc.free(new_z);
167. return linux.renameat(linux.AT.FDCWD, old_z, linux.AT.FDCWD, new_z) == 0;
168. }
170. fn unlink(alloc: std.mem.Allocator, path: []const u8) bool {
171. const path_z = alloc.dupeZ(u8, path) catch return false;
172. defer alloc.free(path_z);
173. return linux.unlinkat(linux.AT.FDCWD, path_z, 0) == 0;
174. }
176. fn mkdirAll(alloc: std.mem.Allocator, dir_path: []const u8) void {
177. if (dir_path.len == 0 or std.mem.eql(u8, dir_path, ".")) return;
178. var i: usize = 1;
179. while (i <= dir_path.len) : (i += 1) {
180. if (i == dir_path.len or dir_path[i] == '/') {
181. const part = alloc.dupeZ(u8, dir_path[0..i]) catch return;
182. defer alloc.free(part);
183. _ = linux.mkdirat(linux.AT.FDCWD, part, 0o755);
184. }
185. }
186. }
188. fn sleepMs(ms: u64) void {
189. var req: linux.timespec = .{ .sec = @intCast(ms / 1000), .nsec = @intCast((ms % 1000) * 1_000_000) };
190. var rem: linux.timespec = undefined;
191. _ = linux.nanosleep(&req, &rem);
192. }
194. fn nowMs() i64 {
195. var ts: linux.timespec = undefined;
196. _ = linux.clock_gettime(linux.CLOCK.REALTIME, &ts);
197. return @as(i64, ts.sec) * 1000 + @divTrunc(@as(i64, ts.nsec), 1_000_000);
198. }
199. };
201. // =========================================================================
202. // JS-compatible helpers
203. // =========================================================================
205. /// Format an f64 the way JS String(n) does for the values mpackdb produces:
206. /// integral values print without a decimal point. (Exotic floats may diverge
207. /// from V8's shortest-round-trip formatting; index keys are ints/strings in
208. /// practice.)
209. pub fn jsNumFmt(buf: []u8, v: f64) []const u8 {
210. if (v == @floor(v) and @abs(v) <= 9007199254740992.0) {
211. const i: i64 = @intFromFloat(v);
212. return std.fmt.bufPrint(buf, "{d}", .{i}) catch buf[0..0];
213. }
214. return std.fmt.bufPrint(buf, "{d}", .{v}) catch buf[0..0];
215. }
217. /// JS parseInt(s, 10) semantics: optional sign, leading digits, ignore rest.
218. /// Returns null for NaN.
219. fn jsParseInt(s: []const u8) ?f64 {
220. var i: usize = 0;
221. while (i < s.len and (s[i] == ' ' or s[i] == '\t')) i += 1;
222. var sign: f64 = 1;
223. if (i < s.len and (s[i] == '+' or s[i] == '-')) {
224. if (s[i] == '-') sign = -1;
225. i += 1;
226. }
227. var got = false;
228. var v: f64 = 0;
229. while (i < s.len and s[i] >= '0' and s[i] <= '9') : (i += 1) {
230. v = v * 10 + @as(f64, @floatFromInt(s[i] - '0'));
231. got = true;
232. }
233. if (!got) return null;
234. return sign * v;
235. }
237. /// _compareKeys: numbers numerically; otherwise stringified byte compare
238. /// (JS uses localeCompare — see divergence note in the header).
239. pub fn cmpKeys(a: Key, b: Key) i32 {
240. if (a == .num and b == .num) {
241. if (a.num < b.num) return -1;
242. if (a.num > b.num) return 1;
243. return 0;
244. }
245. var buf_a: [32]u8 = undefined;
246. var buf_b: [32]u8 = undefined;
247. const sa = if (a == .str) a.str else jsNumFmt(&buf_a, a.num);
248. const sb = if (b == .str) b.str else jsNumFmt(&buf_b, b.num);
249. return switch (std.mem.order(u8, sa, sb)) {
250. .lt => -1,
251. .eq => 0,
252. .gt => 1,
253. };
254. }
256. fn entryLess(_: void, a: Entry, b: Entry) bool {
257. const c = cmpKeys(a.key, b.key);
258. if (c != 0) return c < 0;
259. return a.off < b.off;
260. }
262. /// mpack.js uuid(): 9-char base36 ms timestamp (padStart '0') + 3 base36 chars.
263. pub fn genUuid(buf: *[12]u8) []const u8 {
264. const digits = "0123456789abcdefghijklmnopqrstuvwxyz";
265. const t: u64 = @intCast(fio.nowMs());
266. var tmp: [16]u8 = undefined;
267. var n: usize = 0;
268. var v = t;
269. while (v > 0) : (v /= 36) {
270. tmp[n] = digits[@intCast(v % 36)];
271. n += 1;
272. }
273. var i: usize = 0;
274. while (i < 9) : (i += 1) {
275. buf[8 - i] = if (i < n) tmp[i] else '0';
276. }
277. if (!rng_init) {
278. var ts: linux.timespec = undefined;
279. _ = linux.clock_gettime(linux.CLOCK.MONOTONIC, &ts);
280. rng = std.Random.DefaultPrng.init(@bitCast(@as(i64, ts.sec) *% 1_000_000_000 +% ts.nsec));
281. rng_init = true;
282. }
283. var r = rng.random();
284. for (0..3) |j| {
285. buf[9 + j] = digits[r.uintLessThan(u8, 36)];
286. }
287. return buf[0..12];
288. }
290. var rng: std.Random.DefaultPrng = .{ .s = undefined };
291. var rng_init: bool = false;
293. // =========================================================================
294. // The database
295. // =========================================================================
297. pub const OpenOptions = struct {
298. primary_key: ?[]const u8 = null, // with */@/! prefixes, like the JS API
299. indexes: []const []const u8 = &.{}, // with prefixes
300. compact: bool = true,
301. stale_lock_timeout_ms: i64 = 30000,
302. debug: bool = false,
303. };
305. pub const Db = struct {
306. arena_state: std.heap.ArenaAllocator,
307. alloc: std.mem.Allocator, // arena — freed wholesale on close
309. // paths
310. data_path: []const u8,
311. meta_path: []const u8,
312. lock_path: []const u8,
313. idxstate_path: []const u8,
315. // schema
316. pk: ?[]const u8 = null,
317. pk_type: PkType = .string,
318. index_fields: std.ArrayList([]const u8) = .empty, // pk first (when set), like JS
319. unique_fields: std.ArrayList([]const u8) = .empty,
320. idx: std.StringArrayHashMapUnmanaged(FieldIndex) = .empty,
322. // meta
323. next_id: f64 = 0,
324. deleted: std.ArrayList(u64) = .empty,
325. version: u64 = 0,
327. covered_bytes: u64 = 0,
328. compact_on_open: bool = true,
329. stale_lock_timeout_ms: i64 = 30000,
330. debug: bool = false,
332. // last DuplicateKey detail for the plugin surface
333. last_error_buf: [256]u8 = undefined,
334. last_error: []const u8 = "",
336. pub fn open(gpa: std.mem.Allocator, db_file: []const u8, opts: OpenOptions) Error!*Db {
337. if (db_file.len == 0) return Error.NoDbFile;
339. const self = gpa.create(Db) catch return Error.OutOfMemory;
340. self.* = .{
341. .arena_state = std.heap.ArenaAllocator.init(gpa),
342. .alloc = undefined,
343. .data_path = undefined,
344. .meta_path = undefined,
345. .lock_path = undefined,
346. .idxstate_path = undefined,
347. };
348. self.alloc = self.arena_state.allocator();
349. errdefer {
350. self.arena_state.deinit();
351. gpa.destroy(self);
352. }
353. const a = self.alloc;
355. self.compact_on_open = opts.compact;
356. self.stale_lock_timeout_ms = opts.stale_lock_timeout_ms;
357. self.debug = opts.debug;
359. // dirname / basename (extension stripped, like JS basename(f, extname(f)))
360. const dir = std.fs.path.dirname(db_file) orelse ".";
361. var base = std.fs.path.basename(db_file);
362. if (std.mem.lastIndexOfScalar(u8, base, '.')) |dot| {
363. if (dot > 0) base = base[0..dot];
364. }
365. fio.mkdirAll(a, dir);
367. self.data_path = std.fmt.allocPrint(a, "{s}/{s}.mpack", .{ dir, base }) catch return Error.OutOfMemory;
368. self.meta_path = std.fmt.allocPrint(a, "{s}/{s}.meta.json", .{ dir, base }) catch return Error.OutOfMemory;
369. self.lock_path = std.fmt.allocPrint(a, "{s}/{s}.lock", .{ dir, base }) catch return Error.OutOfMemory;
370. self.idxstate_path = std.fmt.allocPrint(a, "{s}/{s}.idxstate.json", .{ dir, base }) catch return Error.OutOfMemory;
372. // ---- parse primary key ----
373. if (opts.primary_key) |pk_raw| {
374. if (pk_raw.len > 0) {
375. if (pk_raw[0] == '*') {
376. self.pk = a.dupe(u8, pk_raw[1..]) catch return Error.OutOfMemory;
377. self.pk_type = .number;
378. } else if (pk_raw[0] == '@') {
379. self.pk = a.dupe(u8, pk_raw[1..]) catch return Error.OutOfMemory;
380. self.pk_type = .uuid;
381. } else {
382. self.pk = a.dupe(u8, pk_raw) catch return Error.OutOfMemory;
383. self.pk_type = .string;
384. }
385. }
386. }
387. if (self.pk) |p| self.unique_fields.append(a, p) catch return Error.OutOfMemory;
389. // ---- parse indexes (pk first, then configured; dedupe) ----
390. if (self.pk) |p| {
391. self.index_fields.append(a, p) catch return Error.OutOfMemory;
392. const t: IdxType = if (self.pk_type == .number) .numeric else .lexical;
393. self.idx.put(a, p, .{ .typ = t }) catch return Error.OutOfMemory;
394. }
395. for (opts.indexes) |raw| {
396. var clean = raw;
397. var is_unique = false;
398. if (clean.len > 0 and clean[0] == '!') {
399. is_unique = true;
400. clean = clean[1..];
401. }
402. var typ: IdxType = .lexical;
403. if (clean.len > 0 and clean[0] == '*') {
404. typ = .numeric;
405. clean = clean[1..];
406. } else if (clean.len > 0 and clean[0] == '@') {
407. clean = clean[1..];
408. }
409. if (self.idx.contains(clean)) continue;
410. const owned = a.dupe(u8, clean) catch return Error.OutOfMemory;
411. self.index_fields.append(a, owned) catch return Error.OutOfMemory;
412. self.idx.put(a, owned, .{ .typ = typ }) catch return Error.OutOfMemory;
413. if (is_unique) self.unique_fields.append(a, owned) catch return Error.OutOfMemory;
414. }
416. // ---- load meta ----
417. self.loadMeta();
419. // ---- compact on init (JS default; also creates an empty data file) ----
420. // AN OPEN WITH NOTHING TO CHANGE WRITES NOTHING (ticket #21): a store
421. // with no tombstones has nothing to compact, and rewriting it anyway
422. // made opening a backup change it.
423. var did_compact = false;
424. const nothing_to_compact = self.deleted.items.len == 0 and
425. fio.fileSize(self.alloc, self.data_path) != null;
426. if (self.compact_on_open and !nothing_to_compact) {
427. try self.acquireLock();
428. const cr = self.compactLocked();
429. self.releaseLock();
430. try cr;
431. did_compact = true;
432. }
434. // ---- persist schema into meta (JS: init-schema under lock).
435. // compactLocked already wrote meta (with schema); skip the extra bump.
436. if (!did_compact and (self.pk != null or self.index_fields.items.len > 0) and
437. !self.metaOnDiskIsCurrent())
438. {
439. try self.acquireLock();
440. const mr = self.persistMetaLocked();
441. self.releaseLock();
442. try mr;
443. }
445. // ---- indexes (JS: (re)build under the lock; force after compaction) ----
446. if (self.index_fields.items.len > 0) {
447. try self.acquireLock();
448. const ir = self.initIndexes(did_compact);
449. self.releaseLock();
450. try ir;
451. }
453. return self;
454. }
456. pub fn close(self: *Db) void {
457. // persist indexes + idxstate (JS: IndexManager.close → persist under lock)
458. if (self.index_fields.items.len > 0) {
459. if (self.acquireLock()) {
460. self.persistIndexesLocked() catch {};
461. self.releaseLock();
462. } else |_| {}
463. }
464. const gpa = self.arena_state.child_allocator;
465. self.arena_state.deinit();
466. gpa.destroy(self);
467. }
469. fn dbg(self: *Db, comptime fmt: []const u8, args: anytype) void {
470. if (self.debug) std.debug.print("[mpackdb] " ++ fmt ++ "\n", args);
471. }
473. // =====================================================================
474. // Locking (JS _acquireFileLock protocol)
475. // =====================================================================
477. fn acquireLock(self: *Db) Error!void {
478. var retries: u32 = 0;
479. while (true) {
480. if (fio.openExcl(self.alloc, self.lock_path)) |fd| {
481. var pid_buf: [16]u8 = undefined;
482. const pid_s = std.fmt.bufPrint(&pid_buf, "{d}", .{linux.getpid()}) catch "0";
483. _ = fio.writeAll(fd, pid_s);
484. fio.close(fd);
485. return;
486. }
487. // stale lock takeover
488. if (self.stale_lock_timeout_ms > 0) {
489. if (fio.mtimeMs(self.alloc, self.lock_path)) |mt| {
490. if (fio.nowMs() - mt > self.stale_lock_timeout_ms) {
491. _ = fio.unlink(self.alloc, self.lock_path);
492. continue;
493. }
494. }
495. }
496. if (retries > 480) return Error.LockTimeout; // ~12s at 25ms
497. fio.sleepMs(25);
498. retries += 1;
499. }
500. }
502. fn releaseLock(self: *Db) void {
503. _ = fio.unlink(self.alloc, self.lock_path);
504. }
506. // =====================================================================
507. // Meta (.meta.json)
508. // =====================================================================
510. fn appendFmt(self: *Db, list: *std.ArrayList(u8), comptime fmt: []const u8, args: anytype) Error!void {
511. const s = std.fmt.allocPrint(self.alloc, fmt, args) catch return Error.OutOfMemory;
512. defer self.alloc.free(s);
513. list.appendSlice(self.alloc, s) catch return Error.OutOfMemory;
514. }
516. fn loadMeta(self: *Db) void {
517. const content = fio.readAll(self.alloc, self.meta_path) orelse return;
518. self.applyMetaJson(content);
519. }
521. fn applyMetaJson(self: *Db, content: []const u8) void {
522. const parsed = std.json.parseFromSlice(std.json.Value, self.alloc, content, .{}) catch return;
523. defer parsed.deinit();
524. if (parsed.value != .object) return;
525. const obj = parsed.value.object;
526. self.next_id = 0;
527. self.deleted.clearRetainingCapacity();
528. self.version = 0;
529. if (obj.get("nextId")) |v| {
530. self.next_id = switch (v) {
531. .integer => |i| @floatFromInt(i),
532. .float => |f| f,
533. else => 0,
534. };
535. }
536. if (obj.get("version")) |v| {
537. if (v == .integer) self.version = @intCast(@max(v.integer, 0));
538. }
539. if (obj.get("deleted")) |v| {
540. if (v == .array) {
541. for (v.array.items) |it| {
542. const off: u64 = switch (it) {
543. .integer => |i| @intCast(@max(i, 0)),
544. .float => |f| @intFromFloat(@max(f, 0)),
545. else => continue,
546. };
547. self.deleted.append(self.alloc, off) catch {};
548. }
549. }
550. }
551. }
553. /// Pick up other processes' persisted state (JS refresh()): adopt disk meta
554. /// when its version is newer; index appended tail records.
555. pub fn refresh(self: *Db) void {
556. if (fio.readAll(self.alloc, self.meta_path)) |content| {
557. defer self.alloc.free(content);
558. const parsed = std.json.parseFromSlice(std.json.Value, self.alloc, content, .{}) catch return;
559. defer parsed.deinit();
560. if (parsed.value == .object) {
561. var disk_version: u64 = 0;
562. if (parsed.value.object.get("version")) |v| {
563. if (v == .integer) disk_version = @intCast(@max(v.integer, 0));
564. }
565. if (disk_version > self.version) self.applyMetaJson(content);
566. }
567. }
568. // catchUp: index records another process appended
569. if (self.index_fields.items.len > 0) {
570. const size = fio.fileSize(self.alloc, self.data_path) orelse 0;
571. if (size > self.covered_bytes) self.catchUp(size);
572. }
573. }
575. fn persistMetaLocked(self: *Db) Error!void {
576. self.version += 1;
577. var out: std.ArrayList(u8) = .empty;
578. defer out.deinit(self.alloc);
579. try self.renderMeta(&out);
581. // atomic tmp + rename (JS: `${metaPath}.${pid}.tmp`)
582. var tmp_buf: [512]u8 = undefined;
583. const tmp_path = std.fmt.bufPrint(&tmp_buf, "{s}.{d}.tmp", .{ self.meta_path, linux.getpid() }) catch return Error.IoError;
584. const fd = fio.openTrunc(self.alloc, tmp_path) orelse return Error.IoError;
585. const ok = fio.writeAll(fd, out.items);
586. fio.close(fd);
587. if (!ok) return Error.IoError;
588. if (!fio.rename(self.alloc, tmp_path, self.meta_path)) return Error.IoError;
589. }
591. /// Does .meta.json already say, byte for byte, what this handle would write
592. /// at its current version? Then an open has nothing to persist (ticket #21).
593. fn metaOnDiskIsCurrent(self: *Db) bool {
594. const disk = fio.readAll(self.alloc, self.meta_path) orelse return false;
595. defer self.alloc.free(disk);
596. var out: std.ArrayList(u8) = .empty;
597. defer out.deinit(self.alloc);
598. self.renderMeta(&out) catch return false;
599. return std.mem.eql(u8, disk, out.items);
600. }
602. fn renderMeta(self: *Db, out: *std.ArrayList(u8)) Error!void {
603. try self.appendFmt(out, "{{", .{});
604. if (self.pk != null and self.pk_type == .number) {
605. var nbuf: [32]u8 = undefined;
606. try self.appendFmt(out, "\"nextId\":{s},", .{jsNumFmt(&nbuf, self.next_id)});
607. }
608. try self.appendFmt(out, "\"deleted\":[", .{});
609. for (self.deleted.items, 0..) |off, i| {
610. if (i > 0) try self.appendFmt(out, ",", .{});
611. try self.appendFmt(out, "{d}", .{off});
612. }
613. try self.appendFmt(out, "],\"version\":{d}", .{self.version});
614. if (self.pk != null or self.index_fields.items.len > 0) {
615. try self.appendFmt(out, ",\"schema\":{{", .{});
616. var first = true;
617. if (self.pk) |p| {
618. const prefix: []const u8 = switch (self.pk_type) {
619. .number => "*",
620. .uuid => "@",
621. .string => "",
622. };
623. try self.appendFmt(out, "\"primaryKey\":\"{s}{s}\"", .{ prefix, p });
624. first = false;
625. }
626. if (!first) try self.appendFmt(out, ",", .{});
627. try self.appendFmt(out, "\"indexes\":[", .{});
628. var n: usize = 0;
629. for (self.index_fields.items) |field| {
630. if (self.pk != null and std.mem.eql(u8, field, self.pk.?)) continue;
631. if (n > 0) try self.appendFmt(out, ",", .{});
632. const uni = for (self.unique_fields.items) |u| {
633. if (std.mem.eql(u8, u, field)) break true;
634. } else false;
635. const numeric = (self.idx.get(field) orelse FieldIndex{ .typ = .lexical }).typ == .numeric;
636. try self.appendFmt(out, "\"{s}{s}{s}\"", .{
637. if (uni) "!" else "",
638. if (numeric) "*" else "",
639. field,
640. });
641. n += 1;
642. }
643. try self.appendFmt(out, "]}}", .{});
644. }
645. try self.appendFmt(out, "}}", .{});
646. }
648. // =====================================================================
649. // Data file scanning
650. // =====================================================================
652. fn deletedSet(self: *Db, alloc: std.mem.Allocator) std.AutoHashMapUnmanaged(u64, void) {
653. var set: std.AutoHashMapUnmanaged(u64, void) = .empty;
654. for (self.deleted.items) |off| set.put(alloc, off, {}) catch {};
655. return set;
656. }
658. /// Sequential scan yielding all live records. Caller supplies an arena for
659. /// decoded values. Used for rebuilds, compaction and unindexed finds.
660. pub const Scanner = struct {
661. db: *Db,
662. fd: i32 = -1,
663. offset: u64 = 0,
664. size: u64 = 0,
665. skip_deleted: bool = true,
666. deleted_set: std.AutoHashMapUnmanaged(u64, void) = .empty,
667. scratch: std.mem.Allocator,
669. pub fn init(db: *Db, scratch: std.mem.Allocator, skip_deleted: bool) Scanner {
670. var s = Scanner{ .db = db, .scratch = scratch, .skip_deleted = skip_deleted };
671. s.size = fio.fileSize(scratch, db.data_path) orelse 0;
672. if (s.size > 0) {
673. s.fd = fio.openRead(scratch, db.data_path) orelse -1;
674. }
675. if (skip_deleted) s.deleted_set = db.deletedSet(scratch);
676. return s;
677. }
679. pub fn deinit(self: *Scanner) void {
680. if (self.fd >= 0) fio.close(self.fd);
681. self.fd = -1;
682. }
684. /// Returns the next record (decoded into `arena`) or null at EOF.
685. pub fn next(self: *Scanner, arena: std.mem.Allocator) Error!?Record {
686. while (true) {
687. if (self.fd < 0 or self.offset + 4 > self.size) return null;
688. var hdr: [4]u8 = undefined;
689. const got = fio.pread(self.fd, &hdr, self.offset) orelse return Error.IoError;
690. if (got < 4) return null;
691. const rec_size = std.mem.readInt(u32, &hdr, .little);
692. if (rec_size <= 4) return Error.CorruptRecord;
693. if (self.offset + rec_size > self.size) return null; // incomplete tail
694. const off = self.offset;
695. self.offset += rec_size;
696. if (self.skip_deleted and self.deleted_set.contains(off)) continue;
697. const buf = arena.alloc(u8, rec_size - 4) catch return Error.OutOfMemory;
698. const got2 = fio.pread(self.fd, buf, off + 4) orelse return Error.IoError;
699. if (got2 < rec_size - 4) return Error.Truncated;
700. const d = msgpack.decode(arena, buf) catch return Error.CorruptRecord;
701. return Record{ .value = d.value, .off = off, .len = rec_size };
702. }
703. }
705. /// Like next() but without decoding — yields the raw framed bytes.
706. pub fn nextRaw(self: *Scanner, arena: std.mem.Allocator) Error!?struct { bytes: []u8, off: u64 } {
707. while (true) {
708. if (self.fd < 0 or self.offset + 4 > self.size) return null;
709. var hdr: [4]u8 = undefined;
710. const got = fio.pread(self.fd, &hdr, self.offset) orelse return Error.IoError;
711. if (got < 4) return null;
712. const rec_size = std.mem.readInt(u32, &hdr, .little);
713. if (rec_size <= 4) return Error.CorruptRecord;
714. if (self.offset + rec_size > self.size) return null;
715. const off = self.offset;
716. self.offset += rec_size;
717. if (self.skip_deleted and self.deleted_set.contains(off)) continue;
718. const buf = arena.alloc(u8, rec_size) catch return Error.OutOfMemory;
719. const got2 = fio.pread(self.fd, buf, off) orelse return Error.IoError;
720. if (got2 < rec_size) return Error.Truncated;
721. return .{ .bytes = buf, .off = off };
722. }
723. }
724. };
726. /// Read + decode one record by location.
727. pub fn readAt(self: *Db, arena: std.mem.Allocator, loc: Loc) Error!msgpack.Value {
728. const fd = fio.openRead(self.alloc, self.data_path) orelse return Error.IoError;
729. defer fio.close(fd);
730. if (loc.len <= 4) return Error.CorruptRecord;
731. const buf = arena.alloc(u8, loc.len - 4) catch return Error.OutOfMemory;
732. const got = fio.pread(fd, buf, loc.off + 4) orelse return Error.IoError;
733. if (got < buf.len) return Error.Truncated;
734. const d = msgpack.decode(arena, buf) catch return Error.CorruptRecord;
735. return d.value;
736. }
738. // =====================================================================
739. // Index management
740. // =====================================================================
742. fn keyFromValue(v: msgpack.Value) ?Key {
743. return switch (v) {
744. .number => |n| Key{ .num = n },
745. .str => |s| Key{ .str = s },
746. else => null, // null/bool/objects are not indexed (see header note)
747. };
748. }
750. /// Duplicate a key's string into the db arena so it outlives the op arena.
751. fn ownKey(self: *Db, k: Key) Error!Key {
752. return switch (k) {
753. .num => k,
754. .str => |s| Key{ .str = self.alloc.dupe(u8, s) catch return Error.OutOfMemory },
755. };
756. }
758. fn lowerBound(entries: []const Entry, key: Key) usize {
759. var lo: usize = 0;
760. var hi: usize = entries.len;
761. while (lo < hi) {
762. const mid = lo + (hi - lo) / 2;
763. if (cmpKeys(entries[mid].key, key) < 0) {
764. lo = mid + 1;
765. } else {
766. hi = mid;
767. }
768. }
769. return lo;
770. }
772. /// Binary search: all entries whose key equals `key` (appended to `out`).
773. pub fn indexGet(self: *Db, field: []const u8, key: Key, alloc: std.mem.Allocator, out: *std.ArrayList(Entry)) Error!void {
774. const fi = self.idx.getPtr(field) orelse return Error.NoSuchIndex;
775. const parsed = self.parseKeyForField(field, key);
776. const entries = fi.entries.items;
777. var i = lowerBound(entries, parsed);
778. while (i < entries.len and cmpKeys(entries[i].key, parsed) == 0) : (i += 1) {
779. out.append(alloc, entries[i]) catch return Error.OutOfMemory;
780. }
781. }
783. /// Range scan [from, to] (either side optional), ascending.
784. pub fn indexRange(self: *Db, field: []const u8, from: ?Key, to: ?Key, alloc: std.mem.Allocator, out: *std.ArrayList(Entry)) Error!void {
785. const fi = self.idx.getPtr(field) orelse return Error.NoSuchIndex;
786. const entries = fi.entries.items;
787. var i: usize = if (from) |f| lowerBound(entries, self.parseKeyForField(field, f)) else 0;
788. const to_key: ?Key = if (to) |t| self.parseKeyForField(field, t) else null;
789. while (i < entries.len) : (i += 1) {
790. if (to_key) |t| {
791. if (cmpKeys(entries[i].key, t) > 0) break;
792. }
793. out.append(alloc, entries[i]) catch return Error.OutOfMemory;
794. }
795. }
797. /// _parseKey semantics: numeric index fields parse string keys via
798. /// parseInt; on NaN the key stays a string.
799. fn parseKeyForField(self: *Db, field: []const u8, key: Key) Key {
800. const fi = self.idx.get(field) orelse return key;
801. if (fi.typ == .numeric and key == .str) {
802. if (jsParseInt(key.str)) |n| return Key{ .num = n };
803. }
804. return key;
805. }
807. fn indexInsertEntry(self: *Db, field: []const u8, key: Key, loc: Loc) Error!void {
808. const fi = self.idx.getPtr(field) orelse return;
809. const owned = try self.ownKey(key);
810. const e = Entry{ .key = owned, .off = loc.off, .len = loc.len };
811. // insert at sorted position
812. const pos = blk: {
813. var lo: usize = 0;
814. var hi: usize = fi.entries.items.len;
815. while (lo < hi) {
816. const mid = lo + (hi - lo) / 2;
817. if (entryLess({}, fi.entries.items[mid], e)) {
818. lo = mid + 1;
819. } else {
820. hi = mid;
821. }
822. }
823. break :blk lo;
824. };
825. fi.entries.insert(self.alloc, pos, e) catch return Error.OutOfMemory;
826. }
828. /// Index a record's fields at loc.
829. fn indexInsertRecord(self: *Db, rec: msgpack.Value, loc: Loc) Error!void {
830. for (self.index_fields.items) |field| {
831. const v = rec.get(field) orelse continue;
832. const key = keyFromValue(v) orelse continue;
833. try self.indexInsertEntry(field, key, loc);
834. }
835. self.covered_bytes = @max(self.covered_bytes, loc.off + loc.len);
836. }
838. /// Remove all entries pointing at offset `off` (all fields).
839. fn indexRemoveOffset(self: *Db, off: u64) void {
840. var it = self.idx.iterator();
841. while (it.next()) |kv| {
842. const list = &kv.value_ptr.entries;
843. var i: usize = 0;
844. while (i < list.items.len) {
845. if (list.items[i].off == off) {
846. _ = list.orderedRemove(i);
847. } else {
848. i += 1;
849. }
850. }
851. }
852. }
854. fn indexPath(self: *Db, buf: []u8, field: []const u8) []const u8 {
855. // <dir>/<base>.<field>.txt — derive from idxstate path (…/base.idxstate.json)
856. const prefix = self.idxstate_path[0 .. self.idxstate_path.len - "idxstate.json".len];
857. return std.fmt.bufPrint(buf, "{s}{s}.txt", .{ prefix, field }) catch buf[0..0];
858. }
860. fn initIndexes(self: *Db, force_rebuild: bool) Error!void {
861. var need_rebuild = force_rebuild;
862. const data_size = fio.fileSize(self.alloc, self.data_path) orelse 0;
864. if (!need_rebuild) {
865. for (self.index_fields.items) |field| {
866. var pbuf: [512]u8 = undefined;
867. const p = self.indexPath(&pbuf, field);
868. const isize_ = fio.fileSize(self.alloc, p);
869. if (isize_ == null or isize_.? == 0) {
870. if (data_size > 0) {
871. need_rebuild = true;
872. break;
873. }
874. }
875. }
876. }
877. if (!need_rebuild) {
878. // coveredBytes from idxstate — no state file means rebuild (JS)
879. if (self.readIdxState()) |cb| {
880. self.covered_bytes = cb;
881. } else {
882. need_rebuild = true;
883. }
884. }
886. if (need_rebuild) {
887. try self.rebuildIndexes();
888. return;
889. }
891. // Load index files into memory
892. for (self.index_fields.items) |field| {
893. var pbuf: [512]u8 = undefined;
894. const p = self.indexPath(&pbuf, field);
895. const content = fio.readAll(self.alloc, p) orelse continue;
896. defer self.alloc.free(content);
897. self.loadIndexLines(field, content);
898. }
899. // sort (files are sorted by JS localeCompare; re-sort under our order)
900. var it = self.idx.iterator();
901. while (it.next()) |kv| {
902. std.sort.pdq(Entry, kv.value_ptr.entries.items, {}, entryLess);
903. }
904. if (data_size > self.covered_bytes) self.catchUp(data_size);
905. }
907. fn loadIndexLines(self: *Db, field: []const u8, content: []const u8) void {
908. const fi = self.idx.getPtr(field) orelse return;
909. var lines = std.mem.splitScalar(u8, content, '\n');
910. while (lines.next()) |line| {
911. if (line.len == 0) continue;
912. // key = up to FIRST comma (JS split(',')[0]); then offset, length
913. const c1 = std.mem.indexOfScalar(u8, line, ',') orelse continue;
914. const rest = line[c1 + 1 ..];
915. const c2 = std.mem.indexOfScalar(u8, rest, ',') orelse continue;
916. const off_s = rest[0..c2];
917. const len_s = rest[c2 + 1 ..];
918. const off = std.fmt.parseInt(u64, off_s, 10) catch continue;
919. const len = std.fmt.parseInt(u64, len_s, 10) catch continue;
920. if (len == 0) continue;
921. const key_s = line[0..c1];
922. var key: Key = undefined;
923. if (fi.typ == .numeric) {
924. key = if (jsParseInt(key_s)) |n| Key{ .num = n } else Key{ .str = self.alloc.dupe(u8, key_s) catch return };
925. } else {
926. key = Key{ .str = self.alloc.dupe(u8, key_s) catch return };
927. }
928. fi.entries.append(self.alloc, .{ .key = key, .off = off, .len = len }) catch return;
929. }
930. }
932. fn rebuildIndexes(self: *Db) Error!void {
933. var it0 = self.idx.iterator();
934. while (it0.next()) |kv| kv.value_ptr.entries.clearRetainingCapacity();
936. var scratch_state = std.heap.ArenaAllocator.init(self.arena_state.child_allocator);
937. defer scratch_state.deinit();
938. const scratch = scratch_state.allocator();
940. // JS rebuild scans ALL records (no tombstone filter — deleted offsets
941. // simply get filtered at read time). Mirror that.
942. var scanner = Scanner.init(self, scratch, false);
943. defer scanner.deinit();
944. while (try scanner.next(scratch)) |rec| {
945. try self.indexInsertRecord(rec.value, .{ .off = rec.off, .len = rec.len });
946. }
947. self.covered_bytes = fio.fileSize(self.alloc, self.data_path) orelse 0;
948. try self.writeIndexFiles();
949. try self.writeIdxState();
950. }
952. fn catchUp(self: *Db, file_size: u64) void {
953. var scratch_state = std.heap.ArenaAllocator.init(self.arena_state.child_allocator);
954. defer scratch_state.deinit();
955. const scratch = scratch_state.allocator();
957. var scanner = Scanner.init(self, scratch, false);
958. defer scanner.deinit();
959. scanner.offset = self.covered_bytes;
960. while (true) {
961. const maybe = scanner.next(scratch) catch break;
962. const rec = maybe orelse break;
963. self.indexInsertRecord(rec.value, .{ .off = rec.off, .len = rec.len }) catch break;
964. }
965. self.covered_bytes = @max(self.covered_bytes, file_size);
966. }
968. /// A file that already holds these exact bytes is left alone — its mtime
969. /// included — so closing a handle that changed nothing writes nothing.
970. fn fileHolds(alloc: std.mem.Allocator, path: []const u8, bytes: []const u8) bool {
971. const disk = fio.readAll(alloc, path) orelse return false;
972. defer alloc.free(disk);
973. return std.mem.eql(u8, disk, bytes);
974. }
976. fn writeIndexFiles(self: *Db) Error!void {
977. for (self.index_fields.items) |field| {
978. const fi = self.idx.getPtr(field) orelse continue;
979. var out: std.ArrayList(u8) = .empty;
980. defer out.deinit(self.alloc);
981. for (fi.entries.items) |e| {
982. var kbuf: [32]u8 = undefined;
983. const ks = switch (e.key) {
984. .num => |n| jsNumFmt(&kbuf, n),
985. .str => |s| s,
986. };
987. try self.appendFmt(&out, "{s},{d},{d}\n", .{ ks, e.off, e.len });
988. }
989. var pbuf: [512]u8 = undefined;
990. const p = self.indexPath(&pbuf, field);
991. if (fileHolds(self.alloc, p, out.items)) continue;
992. var tbuf: [512]u8 = undefined;
993. const tmp = std.fmt.bufPrint(&tbuf, "{s}.{d}.tmp", .{ p, linux.getpid() }) catch return Error.IoError;
994. const fd = fio.openTrunc(self.alloc, tmp) orelse return Error.IoError;
995. const ok = fio.writeAll(fd, out.items);
996. fio.close(fd);
997. if (!ok) return Error.IoError;
998. if (!fio.rename(self.alloc, tmp, p)) return Error.IoError;
999. }
1000. }
1002. fn readIdxState(self: *Db) ?u64 {
1003. const content = fio.readAll(self.alloc, self.idxstate_path) orelse return null;
1004. defer self.alloc.free(content);
1005. const parsed = std.json.parseFromSlice(std.json.Value, self.alloc, content, .{}) catch return null;
1006. defer parsed.deinit();
1007. if (parsed.value != .object) return null;
1008. const v = parsed.value.object.get("coveredBytes") orelse return null;
1009. return switch (v) {
1010. .integer => |i| @intCast(@max(i, 0)),
1011. .float => |f| @intFromFloat(@max(f, 0)),
1012. else => null,
1013. };
1014. }
1016. fn writeIdxState(self: *Db) Error!void {
1017. var buf: [128]u8 = undefined;
1018. const json = std.fmt.bufPrint(&buf, "{{\"coveredBytes\":{d}}}", .{self.covered_bytes}) catch return Error.IoError;
1019. if (fileHolds(self.alloc, self.idxstate_path, json)) return;
1020. var tbuf: [512]u8 = undefined;
1021. const tmp = std.fmt.bufPrint(&tbuf, "{s}.{d}.tmp", .{ self.idxstate_path, linux.getpid() }) catch return Error.IoError;
1022. const fd = fio.openTrunc(self.alloc, tmp) orelse return Error.IoError;
1023. const ok = fio.writeAll(fd, json);
1024. fio.close(fd);
1025. if (!ok) return Error.IoError;
1026. if (!fio.rename(self.alloc, tmp, self.idxstate_path)) return Error.IoError;
1027. }
1029. fn persistIndexesLocked(self: *Db) Error!void {
1030. try self.writeIndexFiles();
1031. // coverage can only grow (JS: max of ours and on-disk state)
1032. if (self.readIdxState()) |cb| self.covered_bytes = @max(self.covered_bytes, cb);
1033. try self.writeIdxState();
1034. }
1036. // =====================================================================
1037. // Operations
1038. // =====================================================================
1040. /// A mutable record under construction (op-arena entries).
1041. pub const MutableRecord = struct {
1042. entries: std.ArrayList(msgpack.Entry) = .empty,
1044. pub fn get(self: *const MutableRecord, key: []const u8) ?msgpack.Value {
1045. for (self.entries.items) |e| {
1046. if (e.key == .str and std.mem.eql(u8, e.key.str, key)) return e.value;
1047. }
1048. return null;
1049. }
1051. pub fn set(self: *MutableRecord, alloc: std.mem.Allocator, key: []const u8, v: msgpack.Value) Error!void {
1052. for (self.entries.items) |*e| {
1053. if (e.key == .str and std.mem.eql(u8, e.key.str, key)) {
1054. e.value = v;
1055. return;
1056. }
1057. }
1058. self.entries.append(alloc, .{ .key = .{ .str = key }, .value = v }) catch return Error.OutOfMemory;
1059. }
1061. pub fn toValue(self: *const MutableRecord) msgpack.Value {
1062. return .{ .map = self.entries.items };
1063. }
1064. };
1066. pub const InsertResult = union(enum) {
1067. pk_num: f64,
1068. pk_str: []const u8, // op-arena
1069. record: msgpack.Value,
1070. };
1072. /// insert() — auto primary key, unique checks, append, index.
1073. /// `record` must be a map value; `arena` is the op arena (record memory).
1074. pub fn insert(self: *Db, arena: std.mem.Allocator, record: msgpack.Value, skip_pk: bool) Error!InsertResult {
1075. if (record != .map) return Error.CorruptRecord;
1076. try self.acquireLock();
1077. defer self.releaseLock();
1078. self.refresh();
1079. return self.insertLocked(arena, record, skip_pk);
1080. }
1082. fn insertLocked(self: *Db, arena: std.mem.Allocator, record: msgpack.Value, skip_pk: bool) Error!InsertResult {
1083. // copy into mutable form
1084. var rec = MutableRecord{};
1085. for (record.map) |e| rec.entries.append(arena, e) catch return Error.OutOfMemory;
1087. // _hasPrimaryKeyValue: present and not null/undefined
1088. var has_pk_value = false;
1089. if (self.pk) |p| {
1090. if (rec.get(p)) |v| {
1091. has_pk_value = (v != .nil and v != .undef);
1092. }
1093. }
1095. var auto_gen = false;
1096. if (self.pk != null and !skip_pk and !has_pk_value) {
1097. if (self.pk_type == .number) {
1098. rec.set(arena, self.pk.?, .{ .number = self.next_id }) catch return Error.OutOfMemory;
1099. self.next_id += 1;
1100. auto_gen = true;
1101. } else if (self.pk_type == .uuid) {
1102. var ubuf: [12]u8 = undefined;
1103. const u = genUuid(&ubuf);
1104. const owned = arena.dupe(u8, u) catch return Error.OutOfMemory;
1105. rec.set(arena, self.pk.?, .{ .str = owned }) catch return Error.OutOfMemory;
1106. auto_gen = true;
1107. }
1108. }
1110. // unique constraints (skip auto-generated pk)
1111. if (self.index_fields.items.len > 0 and self.unique_fields.items.len > 0) {
1112. var del_set = self.deletedSet(arena);
1113. for (self.unique_fields.items) |field| {
1114. if (auto_gen and self.pk != null and std.mem.eql(u8, field, self.pk.?)) continue;
1115. const v = rec.get(field) orelse continue;
1116. if (v == .undef) continue;
1117. const key = keyFromValue(v) orelse continue;
1118. var hits: std.ArrayList(Entry) = .empty;
1119. defer hits.deinit(arena);
1120. try self.indexGet(field, key, arena, &hits);
1121. var live: usize = 0;
1122. for (hits.items) |h| {
1123. if (!del_set.contains(h.off)) live += 1;
1124. }
1125. if (live > 0) {
1126. var kbuf: [32]u8 = undefined;
1127. const ks = switch (key) {
1128. .num => |n| jsNumFmt(&kbuf, n),
1129. .str => |s| s,
1130. };
1131. self.last_error = std.fmt.bufPrint(&self.last_error_buf, "Duplicate key: {s}={s}", .{ field, ks }) catch "Duplicate key";
1132. return Error.DuplicateKey;
1133. }
1134. }
1135. }
1137. // serialize + append
1138. const framed = msgpack.serialize(arena, rec.toValue()) catch return Error.OutOfMemory;
1139. const offset = fio.fileSize(self.alloc, self.data_path) orelse 0;
1140. const fd = fio.openAppend(self.alloc, self.data_path) orelse return Error.IoError;
1141. const ok = fio.writeAll(fd, framed);
1142. fio.close(fd);
1143. if (!ok) return Error.IoError;
1145. const loc = Loc{ .off = offset, .len = framed.len };
1146. if (self.index_fields.items.len > 0) {
1147. try self.indexInsertRecord(rec.toValue(), loc);
1148. }
1150. if (self.pk != null and self.pk_type == .number and !skip_pk and !has_pk_value) {
1151. try self.persistMetaLocked();
1152. }
1154. if (self.pk) |p| {
1155. const v = rec.get(p) orelse return Error.CorruptRecord;
1156. return switch (v) {
1157. .number => |n| InsertResult{ .pk_num = n },
1158. .str => |s| InsertResult{ .pk_str = s },
1159. else => InsertResult{ .record = rec.toValue() },
1160. };
1161. }
1162. return InsertResult{ .record = rec.toValue() };
1163. }
1165. /// Locate live records by primary key (index-backed when available).
1166. pub fn findPkLocs(self: *Db, arena: std.mem.Allocator, key: Key, out: *std.ArrayList(Loc)) Error!void {
1167. if (self.pk == null) return Error.NoPrimaryKey;
1168. self.refresh();
1169. var del_set = self.deletedSet(arena);
1170. if (self.index_fields.items.len > 0) {
1171. var hits: std.ArrayList(Entry) = .empty;
1172. defer hits.deinit(arena);
1173. try self.indexGet(self.pk.?, key, arena, &hits);
1174. for (hits.items) |h| {
1175. if (del_set.contains(h.off)) continue;
1176. out.append(arena, .{ .off = h.off, .len = h.len }) catch return Error.OutOfMemory;
1177. }
1178. return;
1179. }
1180. // no indexes: full scan comparing the pk field
1181. var scanner = Scanner.init(self, arena, true);
1182. defer scanner.deinit();
1183. while (try scanner.next(arena)) |rec| {
1184. const v = rec.value.get(self.pk.?) orelse continue;
1185. const k = keyFromValue(v) orelse continue;
1186. if (cmpKeys(self.parseKeyForField(self.pk.?, k), self.parseKeyForField(self.pk.?, key)) == 0) {
1187. out.append(arena, .{ .off = rec.off, .len = rec.len }) catch return Error.OutOfMemory;
1188. }
1189. }
1190. }
1192. /// Locate live records by secondary index equality.
1193. pub fn findIndexLocs(self: *Db, arena: std.mem.Allocator, field: []const u8, key: Key, out: *std.ArrayList(Loc)) Error!void {
1194. self.refresh();
1195. var del_set = self.deletedSet(arena);
1196. var hits: std.ArrayList(Entry) = .empty;
1197. defer hits.deinit(arena);
1198. try self.indexGet(field, key, arena, &hits);
1199. for (hits.items) |h| {
1200. if (del_set.contains(h.off)) continue;
1201. out.append(arena, .{ .off = h.off, .len = h.len }) catch return Error.OutOfMemory;
1202. }
1203. }
1205. /// Locate live records by index range [from, to] ascending.
1206. pub fn findRangeLocs(self: *Db, arena: std.mem.Allocator, field: []const u8, from: ?Key, to: ?Key, out: *std.ArrayList(Loc)) Error!void {
1207. self.refresh();
1208. var del_set = self.deletedSet(arena);
1209. var hits: std.ArrayList(Entry) = .empty;
1210. defer hits.deinit(arena);
1211. try self.indexRange(field, from, to, arena, &hits);
1212. var seen: std.AutoHashMapUnmanaged(u64, void) = .empty;
1213. for (hits.items) |h| {
1214. if (del_set.contains(h.off)) continue;
1215. if (seen.contains(h.off)) continue;
1216. seen.put(arena, h.off, {}) catch {};
1217. out.append(arena, .{ .off = h.off, .len = h.len }) catch return Error.OutOfMemory;
1218. }
1219. }
1221. /// delete by primary key. Returns the deleted records (decoded into arena).
1222. pub fn deletePk(self: *Db, arena: std.mem.Allocator, key: Key, out_records: *std.ArrayList(msgpack.Value)) Error!usize {
1223. try self.acquireLock();
1224. defer self.releaseLock();
1225. self.refresh();
1226. return self.deletePkLocked(arena, key, out_records);
1227. }
1229. fn deletePkLocked(self: *Db, arena: std.mem.Allocator, key: Key, out_records: *std.ArrayList(msgpack.Value)) Error!usize {
1230. var locs: std.ArrayList(Loc) = .empty;
1231. defer locs.deinit(arena);
1232. try self.findPkLocs(arena, key, &locs);
1233. for (locs.items) |loc| {
1234. const rec = try self.readAt(arena, loc);
1235. out_records.append(arena, rec) catch return Error.OutOfMemory;
1236. self.deleted.append(self.alloc, loc.off) catch return Error.OutOfMemory;
1237. self.indexRemoveOffset(loc.off);
1238. }
1239. try self.persistMetaLocked();
1240. return locs.items.len;
1241. }
1243. /// update by primary key: delete + insert(skip_pk) — JS update() semantics.
1244. /// `new_record` must already carry the primary key (the JS callback
1245. /// contract: the record keeps its pk unless the caller removes it).
1246. pub fn updatePk(self: *Db, arena: std.mem.Allocator, key: Key, new_record: msgpack.Value) Error!usize {
1247. if (self.pk) |p| {
1248. const carried = for (new_record.map) |e| {
1249. if (e.key == .str and std.mem.eql(u8, e.key.str, p)) break e.value != .nil and e.value != .undef;
1250. } else false;
1251. if (!carried) return Error.RecordLacksPrimaryKey;
1252. }
1253. try self.acquireLock();
1254. defer self.releaseLock();
1255. self.refresh();
1256. var old: std.ArrayList(msgpack.Value) = .empty;
1257. defer old.deinit(arena);
1258. const n = try self.deletePkLocked(arena, key, &old);
1259. if (n == 0) return 0;
1260. var i: usize = 0;
1261. while (i < n) : (i += 1) {
1262. _ = try self.insertLocked(arena, new_record, true);
1263. }
1264. return n;
1265. }
1267. /// compact() — rewrite the data file without tombstones, rebuild indexes.
1268. pub fn compact(self: *Db) Error!void {
1269. try self.acquireLock();
1270. defer self.releaseLock();
1271. self.refresh();
1272. try self.compactLocked();
1273. if (self.index_fields.items.len > 0) {
1274. try self.rebuildIndexes();
1275. }
1276. }
1278. fn compactLocked(self: *Db) Error!void {
1279. var scratch_state = std.heap.ArenaAllocator.init(self.arena_state.child_allocator);
1280. defer scratch_state.deinit();
1281. const scratch = scratch_state.allocator();
1283. var tbuf: [512]u8 = undefined;
1284. const tmp = std.fmt.bufPrint(&tbuf, "{s}.tmp", .{self.data_path}) catch return Error.IoError;
1285. const out_fd = fio.openTrunc(self.alloc, tmp) orelse return Error.IoError;
1287. var scanner = Scanner.init(self, scratch, true);
1288. var ok = true;
1289. while (true) {
1290. const maybe = scanner.nextRaw(scratch) catch {
1291. ok = false;
1292. break;
1293. };
1294. const raw = maybe orelse break;
1295. if (!fio.writeAll(out_fd, raw.bytes)) {
1296. ok = false;
1297. break;
1298. }
1299. }
1300. scanner.deinit();
1301. fio.close(out_fd);
1302. if (!ok) return Error.IoError;
1303. if (!fio.rename(self.alloc, tmp, self.data_path)) return Error.IoError;
1305. self.deleted.clearRetainingCapacity();
1306. try self.persistMetaLocked();
1307. }
1309. pub fn persistNow(self: *Db) Error!void {
1310. try self.acquireLock();
1311. defer self.releaseLock();
1312. self.refresh();
1313. try self.persistIndexesLocked();
1314. }
1315. };
1317. // =========================================================================
1318. // Tests
1319. // =========================================================================
1321. const testing = std.testing;
1323. fn tmpBase(buf: []u8, comptime name: []const u8) []const u8 {
1324. return std.fmt.bufPrint(buf, "/tmp/mpackdb-zigtest-{d}-" ++ name, .{linux.getpid()}) catch unreachable;
1325. }
1327. fn cleanup(alloc: std.mem.Allocator, base: []const u8) void {
1328. var buf: [512]u8 = undefined;
1329. const suffixes = [_][]const u8{ ".mpack", ".meta.json", ".idxstate.json", ".lock", ".id.txt", ".email.txt", ".age.txt", ".uuid.txt" };
1330. for (suffixes) |suffix| {
1331. const p = std.fmt.bufPrint(&buf, "{s}{s}", .{ base, suffix }) catch continue;
1332. _ = fio.unlink(alloc, p);
1333. }
1334. }
1336. fn strKey(s: []const u8) Key {
1337. return .{ .str = s };
1338. }
1340. fn numKey(n: f64) Key {
1341. return .{ .num = n };
1342. }
1344. fn makeUser(arena: std.mem.Allocator, name: []const u8, email: []const u8, age: f64) !msgpack.Value {
1345. const entries = try arena.alloc(msgpack.Entry, 3);
1346. entries[0] = .{ .key = .{ .str = "name" }, .value = .{ .str = name } };
1347. entries[1] = .{ .key = .{ .str = "email" }, .value = .{ .str = email } };
1348. entries[2] = .{ .key = .{ .str = "age" }, .value = .{ .number = age } };
1349. return .{ .map = entries };
1350. }
1352. test "engine: insert/find/delete/update with numeric pk + indexes" {
1353. var base_buf: [128]u8 = undefined;
1354. const base = tmpBase(&base_buf, "crud");
1355. cleanup(testing.allocator, base);
1356. defer cleanup(testing.allocator, base);
1358. var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
1359. defer arena_state.deinit();
1360. const arena = arena_state.allocator();
1362. const db = try Db.open(testing.allocator, base, .{
1363. .primary_key = "*id",
1364. .indexes = &.{ "email", "*age" },
1365. });
1367. // insert three
1368. const r1 = try db.insert(arena, try makeUser(arena, "Alice", "[email protected]", 30), false);
1369. try testing.expectEqual(@as(f64, 0), r1.pk_num);
1370. const r2 = try db.insert(arena, try makeUser(arena, "Bob", "[email protected]", 25), false);
1371. try testing.expectEqual(@as(f64, 1), r2.pk_num);
1372. _ = try db.insert(arena, try makeUser(arena, "Carol", "[email protected]", 35), false);
1374. // find by pk
1375. var locs: std.ArrayList(Loc) = .empty;
1376. try db.findPkLocs(arena, numKey(1), &locs);
1377. try testing.expectEqual(@as(usize, 1), locs.items.len);
1378. const bob = try db.readAt(arena, locs.items[0]);
1379. try testing.expectEqualStrings("Bob", bob.get("name").?.str);
1381. // find by secondary index
1382. var locs2: std.ArrayList(Loc) = .empty;
1383. try db.findIndexLocs(arena, "email", strKey("[email protected]"), &locs2);
1384. try testing.expectEqual(@as(usize, 1), locs2.items.len);
1386. // range on numeric index: age 26..40 → Alice(30), Carol(35)
1387. var locs3: std.ArrayList(Loc) = .empty;
1388. try db.findRangeLocs(arena, "age", numKey(26), numKey(40), &locs3);
1389. try testing.expectEqual(@as(usize, 2), locs3.items.len);
1391. // update Bob's age
1392. var bob_new = Db.MutableRecord{};
1393. for (bob.map) |e| try bob_new.entries.append(arena, e);
1394. try bob_new.set(arena, "age", .{ .number = 26 });
1395. const updated = try db.updatePk(arena, numKey(1), bob_new.toValue());
1396. try testing.expectEqual(@as(usize, 1), updated);
1398. var locs4: std.ArrayList(Loc) = .empty;
1399. try db.findPkLocs(arena, numKey(1), &locs4);
1400. try testing.expectEqual(@as(usize, 1), locs4.items.len);
1401. const bob2 = try db.readAt(arena, locs4.items[0]);
1402. try testing.expectEqual(@as(f64, 26), bob2.get("age").?.number);
1404. // delete Alice
1405. var deleted_recs: std.ArrayList(msgpack.Value) = .empty;
1406. const dn = try db.deletePk(arena, numKey(0), &deleted_recs);
1407. try testing.expectEqual(@as(usize, 1), dn);
1408. try testing.expectEqualStrings("Alice", deleted_recs.items[0].get("name").?.str);
1410. var locs5: std.ArrayList(Loc) = .empty;
1411. try db.findPkLocs(arena, numKey(0), &locs5);
1412. try testing.expectEqual(@as(usize, 0), locs5.items.len);
1414. db.close();
1416. // reopen (compaction drops the tombstone) and verify persistence
1417. const db2 = try Db.open(testing.allocator, base, .{
1418. .primary_key = "*id",
1419. .indexes = &.{ "email", "*age" },
1420. });
1421. defer db2.close();
1422. var locs6: std.ArrayList(Loc) = .empty;
1423. try db2.findPkLocs(arena, numKey(1), &locs6);
1424. try testing.expectEqual(@as(usize, 1), locs6.items.len);
1425. const bob3 = try db2.readAt(arena, locs6.items[0]);
1426. try testing.expectEqual(@as(f64, 26), bob3.get("age").?.number);
1427. // nextId continues after reopen
1428. const r4 = try db2.insert(arena, try makeUser(arena, "Dan", "[email protected]", 40), false);
1429. try testing.expectEqual(@as(f64, 3), r4.pk_num);
1430. }
1432. test "engine: unique index rejects duplicates" {
1433. var base_buf: [128]u8 = undefined;
1434. const base = tmpBase(&base_buf, "uniq");
1435. cleanup(testing.allocator, base);
1436. defer cleanup(testing.allocator, base);
1438. var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
1439. defer arena_state.deinit();
1440. const arena = arena_state.allocator();
1442. const db = try Db.open(testing.allocator, base, .{
1443. .primary_key = "*id",
1444. .indexes = &.{"!email"},
1445. });
1446. defer db.close();
1448. _ = try db.insert(arena, try makeUser(arena, "Alice", "[email protected]", 30), false);
1449. const dup = db.insert(arena, try makeUser(arena, "Evil", "[email protected]", 31), false);
1450. try testing.expectError(Error.DuplicateKey, dup);
1451. try testing.expect(std.mem.indexOf(u8, db.last_error, "[email protected]") != null);
1452. }
1454. test "engine: uuid primary key" {
1455. var base_buf: [128]u8 = undefined;
1456. const base = tmpBase(&base_buf, "uuid");
1457. cleanup(testing.allocator, base);
1458. defer cleanup(testing.allocator, base);
1460. var arena_state = std.heap.ArenaAllocator.init(testing.allocator);
1461. defer arena_state.deinit();
1462. const arena = arena_state.allocator();
1464. const db = try Db.open(testing.allocator, base, .{ .primary_key = "@uuid" });
1465. defer db.close();
1467. const r = try db.insert(arena, try makeUser(arena, "Alice", "[email protected]", 1), false);
1468. try testing.expectEqual(@as(usize, 12), r.pk_str.len);
1469. var locs: std.ArrayList(Loc) = .empty;
1470. try db.findPkLocs(arena, strKey(r.pk_str), &locs);
1471. try testing.expectEqual(@as(usize, 1), locs.items.len);
1472. }
1474. test "jsNumFmt / jsParseInt / uuid shape" {
1475. var buf: [32]u8 = undefined;
1476. try testing.expectEqualStrings("42", jsNumFmt(&buf, 42));
1477. try testing.expectEqualStrings("-7", jsNumFmt(&buf, -7));
1478. try testing.expectEqualStrings("1.5", jsNumFmt(&buf, 1.5));
1479. try testing.expectEqual(@as(f64, 1), jsParseInt("1.5").?);
1480. try testing.expectEqual(@as(f64, -12), jsParseInt("-12abc").?);
1481. try testing.expect(jsParseInt("abc") == null);
1482. var ubuf: [12]u8 = undefined;
1483. const u = genUuid(&ubuf);
1484. try testing.expectEqual(@as(usize, 12), u.len);
1485. for (u) |ch| try testing.expect((ch >= '0' and ch <= '9') or (ch >= 'a' and ch <= 'z'));
1486. }