diff --git a/README.md b/README.md index aa3c060..36d79eb 100644 --- a/README.md +++ b/README.md @@ -39,8 +39,9 @@ mongosh --port 27017 classes, groups, alternation, `i`/`s` options) `$not` `$and` `$or` `$nor` `$size` `$all` `$elemMatch`, with dot paths and array multikey semantics. - **Secondary indexes**: `createIndex`/`listIndexes`/`dropIndex` via the - three driver commands, single-field and compound, with `unique` and - `sparse` options, persisted in the log and rebuilt on open (compaction + three driver commands, single-field and compound, with `unique`, + `sparse` and `expireAfterSeconds` (TTL) options, persisted in the log + and rebuilt on open (compaction re-emits them). The query planner turns equality / `$in` / range predicates into index lookups across `find`, `count`, `update`, `delete`, `findAndModify`, and a leading `$match` in `aggregate`; every @@ -68,13 +69,13 @@ src/ bson.zig BSON parse/serialize, ObjectId, canonical comparison order wire.zig OP_MSG/OP_QUERY framing, message + reply builders commands.zig command dispatch (hello, CRUD, aggregate, admin, indexes) - server.zig TCP accept loop, per-connection handlers + server.zig TCP accept loop, per-connection handlers, TTL sweep monitor db.zig in-memory engine: db → collection → _id → document maps storage.zig append-only log: records, replay, CRC validation query.zig filter matcher, regex engine, sort, projection index.zig secondary indexes: entries, search, query planner update.zig update operators with dot-path navigation - main.zig CLI: --port, --bind, --db + main.zig CLI: --port, --bind, --db, --ttl-sweep-secs ``` ## Indexes @@ -91,6 +92,18 @@ the query planner to narrow scans. - **Options**: `unique` (a conflicting write fails with E11000 naming the index; per-document entries are deduped first, so `{a: [1,1]}` is legal) and `sparse` (documents missing an indexed field are skipped). +- **TTL**: `createIndex({expireAt: 1}, {expireAfterSeconds: 60})` deletes a + document once its indexed date is that many seconds old. A background + sweeper runs every `--ttl-sweep-secs` seconds (default 60, `0` disables + it) and deletes through the ordinary write path, so each expiry is logged + and fsynced and holds across a restart. As in MongoDB the option is + single-field only (a compound key is `CannotCreateIndex`, code 67), + `expireAfterSeconds` must be a whole number in `[0, 2147483647]` (`0` + means "expire at the stored instant"), a non-date value at the path never + expires, an array of dates expires on its earliest member, and expiry is + coarse: a document stays visible until the next sweep. Changing the + expiry of an existing index is `IndexOptionsConflict` — `collMod` is not + implemented. - **Multikey**: an array at an indexed path is indexed as a whole *and* element-wise, mirroring the query matcher exactly, so both `{tags: "a"}` and `{tags: ["a","b"]}` hit the index. A compound index @@ -106,9 +119,9 @@ the query planner to narrow scans. serialization-ambiguous (int32 1, int64 1, double 1.0 compare equal but hash differently — those fall back to a scan, as do string/symbol/code). -v1 limits: no index-accelerated sort, no hashed/text/geo/TTL/partial -indexes, and entry insert/removal is O(n) (a sorted array) — fine for a -light database, with a B-tree or id→entry map as the follow-up. +v1 limits: no index-accelerated sort, no hashed/text/geo/partial indexes, +and entry insert/removal is O(n) (a sorted array) — fine for a light +database, with a B-tree or id→entry map as the follow-up. ## Not (yet) implemented diff --git a/src/commands.zig b/src/commands.zig index db2cef1..34066b1 100644 --- a/src/commands.zig +++ b/src/commands.zig @@ -32,6 +32,8 @@ pub const ErrorCode = enum(i32) { internal_error = 1, invalid_pipeline_operator = 40324, index_options_conflict = 85, + cannot_create_index = 67, + invalid_index_specification_option = 197, }; /// Which lock (if any) a command needs on the engine. Contract: only @@ -344,7 +346,19 @@ fn cmd_create_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !vo const is_id_index = key_pairs.len == 1 and ((only == .int32 and only.int32 == 1) or (only == .int64 and only.int64 == 1) or (only == .double and only.double == 1.0)); - if (is_id_index) continue; + if (is_id_index) { + // The no-op must not swallow options that would change what + // the index does — MongoDB rejects a TTL _id index rather + // than quietly ignoring the expiry. + if (bson.get_pair(spec, "expireAfterSeconds") != null) { + return reply.put_error( + @intFromEnum(ErrorCode.invalid_index_specification_option), + "InvalidIndexSpecificationOption", + "the field 'expireAfterSeconds' is not valid for an _id index specification", + ); + } + continue; + } return bad_value(reply, "cannot create a secondary index on the _id field"); } const name = bson.get_pair(spec, "name") orelse bson.Value.null; @@ -355,6 +369,16 @@ fn cmd_create_indexes(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !vo const spec_doc = bson.Document{ .arena = undefined, .pairs = spec }; _ = ctx.engine.create_index(db_name, coll_name, &spec_doc) catch |err| switch (err) { error.InvalidIndexSpec => return bad_value(reply, "invalid index spec"), + error.TtlOnCompoundIndex => return reply.put_error( + @intFromEnum(ErrorCode.cannot_create_index), + "CannotCreateIndex", + "TTL indexes are single-field indexes, compound indexes do not support TTL", + ), + error.InvalidExpireAfterSeconds => return reply.put_error( + @intFromEnum(ErrorCode.cannot_create_index), + "CannotCreateIndex", + "TTL index 'expireAfterSeconds' option must be a whole number between 0 and 2147483647", + ), error.IndexOptionsConflict => return reply.put_error(@intFromEnum(ErrorCode.index_options_conflict), "IndexOptionsConflict", "index already exists with a different specification"), error.DuplicateKeyIndex => { const ix_name = if (name == .string) name.string else "index"; @@ -1642,6 +1666,90 @@ test "createIndexes, listIndexes, dropIndexes, and idempotent re-create" { } } +test "TTL index round-trips through createIndexes/listIndexes; bad specs give 67" { + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tdb = try TestDb.init(io); + defer tdb.deinit(); + + // A driver sends expireAfterSeconds as a double. + try dispatch_create_index(&tdb, io, "sessions", .{ .doc = &.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ .key = "name", .value = .{ .string = "expireAt_1" } }, + .{ .key = "expireAfterSeconds", .value = .{ .double = 60.0 } }, + } }); + try testing.expectEqual(@as(?i64, 60), tdb.engine.get_collection("test", "sessions").?.indexes.items[0].ttl); + + // listIndexes reports it back. + { + var ctx = tdb.ctx(io); + var msg = try parse_fake_msg("listIndexes", .{ .string = "sessions" }, &.{}); + defer msg.deinit(); + var reply = wire.Reply.init(testing.allocator); + defer reply.deinit(); + try dispatch(&ctx, &msg, &reply); + const cursor = bson.get_pair(reply.pairs.items, "cursor").?; + const batch = bson.get_pair(cursor.doc, "firstBatch").?.array; + try testing.expectEqual(@as(usize, 2), batch.len); + try testing.expectEqual(@as(i32, 60), bson.get_pair(batch[1].doc, "expireAfterSeconds").?.int32); + // The _id_ entry never carries one. + try testing.expect(bson.get_pair(batch[0].doc, "expireAfterSeconds") == null); + } + + // Same name and key, different expiry: IndexOptionsConflict, as in + // MongoDB (changing it is a collMod, which this server does not have). + const bad = [_]struct { spec: bson.Value, code: i32 }{ + .{ .code = 85, .spec = .{ .doc = &.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ .key = "name", .value = .{ .string = "expireAt_1" } }, + .{ .key = "expireAfterSeconds", .value = .{ .int32 = 90 } }, + } } }, + // TTL on a compound key. + .{ .code = 67, .spec = .{ .doc = &.{ + .{ .key = "key", .value = .{ .doc = &.{ + .{ .key = "a", .value = .{ .int32 = 1 } }, + .{ .key = "b", .value = .{ .int32 = 1 } }, + } } }, + .{ .key = "expireAfterSeconds", .value = .{ .int32 = 60 } }, + } } }, + // Negative and non-numeric expiries. + .{ .code = 67, .spec = .{ .doc = &.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } }, + .{ .key = "expireAfterSeconds", .value = .{ .int32 = -1 } }, + } } }, + .{ .code = 67, .spec = .{ .doc = &.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } }, + .{ .key = "expireAfterSeconds", .value = .{ .string = "60" } }, + } } }, + // Past MongoDB's 2147483647 bound. + .{ .code = 67, .spec = .{ .doc = &.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } }, + .{ .key = "expireAfterSeconds", .value = .{ .int64 = index.max_expire_after_seconds + 1 } }, + } } }, + // {_id: 1} is otherwise an idempotent no-op, but an expiry on it + // would be silently dropped, so it is rejected instead. + .{ .code = 197, .spec = .{ .doc = &.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 1 } }} } }, + .{ .key = "expireAfterSeconds", .value = .{ .int32 = 60 } }, + } } }, + }; + for (bad) |case| { + var ctx = tdb.ctx(io); + const specs = [_]bson.Value{case.spec}; + var msg = try parse_fake_msg("createIndexes", .{ .string = "sessions" }, &.{ + .{ .key = "indexes", .value = .{ .array = &specs } }, + }); + defer msg.deinit(); + var reply = wire.Reply.init(testing.allocator); + defer reply.deinit(); + try dispatch(&ctx, &msg, &reply); + try testing.expectEqual(case.code, bson.get_pair(reply.pairs.items, "code").?.int32); + } + // Nothing partial was registered by the rejected specs. + try testing.expectEqual(@as(usize, 1), tdb.engine.get_collection("test", "sessions").?.indexes.items.len); +} + test "unique index constraint returns 11000 through insert and update" { var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); diff --git a/src/db.zig b/src/db.zig index de10bcc..03a4766 100644 --- a/src/db.zig +++ b/src/db.zig @@ -360,6 +360,76 @@ pub const Engine = struct { return true; } + /// Delete every document expired as of `now_ms` (Unix milliseconds) + /// under some TTL index, and return how many were deleted. Callers must + /// hold the write lock; the server's monitor coroutine (src/server.zig) + /// is the only caller in production, tests call it with a fixed clock. + /// + /// Each expiry goes through `remove`, so it is logged and fsynced like + /// any other delete and survives a restart. Expiry is therefore coarse + /// by design (as in MongoDB): an expired document stays visible until + /// the next sweep. + pub fn ttl_sweep(self: *Engine, now_ms: i64) !usize { + var deleted: usize = 0; + // Ids are duped rather than aliased: `remove` frees the docs-map key + // that `Entry.id` points at, which would leave the rest of the batch + // pointing into freed memory. + var ids: std.ArrayListUnmanaged([]u8) = .empty; + defer { + for (ids.items) |id| self.gpa.free(id); + ids.deinit(self.gpa); + } + + var db_it = self.dbs.iterator(); + while (db_it.next()) |db_entry| { + var coll_it = db_entry.value_ptr.collections.iterator(); + while (coll_it.next()) |coll_entry| { + for (ids.items) |id| self.gpa.free(id); + ids.clearRetainingCapacity(); + + for (coll_entry.value_ptr.indexes.items) |*ix| { + const ttl = ix.ttl orelse continue; + const cutoff: i128 = @as(i128, now_ms) - @as(i128, ttl) * 1000; + for (ix.entries.items) |e| { + // The type test cannot be a range lookup: bson + // compare order ranks datetime above null, numbers + // and strings, so a datetime upper bound would also + // select every value of a lesser type. + if (e.key[0] != .datetime) continue; + if (@as(i128, e.key[0].datetime) > cutoff) continue; + try ids.append(self.gpa, try self.gpa.dupe(u8, e.id)); + } + } + if (ids.items.len == 0) continue; + + // One document can be expired by several entries (an array + // of dates) or by several TTL indexes. + std.mem.sort([]u8, ids.items, {}, less_id_bytes); + var w: usize = 1; + for (ids.items[1..]) |id| { + if (std.mem.eql(u8, id, ids.items[w - 1])) { + self.gpa.free(id); + } else { + ids.items[w] = id; + w += 1; + } + } + ids.items.len = w; + + // `remove` only mutates the collection's docs map and index + // entries, never the dbs/collections maps, so both iterators + // above stay valid. + for (ids.items) |id| { + if (try self.remove(db_entry.key_ptr.*, coll_entry.key_ptr.*, id)) deleted += 1; + } + } + } + // A TTL-only workload never reaches the threshold check in `upsert`, + // so the log would otherwise grow without bound. + if (deleted > 0) try self.maybe_compact(); + return deleted; + } + pub fn database_names(self: *Engine, out: *std.ArrayListUnmanaged([]const u8)) !void { var it = self.dbs.iterator(); while (it.next()) |entry| try out.append(self.gpa, entry.key_ptr.*); @@ -505,6 +575,10 @@ pub const Engine = struct { } }; +fn less_id_bytes(_: void, a: []const u8, b: []const u8) bool { + return std.mem.order(u8, a, b) == .lt; +} + fn parent_dir(path: []const u8) []const u8 { const last = std.mem.lastIndexOfScalar(u8, path, '/') orelse return "."; if (last == 0) return "/"; @@ -826,16 +900,19 @@ test "concurrent readers and writers on a threaded Io" { /// A spec document for a single-path index, built by serializing and /// re-parsing so the pairs are arena-owned. -fn index_spec(gpa: std.mem.Allocator, path: []const u8, name: []const u8, unique: bool, sparse: bool) !bson.Document { +fn index_spec(gpa: std.mem.Allocator, path: []const u8, name: []const u8, unique: bool, sparse: bool, ttl: ?i64) !bson.Document { var out: std.ArrayListUnmanaged(u8) = .empty; defer out.deinit(gpa); - const pairs = [_]bson.Pair{ + var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty; + defer pairs.deinit(gpa); + try pairs.appendSlice(gpa, &.{ .{ .key = "key", .value = .{ .doc = &.{.{ .key = path, .value = .{ .int32 = 1 } }} } }, .{ .key = "name", .value = .{ .string = name } }, .{ .key = "unique", .value = .{ .bool = unique } }, .{ .key = "sparse", .value = .{ .bool = sparse } }, - }; - try bson.write_doc(&pairs, gpa, &out); + }); + if (ttl) |secs| try pairs.append(gpa, .{ .key = "expireAfterSeconds", .value = .{ .int64 = secs } }); + try bson.write_doc(pairs.items, gpa, &out); return bson.Document.parse(gpa, out.items); } @@ -874,7 +951,7 @@ test "unique index enforced on insert, replace, and upsert-conflict" { var engine = try Engine.open(gpa, io, tmp.path); defer engine.deinit(); - var spec = try index_spec(gpa, "email", "email_1", true, false); + var spec = try index_spec(gpa, "email", "email_1", true, false, null); defer spec.deinit(); try engine.lock(); _ = try engine.create_index("app", "users", &spec); @@ -919,7 +996,7 @@ test "index maintained across update and delete" { var engine = try Engine.open(gpa, io, tmp.path); defer engine.deinit(); - var spec = try index_spec(gpa, "a", "a_1", false, false); + var spec = try index_spec(gpa, "a", "a_1", false, false, null); defer spec.deinit(); try engine.lock(); _ = try engine.create_index("app", "items", &spec); @@ -969,7 +1046,7 @@ test "index survives reopen and compaction" { var engine = try Engine.open(gpa, io, tmp.path); defer engine.deinit(); engine.compact_threshold = 1; // every write compacts - var spec = try index_spec(gpa, "email", "email_1", false, false); + var spec = try index_spec(gpa, "email", "email_1", false, false, null); defer spec.deinit(); try engine.lock(); _ = try engine.create_index("app", "users", &spec); @@ -1002,7 +1079,7 @@ test "index drop survives reopen" { { var engine = try Engine.open(gpa, io, tmp.path); defer engine.deinit(); - var spec = try index_spec(gpa, "email", "email_1", false, false); + var spec = try index_spec(gpa, "email", "email_1", false, false, null); defer spec.deinit(); try engine.lock(); _ = try engine.create_index("app", "users", &spec); @@ -1032,7 +1109,7 @@ test "drop_collection frees indexes; log without index records replays" { { var engine = try Engine.open(gpa, io, tmp.path); defer engine.deinit(); - var spec = try index_spec(gpa, "email", "email_1", false, false); + var spec = try index_spec(gpa, "email", "email_1", false, false, null); defer spec.deinit(); try engine.lock(); _ = try engine.create_index("app", "users", &spec); @@ -1063,3 +1140,150 @@ test "drop_collection frees indexes; log without index records replays" { try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine2, "app", "users", "email_1", .{ .string = "a@x.io" })); engine2.unlock(); } + +/// A document with an `expireAt` field of any type (omitted when null). +fn doc_with_expire(gpa: std.mem.Allocator, id: i32, expire: ?bson.Value) !bson.Document { + var arena = std.heap.ArenaAllocator.init(gpa); + errdefer arena.deinit(); + const n: usize = if (expire == null) 1 else 2; + const pairs = try arena.allocator().alloc(bson.Pair, n); + pairs[0] = .{ .key = try arena.allocator().dupe(u8, "_id"), .value = .{ .int32 = id } }; + if (expire) |v| { + const value = switch (v) { + .string => |s| bson.Value{ .string = try arena.allocator().dupe(u8, s) }, + else => v, + }; + pairs[1] = .{ .key = try arena.allocator().dupe(u8, "expireAt"), .value = value }; + } + return .{ .arena = arena, .pairs = pairs }; +} + +test "ttl_sweep deletes expired documents and the deletion survives reopen" { + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + var env = test_env(&threaded); + const io = env.io; + const gpa = testing.allocator; + + // A fixed clock: the sweep takes `now` as a parameter precisely so the + // test does not depend on the wall clock. + const now_ms: i64 = 1_700_000_000_000; + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + { + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + var spec = try index_spec(gpa, "expireAt", "expireAt_1", false, false, 60); + defer spec.deinit(); + try engine.lock(); + defer engine.unlock(); + _ = try engine.create_index("app", "sessions", &spec); + + const docs = [_]struct { id: i32, expire: ?bson.Value }{ + .{ .id = 1, .expire = .{ .datetime = now_ms - 120_000 } }, // long expired + .{ .id = 2, .expire = .{ .datetime = now_ms - 60_000 } }, // exactly at the cutoff + .{ .id = 3, .expire = .{ .datetime = now_ms - 30_000 } }, // not yet + .{ .id = 4, .expire = .{ .datetime = now_ms + 3_600_000 } }, // future + .{ .id = 5, .expire = .{ .string = "tomorrow" } }, // not a date: never expires + .{ .id = 6, .expire = null }, // no field: indexed as null + }; + for (docs) |d| { + var doc = try doc_with_expire(gpa, d.id, d.expire); + defer doc.deinit(); + try engine.insert("app", "sessions", &doc, &env.gen); + } + const coll = engine.get_collection("app", "sessions").?; + try testing.expectEqual(@as(usize, 6), coll.docs.count()); + try testing.expectEqual(@as(usize, 6), coll.indexes.items[0].entries.items.len); + + // The cutoff is inclusive: doc 2 goes with doc 1. + try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms)); + try testing.expectEqual(@as(usize, 4), coll.docs.count()); + try testing.expectEqual(@as(usize, 4), coll.indexes.items[0].entries.items.len); + try testing.expectEqual(@as(usize, 0), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .{ .datetime = now_ms - 120_000 })); + // The string and the missing field are untouched by any sweep. + try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .{ .string = "tomorrow" })); + try testing.expectEqual(@as(usize, 1), try index_count(gpa, &engine, "app", "sessions", "expireAt_1", .null)); + + // Idempotent: nothing else is expired at the same instant. + try testing.expectEqual(@as(usize, 0), try engine.ttl_sweep(now_ms)); + // An hour later doc 3 has expired too; doc 4 still has not. + try testing.expectEqual(@as(usize, 1), try engine.ttl_sweep(now_ms + 3_000_000)); + try testing.expectEqual(@as(usize, 3), coll.docs.count()); + } + + // Sweeps go through `remove`, so they are logged: the deletions hold + // across a restart, and the TTL index comes back with its expiry. + var engine2 = try Engine.open(gpa, io, tmp.path); + defer engine2.deinit(); + try engine2.lock(); + defer engine2.unlock(); + const coll = engine2.get_collection("app", "sessions").?; + try testing.expectEqual(@as(usize, 3), coll.docs.count()); + try testing.expectEqual(@as(usize, 1), coll.indexes.items.len); + try testing.expectEqual(@as(?i64, 60), coll.indexes.items[0].ttl); + try testing.expectEqual(@as(usize, 3), coll.indexes.items[0].entries.items.len); + for ([_]i32{ 1, 2, 3 }) |id| { + const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = id }); + defer gpa.free(id_key); + try testing.expect(engine2.get_doc("app", "sessions", id_key) == null); + } + const alive = try bson.serialize_value(gpa, bson.Value{ .int32 = 4 }); + defer gpa.free(alive); + try testing.expect(engine2.get_doc("app", "sessions", alive) != null); +} + +test "ttl_sweep spans collections and several TTL indexes on one collection" { + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + var env = test_env(&threaded); + const io = env.io; + const gpa = testing.allocator; + const now_ms: i64 = 1_700_000_000_000; + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + try engine.lock(); + defer engine.unlock(); + + // Two TTL indexes over the same collection (MongoDB allows this): one + // document is expired by both, and must only be deleted once. + var spec_a = try index_spec(gpa, "expireAt", "expireAt_1", false, false, 60); + defer spec_a.deinit(); + var spec_b = try index_spec(gpa, "seenAt", "seenAt_1", false, false, 10); + defer spec_b.deinit(); + _ = try engine.create_index("app", "sessions", &spec_a); + _ = try engine.create_index("app", "sessions", &spec_b); + + var both = try bson.Document.alloc(gpa, &.{ + .{ .key = "_id", .value = .{ .int32 = 1 } }, + .{ .key = "expireAt", .value = .{ .datetime = now_ms - 120_000 } }, + .{ .key = "seenAt", .value = .{ .datetime = now_ms - 120_000 } }, + }); + defer both.deinit(); + try engine.insert("app", "sessions", &both, &env.gen); + + // A second collection with its own TTL index, and a plain collection + // that no sweep may touch. + var spec_c = try index_spec(gpa, "at", "at_1", false, false, 0); + defer spec_c.deinit(); + _ = try engine.create_index("app", "events", &spec_c); + var ev = try bson.Document.alloc(gpa, &.{ + .{ .key = "_id", .value = .{ .int32 = 2 } }, + // expireAfterSeconds 0: expires at exactly the stored instant. + .{ .key = "at", .value = .{ .datetime = now_ms } }, + }); + defer ev.deinit(); + try engine.insert("app", "events", &ev, &env.gen); + var plain = try make_doc(gpa, 3, "keep"); + defer plain.deinit(); + try engine.insert("other", "plain", &plain, &env.gen); + + try testing.expectEqual(@as(usize, 2), try engine.ttl_sweep(now_ms)); + try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "sessions").?.docs.count()); + try testing.expectEqual(@as(usize, 0), engine.get_collection("app", "events").?.docs.count()); + try testing.expectEqual(@as(usize, 1), engine.get_collection("other", "plain").?.docs.count()); +} diff --git a/src/index.zig b/src/index.zig index 049f89e..03a74ed 100644 --- a/src/index.zig +++ b/src/index.zig @@ -66,15 +66,22 @@ pub const Index = struct { keys: []const IndexKey, unique: bool, sparse: bool, + /// expireAfterSeconds when this is a TTL index, else null. Documents + /// whose indexed value is a datetime older than this many seconds are + /// deleted by Engine.ttl_sweep (src/db.zig); the index itself only + /// carries the setting. Always within + /// [0, max_expire_after_seconds] — parse_spec is the only producer. + ttl: ?i64, multikey: bool, entries: std.ArrayListUnmanaged(Entry), - pub fn init(gpa: std.mem.Allocator, name: []const u8, keys: []const IndexKey, unique: bool, sparse: bool) !Index { + pub fn init(gpa: std.mem.Allocator, name: []const u8, keys: []const IndexKey, unique: bool, sparse: bool, ttl: ?i64) !Index { var self: Index = .{ .name = undefined, .keys = undefined, .unique = unique, .sparse = sparse, + .ttl = ttl, .multikey = false, .entries = .empty, }; @@ -311,8 +318,9 @@ pub const Index = struct { // -- serialization ------------------------------------------------------ - /// The canonical spec document bytes ({v, key, name, unique?, sparse?}) - /// stored in the log and used to rebuild the index on replay. + /// The canonical spec document bytes + /// ({v, key, name, unique?, sparse?, expireAfterSeconds?}) stored in the + /// log and used to rebuild the index on replay. pub fn write_spec(self: *const Index, gpa: std.mem.Allocator, out: *std.ArrayListUnmanaged(u8)) !void { var arena = std.heap.ArenaAllocator.init(gpa); defer arena.deinit(); @@ -333,11 +341,15 @@ pub const Index = struct { try out.append(arena, .{ .key = "name", .value = .{ .string = self.name } }); if (self.unique) try out.append(arena, .{ .key = "unique", .value = .{ .bool = true } }); if (self.sparse) try out.append(arena, .{ .key = "sparse", .value = .{ .bool = true } }); + // int32 like MongoDB: parse_spec caps the value at + // max_expire_after_seconds, so the cast always fits. + if (self.ttl) |secs| try out.append(arena, .{ .key = "expireAfterSeconds", .value = .{ .int32 = @intCast(secs) } }); } pub fn spec_equal(a: *const Index, b: *const Index) bool { if (!std.mem.eql(u8, a.name, b.name)) return false; if (a.unique != b.unique or a.sparse != b.sparse) return false; + if (!std.meta.eql(a.ttl, b.ttl)) return false; if (a.keys.len != b.keys.len) return false; for (a.keys, b.keys) |ka, kb| { if (!std.mem.eql(u8, ka.path, kb.path)) return false; @@ -365,10 +377,10 @@ pub fn find_by_key_pattern(indexes: []const Index, key_pairs: []const bson.Pair) return null; } -pub const SpecError = error{ InvalidIndexSpec, OutOfMemory }; +pub const SpecError = error{ InvalidIndexSpec, TtlOnCompoundIndex, InvalidExpireAfterSeconds, OutOfMemory }; -/// Parse {key: {...}, name?, unique?, sparse?} from a spec document — the -/// form drivers send and the form the log stores. +/// Parse {key: {...}, name?, unique?, sparse?, expireAfterSeconds?} from a +/// spec document — the form drivers send and the form the log stores. pub fn parse_spec(gpa: std.mem.Allocator, spec: *const bson.Document) SpecError!Index { const key_value = bson.get_pair(spec.pairs, "key") orelse return error.InvalidIndexSpec; const key_pairs = switch (key_value) { @@ -391,16 +403,48 @@ pub fn parse_spec(gpa: std.mem.Allocator, spec: *const bson.Document) SpecError! var sparse = false; if (bson.get_pair(spec.pairs, "unique")) |v| unique = query.truthy(v); if (bson.get_pair(spec.pairs, "sparse")) |v| sparse = query.truthy(v); + var ttl: ?i64 = null; + if (bson.get_pair(spec.pairs, "expireAfterSeconds")) |v| { + ttl = try expire_after_seconds(v); + // MongoDB's rule: TTL is a single-field index option. A compound key + // would leave it ambiguous which component dates the document. + if (key_pairs.len != 1) return error.TtlOnCompoundIndex; + } const name_value = bson.get_pair(spec.pairs, "name") orelse { const nm = try default_name(gpa, key_pairs); defer gpa.free(nm); - return Index.init(gpa, nm, keys[0..key_pairs.len], unique, sparse); + return Index.init(gpa, nm, keys[0..key_pairs.len], unique, sparse, ttl); }; const name = switch (name_value) { .string => |s| s, else => return error.InvalidIndexSpec, }; - return Index.init(gpa, name, keys[0..key_pairs.len], unique, sparse); + return Index.init(gpa, name, keys[0..key_pairs.len], unique, sparse, ttl); +} + +/// MongoDB's bound on expireAfterSeconds. Keeping it means a TTL always +/// round-trips as an int32, exactly as a real server reports it. +pub const max_expire_after_seconds: i64 = 2147483647; + +/// The seconds in an `expireAfterSeconds` option: integral and within +/// [0, max_expire_after_seconds]. 0 is legal (expire at the stored instant); +/// a negative, fractional, out-of-range or non-numeric value is not. A +/// double is accepted because that is what a JavaScript driver sends for a +/// plain number — and the range check runs before the conversion, so +/// @intFromFloat is always defined. +fn expire_after_seconds(v: bson.Value) error{InvalidExpireAfterSeconds}!i64 { + const secs: i64 = switch (v) { + .int32 => |n| n, + .int64 => |n| n, + .double => |d| blk: { + if (!std.math.isFinite(d) or @trunc(d) != d) return error.InvalidExpireAfterSeconds; + if (d < 0 or d > @as(f64, @floatFromInt(max_expire_after_seconds))) return error.InvalidExpireAfterSeconds; + break :blk @intFromFloat(d); + }, + else => return error.InvalidExpireAfterSeconds, + }; + if (secs < 0 or secs > max_expire_after_seconds) return error.InvalidExpireAfterSeconds; + return secs; } /// Whether a key-pattern direction value means descending. The single @@ -851,7 +895,7 @@ fn doc_of(pairs: []const bson.Pair) bson.Document { fn simple_index(gpa: std.mem.Allocator, paths: []const []const u8, unique: bool, sparse: bool) !Index { var keys: [max_index_keys]IndexKey = undefined; for (paths, 0..) |p, i| keys[i] = .{ .path = p, .descending = false }; - return Index.init(gpa, "test", keys[0..paths.len], unique, sparse); + return Index.init(gpa, "test", keys[0..paths.len], unique, sparse, null); } /// Look up ids and compare with the expected set. Entry ids alias the @@ -1103,6 +1147,119 @@ test "id fast path guards and $in" { try testing.expect(plan_id(&doc_safe) != null); } +test "TTL spec round-trips through write_spec and compares in spec_equal" { + const gpa = testing.allocator; + const spec = doc_of(&.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ .key = "name", .value = .{ .string = "expireAt_1" } }, + // A driver sends a plain JS number as a double. + .{ .key = "expireAfterSeconds", .value = .{ .double = 60.0 } }, + }); + var ix = try parse_spec(gpa, &spec); + defer ix.deinit(gpa); + try testing.expectEqual(@as(?i64, 60), ix.ttl); + + // Serialize and reparse: the log/compaction path. + var bytes: std.ArrayListUnmanaged(u8) = .empty; + defer bytes.deinit(gpa); + try ix.write_spec(gpa, &bytes); + var reparsed_doc = try bson.Document.parse(gpa, bytes.items); + defer reparsed_doc.deinit(); + try testing.expectEqual(@as(i32, 60), reparsed_doc.get("expireAfterSeconds").?.int32); + var ix2 = try parse_spec(gpa, &reparsed_doc); + defer ix2.deinit(gpa); + try testing.expect(Index.spec_equal(&ix, &ix2)); + + // Same name and key, different expiry: not the same spec (the command + // layer turns this into IndexOptionsConflict). + const other = doc_of(&.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ .key = "name", .value = .{ .string = "expireAt_1" } }, + .{ .key = "expireAfterSeconds", .value = .{ .int32 = 90 } }, + }); + var ix3 = try parse_spec(gpa, &other); + defer ix3.deinit(gpa); + try testing.expect(!Index.spec_equal(&ix, &ix3)); + + // The largest legal expiry, arriving as an int64, still round-trips as + // an int32 — the only type this ever emits. + const big = doc_of(&.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ .key = "expireAfterSeconds", .value = .{ .int64 = max_expire_after_seconds } }, + }); + var ix_big = try parse_spec(gpa, &big); + defer ix_big.deinit(gpa); + bytes.clearRetainingCapacity(); + try ix_big.write_spec(gpa, &bytes); + var big_doc = try bson.Document.parse(gpa, bytes.items); + defer big_doc.deinit(); + try testing.expectEqual(@as(i32, 2147483647), big_doc.get("expireAfterSeconds").?.int32); + var ix_big2 = try parse_spec(gpa, &big_doc); + defer ix_big2.deinit(gpa); + try testing.expectEqual(@as(?i64, max_expire_after_seconds), ix_big2.ttl); + + // No option at all: ttl null, and nothing emitted (old logs reparse + // unchanged). + const plain = doc_of(&.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + }); + var ix4 = try parse_spec(gpa, &plain); + defer ix4.deinit(gpa); + try testing.expect(ix4.ttl == null); + bytes.clearRetainingCapacity(); + try ix4.write_spec(gpa, &bytes); + var plain_doc = try bson.Document.parse(gpa, bytes.items); + defer plain_doc.deinit(); + try testing.expect(plain_doc.get("expireAfterSeconds") == null); +} + +test "TTL spec rejects compound keys and bad expireAfterSeconds" { + const gpa = testing.allocator; + const compound = doc_of(&.{ + .{ .key = "key", .value = .{ .doc = &.{ + .{ .key = "a", .value = .{ .int32 = 1 } }, + .{ .key = "b", .value = .{ .int32 = 1 } }, + } } }, + .{ .key = "expireAfterSeconds", .value = .{ .int32 = 60 } }, + }); + try testing.expectError(error.TtlOnCompoundIndex, parse_spec(gpa, &compound)); + + const bad = [_]bson.Value{ + .{ .int32 = -1 }, + .{ .int64 = -1 }, + .{ .double = -0.5 }, + .{ .double = 1.5 }, + // Past MongoDB's bound, as an int and as a double (1e19 would also + // overflow the conversion, which the range check runs before). + .{ .int64 = max_expire_after_seconds + 1 }, + .{ .double = 3e9 }, + .{ .double = 1e19 }, + .{ .double = std.math.inf(f64) }, + .{ .double = std.math.nan(f64) }, + .{ .string = "60" }, + .{ .bool = true }, + .null, + }; + for (bad) |v| { + const spec = doc_of(&.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ .key = "expireAfterSeconds", .value = v }, + }); + try testing.expectError(error.InvalidExpireAfterSeconds, parse_spec(gpa, &spec)); + } + + // 0 is legal: expire at exactly the stored instant. + const zero = doc_of(&.{ + .{ .key = "key", .value = .{ .doc = &.{.{ .key = "expireAt", .value = .{ .int32 = 1 } }} } }, + .{ .key = "expireAfterSeconds", .value = .{ .int32 = 0 } }, + }); + var ix = try parse_spec(gpa, &zero); + defer ix.deinit(gpa); + try testing.expectEqual(@as(?i64, 0), ix.ttl); + // The default name still comes from the key pattern. + try testing.expectEqualStrings("expireAt_1", ix.name); +} + test "planner picks eq run, ranges, and bails on sparse null" { const gpa = testing.allocator; var ix = try simple_index(gpa, &.{ "a", "b" }, false, false); diff --git a/src/main.zig b/src/main.zig index af6f7f5..663002d 100644 --- a/src/main.zig +++ b/src/main.zig @@ -8,6 +8,8 @@ const usage = \\ --port listen port (default 27017) \\ --bind bind address (default 127.0.0.1) \\ --db database file (default mongo-light.log) + \\ --ttl-sweep-secs + \\ seconds between TTL index sweeps (default 60, 0 disables) \\ --help show this help \\ ; @@ -16,6 +18,7 @@ pub fn main(init: std.process.Init) !void { var port: u16 = 27017; var bind_ip: []const u8 = "127.0.0.1"; var db_path: []const u8 = "mongo-light.log"; + var ttl_sweep_secs: i64 = 60; var it = std.process.Args.Iterator.init(init.minimal.args); defer it.deinit(); @@ -31,6 +34,16 @@ pub fn main(init: std.process.Init) !void { bind_ip = it.next() orelse return error.MissingValue; } else if (std.mem.eql(u8, arg, "--db")) { db_path = it.next() orelse return error.MissingValue; + } else if (std.mem.eql(u8, arg, "--ttl-sweep-secs")) { + const v = it.next() orelse return error.MissingValue; + // i64 is the width std.Io.Duration.fromSeconds takes, so the + // value reaches the sweeper without a cast; negatives are the + // only thing parseInt would otherwise let through. + ttl_sweep_secs = std.fmt.parseInt(i64, v, 10) catch -1; + if (ttl_sweep_secs < 0) { + std.debug.print("mongo-light: invalid ttl sweep interval '{s}'\n", .{v}); + return error.InvalidTtlSweepSecs; + } } else if (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h")) { try std.Io.File.writeStreamingAll(.stdout(), init.io, usage); return; @@ -53,6 +66,7 @@ pub fn main(init: std.process.Init) !void { .connection_counter = .init(1), .engine = &engine, .start_time = std.Io.Timestamp.now(init.io, .real), + .ttl_sweep_secs = ttl_sweep_secs, }; try server.run(); } diff --git a/src/server.zig b/src/server.zig index 1f6f35b..1a9e966 100644 --- a/src/server.zig +++ b/src/server.zig @@ -16,6 +16,10 @@ pub const Server = struct { connection_counter: std.atomic.Value(u32), engine: *db.Engine, start_time: std.Io.Timestamp, + /// Seconds between TTL sweeps; 0 leaves the monitor unspawned. Signed + /// because that is what std.Io.Duration.fromSeconds takes — the CLI + /// rejects negatives. + ttl_sweep_secs: i64, pub fn run(self: *Server) !void { // Unbounded async limit: connection handlers otherwise fall back to @@ -36,6 +40,10 @@ pub const Server = struct { var group: std.Io.Group = .init; defer group.cancel(io); + // The TTL monitor is just another member of the connection group, so + // the `group.cancel` above stops it with everything else. + if (self.ttl_sweep_secs > 0) group.async(io, ttl_monitor, .{ io, self }); + while (true) { const stream = listener.accept(io) catch |err| switch (err) { error.Canceled => return, @@ -49,6 +57,27 @@ pub const Server = struct { } }; +/// Expire documents under TTL indexes every `ttl_sweep_secs` seconds, until +/// the group is canceled. Sweeping takes the engine's write lock, so it is +/// serialized with commands exactly like any other write; a sweep failure is +/// logged rather than fatal, since the next one will retry. +fn ttl_monitor(io: std.Io, server: *Server) error{Canceled}!void { + const interval: std.Io.Duration = .fromSeconds(server.ttl_sweep_secs); + while (true) { + // Sleep first: at startup the engine has just replayed the log, and + // an immediate sweep would race the listener's first connections for + // the write lock. + try std.Io.sleep(io, interval, .awake); + try server.engine.lock(); + defer server.engine.unlock(); + const now_ms = std.Io.Timestamp.now(io, .real).toMilliseconds(); + _ = server.engine.ttl_sweep(now_ms) catch |err| { + std.debug.print("mongo-light: TTL sweep failed: {s}\n", .{@errorName(err)}); + continue; + }; + } +} + /// Entry point required by `Group.async`: must return only `error.Canceled`. fn handle_connection(io: std.Io, stream: std.Io.net.Stream, server: *Server) error{Canceled}!void { handle_connection_inner(io, stream, server) catch {}; diff --git a/tests/e2e/README.md b/tests/e2e/README.md index a4b0e3d..8c223e3 100644 --- a/tests/e2e/README.md +++ b/tests/e2e/README.md @@ -17,13 +17,24 @@ Start the server, then run the suites against it (defaults to port 27020): ```sh zig build -zig-out/bin/mongo-light --port 27020 --db /tmp/ml-e2e.log & +zig-out/bin/mongo-light --port 27020 --db /tmp/ml-e2e.log --ttl-sweep-secs 1 & node tests/e2e/e2e.js # CRUD + operators + aggregate + errors (29 checks) node tests/e2e/e2e2.js concurrent # 8 clients: 4 writers + 4 readers (2 checks) node tests/e2e/e2e2.js crash-a # write 50 docs, then kill -9 the server node tests/e2e/e2e2.js crash-b # restart and verify all 50 survived +node tests/e2e/e2e3.js # secondary indexes: unique/sparse/compound (16 checks) +node tests/e2e/e2e4.js # TTL indexes: expiry + rejected specs (15 checks) ``` +`e2e4.js` needs the server started with `--ttl-sweep-secs 1` (the default is +60 seconds, which is longer than the suite waits); the other suites do not +care about the flag. + +Rebuild with `zig build` after any change under `src/` before restarting the +server: `zig build test` compiles the test binary only and leaves +`zig-out/bin/mongo-light` stale, so the suites keep running against the old +rules and report failures that the source no longer explains. + `e2e2.js concurrent` is safe to repeat against a running server (it drops its collection first); `crash-a`/`crash-b` are two halves of one scenario. diff --git a/tests/e2e/e2e4.js b/tests/e2e/e2e4.js new file mode 100644 index 0000000..cb01e7b --- /dev/null +++ b/tests/e2e/e2e4.js @@ -0,0 +1,138 @@ +// E2E part 4: TTL indexes through the official Node driver. +// createIndex({expireAfterSeconds}), the option round-tripping through +// listIndexes, background expiry of past-dated documents, and the specs the +// server must reject with CannotCreateIndex (67). +// +// The server must run with a short sweep interval, e.g. +// zig-out/bin/mongo-light --port 27020 --db /tmp/ml-e2e.log --ttl-sweep-secs 1 +const { MongoClient } = require('mongodb'); + +const URL = 'mongodb://127.0.0.1:27020'; +const results = []; +function check(name, cond, detail = '') { + results.push({ name, ok: !!cond, detail: String(detail) }); + if (!cond) console.error(` ✗ ${name} ${detail}`); +} + +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +/// Poll until `fn()` is true or the deadline passes; expiry is a background +/// sweep, so the exact moment a document disappears is not fixed. +async function waitFor(fn, timeoutMs = 15000, stepMs = 250) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await fn()) return true; + await sleep(stepMs); + } + return false; +} + +async function expectCode(fn, code, name) { + let err = null; + try { + await fn(); + } catch (e) { + err = e; + } + check(name, err && err.code === code, err ? `code ${err.code}: ${err.message}` : 'no error'); +} + +async function main() { + const client = new MongoClient(URL, { serverSelectionTimeoutMS: 5000 }); + await client.connect(); + const db = client.db('e2e4'); + const coll = db.collection('sessions'); + await coll.drop().catch(() => {}); + + // --- createIndex + listIndexes round-trip --- + await coll.createIndex({ expireAt: 1 }, { expireAfterSeconds: 1 }); + const idxs = await coll.indexes(); + const ttlIdx = idxs.find((i) => i.name === 'expireAt_1'); + check('TTL index listed', !!ttlIdx, JSON.stringify(idxs)); + check('expireAfterSeconds reported', ttlIdx && Number(ttlIdx.expireAfterSeconds) === 1, JSON.stringify(ttlIdx)); + check('_id_ has no expireAfterSeconds', idxs.find((i) => i.name === '_id_').expireAfterSeconds === undefined); + + // Idempotent re-create of the same spec. + await coll.createIndex({ expireAt: 1 }, { expireAfterSeconds: 1 }); + check('idempotent re-create', (await coll.indexes()).filter((i) => i.name === 'expireAt_1').length === 1); + + // --- expiry --- + const now = Date.now(); + await coll.insertMany([ + { _id: 'past', expireAt: new Date(now - 60_000) }, + { _id: 'future', expireAt: new Date(now + 3_600_000) }, + { _id: 'string', expireAt: 'tomorrow' }, + { _id: 'missing' }, + // An array of dates expires on its earliest member. + { _id: 'array', expireAt: [new Date(now - 60_000), new Date(now + 3_600_000)] }, + ]); + check('all five inserted', (await coll.countDocuments({})) === 5); + + const gone = await waitFor(async () => (await coll.countDocuments({ _id: 'past' })) === 0); + check('past-dated doc expired', gone); + check('array doc expired on earliest date', (await coll.countDocuments({ _id: 'array' })) === 0); + const survivors = (await coll.find({}).toArray()).map((d) => d._id).sort(); + check('future/string/missing survive', JSON.stringify(survivors) === '["future","missing","string"]', JSON.stringify(survivors)); + + // Expiry is a real delete: it holds after the sweeper has run again. + await sleep(1500); + check('expired docs stay deleted', (await coll.countDocuments({})) === 3); + + // --- rejected specs --- + const bad = db.collection('bad'); + await expectCode( + () => bad.createIndex({ a: 1, b: 1 }, { expireAfterSeconds: 60 }), + 67, + 'compound TTL rejected with 67', + ); + await expectCode( + () => bad.createIndex({ a: 1 }, { expireAfterSeconds: -1 }), + 67, + 'negative expireAfterSeconds rejected with 67', + ); + await expectCode( + () => bad.createIndex({ a: 1 }, { expireAfterSeconds: 'soon' }), + 67, + 'non-numeric expireAfterSeconds rejected with 67', + ); + await expectCode( + () => bad.createIndex({ a: 1 }, { expireAfterSeconds: 2147483648 }), + 67, + 'expireAfterSeconds past 2147483647 rejected with 67', + ); + await expectCode( + () => bad.createIndex({ _id: 1 }, { expireAfterSeconds: 60 }), + 197, + 'TTL on _id rejected with 197', + ); + // Same name and key with a different expiry: IndexOptionsConflict (85), + // exactly as MongoDB does (the change belongs to collMod). + await expectCode( + () => coll.createIndex({ expireAt: 1 }, { expireAfterSeconds: 90 }), + 85, + 'changed expiry gives IndexOptionsConflict', + ); + + // expireAfterSeconds 0 is legal: expire at exactly the stored instant. + const zero = db.collection('zero'); + await zero.drop().catch(() => {}); + await zero.createIndex({ at: 1 }, { expireAfterSeconds: 0 }); + await zero.insertOne({ _id: 'now', at: new Date(Date.now() - 1000) }); + check('expireAfterSeconds 0 accepted', (await zero.indexes()).some((i) => Number(i.expireAfterSeconds) === 0)); + check('expireAfterSeconds 0 expires', await waitFor(async () => (await zero.countDocuments({})) === 0)); + + await client.close(); + + const failed = results.filter((r) => !r.ok); + console.log(`\n${results.length - failed.length}/${results.length} checks passed`); + if (failed.length) { + console.log('FAILED:', failed.map((f) => f.name).join(', ')); + process.exit(1); + } + console.log('E2E4_OK'); +} + +main().catch((e) => { + console.error('E2E4_FAIL', e); + process.exit(1); +});