From f61642313159172ed23e1879a5f531dbed415931 Mon Sep 17 00:00:00 2001 From: "William K. Santiago" Date: Fri, 17 Jul 2026 11:16:46 -0400 Subject: [PATCH 1/5] Bound query scan to prevent full-DB page-fault thrash --- src/config.zig | 10 ++++++++++ src/main.zig | 3 ++- src/spider.zig | 4 ++-- src/store.zig | 22 ++++++++++++++++++++-- 4 files changed, 34 insertions(+), 5 deletions(-) diff --git a/src/config.zig b/src/config.zig index 63b4a1a..7fed207 100644 --- a/src/config.zig +++ b/src/config.zig @@ -25,6 +25,10 @@ pub const Config = struct { max_content_length: u32, query_limit_default: u32, query_limit_max: u32, + // Bounds the entries a single query may scan to `limit * query_scan_multiplier` + // before stopping, so a selective filter (few matches) cannot page-fault the + // entire event DB looking for `limit` results. 0 disables the cap. + query_scan_multiplier: u32, max_event_age: i64, max_future_seconds: i64, storage_path: []const u8, @@ -90,6 +94,7 @@ pub const Config = struct { .max_content_length = 102400, .query_limit_default = 500, .query_limit_max = 5000, + .query_scan_multiplier = 20, .max_event_age = 94608000, .max_future_seconds = 900, .storage_path = "./data", @@ -201,6 +206,8 @@ pub const Config = struct { self.query_limit_default = try std.fmt.parseInt(u32, value, 10); } else if (std.mem.eql(u8, key, "query_limit_max")) { self.query_limit_max = try std.fmt.parseInt(u32, value, 10); + } else if (std.mem.eql(u8, key, "query_scan_multiplier")) { + self.query_scan_multiplier = try std.fmt.parseInt(u32, value, 10); } else if (std.mem.eql(u8, key, "max_event_age")) { self.max_event_age = try std.fmt.parseInt(i64, value, 10); } else if (std.mem.eql(u8, key, "max_future_seconds")) { @@ -316,6 +323,9 @@ pub const Config = struct { if (getenv("WISP_QUERIES_PER_MINUTE")) |v| { self.queries_per_minute = std.fmt.parseInt(u32, v, 10) catch self.queries_per_minute; } + if (getenv("WISP_QUERY_SCAN_MULTIPLIER")) |v| { + self.query_scan_multiplier = std.fmt.parseInt(u32, v, 10) catch self.query_scan_multiplier; + } if (getenv("WISP_IDLE_SECONDS")) |v| { self.idle_seconds = std.fmt.parseInt(u32, v, 10) catch self.idle_seconds; } diff --git a/src/main.zig b/src/main.zig index 671c213..b600323 100644 --- a/src/main.zig +++ b/src/main.zig @@ -164,6 +164,7 @@ pub fn main(init: std.process.Init) !void { defer lmdb.deinit(); var store = try Store.init(allocator, &lmdb); + store.query_scan_multiplier = config.query_scan_multiplier; defer store.deinit(); var mgmt_store = try ManagementStore.init(allocator, &lmdb); @@ -405,7 +406,7 @@ fn runExport(allocator: std.mem.Allocator, db_path: []const u8) !void { const stderr_file = stdFile(std.posix.STDERR_FILENO); const empty_filters = [_]nostr.Filter{}; - var iter = try store.query(&empty_filters, std.math.maxInt(u32)); + var iter = try store.queryFull(&empty_filters, std.math.maxInt(u32)); defer iter.deinit(); var exported: u64 = 0; diff --git a/src/spider.zig b/src/spider.zig index fc5f65c..34be299 100644 --- a/src/spider.zig +++ b/src/spider.zig @@ -238,7 +238,7 @@ pub const Spider = struct { .limit_val = 1, }}; - var iter = self.store.query(&filters, 1) catch return false; + var iter = self.store.queryFull(&filters, 1) catch return false; defer iter.deinit(); const json = (iter.next() catch return false) orelse return false; @@ -557,7 +557,7 @@ pub const Spider = struct { .{ .authors_bytes = pubkeys }, }; - var iter = self.store.query(&filters, 100000) catch { + var iter = self.store.queryFull(&filters, 100000) catch { log.err("{s}: Failed to query local events for negentropy", .{relay_url}); return true; }; diff --git a/src/store.zig b/src/store.zig index 1ecfb34..25bca60 100644 --- a/src/store.zig +++ b/src/store.zig @@ -10,6 +10,10 @@ pub const Store = struct { lmdb: *Lmdb, allocator: std.mem.Allocator, query_cache: QueryCache, + // Caps QueryIterator scanning at `limit * query_scan_multiplier` entries. Set + // from config on the serving path; 0 disables. Defaults to 20 (matches + // queryMultiKind) so tooling that builds a Store directly stays bounded. + query_scan_multiplier: u32 = 20, events: Dbi, idx_created: Dbi, @@ -360,7 +364,13 @@ pub const Store = struct { } pub fn query(self: *Store, filters: []const nostr.Filter, limit: u32) !QueryIterator { - return QueryIterator.init(self, filters, limit); + return QueryIterator.init(self, filters, limit, self.query_scan_multiplier); + } + + // Uncapped scan for trusted internal callers (spider lookups, full export) + // that must find every match regardless of how selective the filter is. + pub fn queryFull(self: *Store, filters: []const nostr.Filter, limit: u32) !QueryIterator { + return QueryIterator.init(self, filters, limit, 0); } pub fn queryMultiKind(self: *Store, kinds: []const i32, limit: u32) !MultiKindResult { @@ -494,6 +504,10 @@ pub const QueryIterator = struct { filters: []const nostr.Filter, limit: u32, returned: u32 = 0, + scanned: u32 = 0, + // Stop after visiting this many index entries even if fewer than `limit` + // matched, so a selective filter cannot fault the whole DB. 0 disables. + max_scan: u32 = 0, txn: ?Txn = null, cursor: ?Cursor = null, started: bool = false, @@ -504,11 +518,12 @@ pub const QueryIterator = struct { const IndexType = enum { created, kind, pubkey, tag }; - pub fn init(store: *Store, filters: []const nostr.Filter, limit: u32) QueryIterator { + pub fn init(store: *Store, filters: []const nostr.Filter, limit: u32, scan_multiplier: u32) QueryIterator { var iter = QueryIterator{ .store = store, .filters = filters, .limit = limit, + .max_scan = if (scan_multiplier == 0) 0 else limit *| scan_multiplier, }; if (filters.len > 0) { @@ -614,6 +629,7 @@ pub const QueryIterator = struct { { return null; } + self.scanned += 1; if (try self.processEntry(entry)) |json| { return json; } @@ -623,7 +639,9 @@ pub const QueryIterator = struct { } while (true) { + if (self.max_scan != 0 and self.scanned >= self.max_scan) return null; const entry = try self.cursor.?.get(.prev) orelse return null; + self.scanned += 1; if (self.prefix_len > 0) { if (entry.key.len < self.prefix_len or From 53daa1dc1d4b9e253fcfff2ba8f703acf254f8ab Mon Sep 17 00:00:00 2001 From: "William K. Santiago" Date: Fri, 17 Jul 2026 11:48:53 -0400 Subject: [PATCH 2/5] Add ids fast-path, honor scan-multiplier config, and cover with tests --- docs/configuration.md | 1 + src/handler.zig | 6 ++ src/main.zig | 1 + src/store.zig | 169 +++++++++++++++++++++++++++++++++++++++++- wisp.toml.example | 2 + 5 files changed, 175 insertions(+), 4 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 39e05ac..279f6dd 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -103,6 +103,7 @@ write path, since there is no fsync to amortize. | `max_content_length` | — | u32 | `102400` | Maximum event `content` length, in bytes. | | `query_limit_default` | — | u32 | `500` | Events returned per REQ when the client sets no `limit`. | | `query_limit_max` | — | u32 | `5000` | Hard cap on events returned per REQ. | +| `query_scan_multiplier` | `WISP_QUERY_SCAN_MULTIPLIER` | u32 | `20` | Caps entries scanned per query at `limit × multiplier` before stopping, so a selective filter cannot page-fault the whole event DB. `0` disables the cap. | | `max_event_age` | — | i64 (seconds) | `94608000` | Reject events whose `created_at` is older than this (default 3 years). | | `max_future_seconds` | — | i64 (seconds) | `900` | Reject events dated more than this far in the future (default 15 minutes). | | `min_pow_difficulty` | `WISP_MIN_POW_DIFFICULTY` | u8 | `0` | Required NIP-13 proof-of-work leading-zero bits. `0` disables. | diff --git a/src/handler.zig b/src/handler.zig index 9b8f696..60a43bd 100644 --- a/src/handler.zig +++ b/src/handler.zig @@ -584,6 +584,9 @@ pub const Handler = struct { var total_count: u64 = 0; for (filters) |filter| { + // Uses the capped query() intentionally: for a selective filter this + // count is approximate (bounded by the scan cap) rather than exact, + // which is the accepted DoS tradeoff for network-reachable COUNT. var iter = self.store.query(&[_]nostr.Filter{filter}, self.config.query_limit_max) catch { self.sendClosed(conn, sub_id, "error: query failed"); return; @@ -708,6 +711,9 @@ pub const Handler = struct { }; if (self.shutdown.load(.acquire)) return; + // Serving-side enumeration uses the capped query() by design: this path is + // network-reachable, so reconciliation stays DoS-safe at the cost of + // possibly under-enumerating on a pathologically large DB. var iter = self.store.query(&[_]nostr.Filter{filter}, self.config.negentropy_max_sync_events) catch { conn.removeNegSession(sub_id); self.sendNegErr(conn, sub_id, "error: query failed"); diff --git a/src/main.zig b/src/main.zig index b600323..8fbceb7 100644 --- a/src/main.zig +++ b/src/main.zig @@ -434,4 +434,5 @@ test { _ = @import("server.zig"); _ = @import("relay_metrics.zig"); _ = @import("subscriptions.zig"); + _ = @import("store.zig"); } diff --git a/src/store.zig b/src/store.zig index 25bca60..06a5e3e 100644 --- a/src/store.zig +++ b/src/store.zig @@ -373,6 +373,10 @@ pub const Store = struct { return QueryIterator.init(self, filters, limit, 0); } + fn scanCap(limit: u32, multiplier: u32) u32 { + return if (multiplier == 0) 0 else limit *| multiplier; + } + pub fn queryMultiKind(self: *Store, kinds: []const i32, limit: u32) !MultiKindResult { var txn = try self.lmdb.beginTxn(true); errdefer txn.abort(); @@ -396,9 +400,9 @@ pub const Store = struct { var entry = try cursor.get(.last); var collected: u32 = 0; var scanned: u32 = 0; - const max_scan: u32 = limit * 20; + const max_scan = scanCap(limit, self.query_scan_multiplier); - while (entry != null and collected < limit and scanned < max_scan) : (entry = try cursor.get(.prev)) { + while (entry != null and collected < limit and (max_scan == 0 or scanned < max_scan)) : (entry = try cursor.get(.prev)) { const e = entry.?; scanned += 1; @@ -515,20 +519,33 @@ pub const QueryIterator = struct { prefix: [44]u8 = undefined, prefix_len: usize = 0, skip_filter: bool = false, + ids: ?[][32]u8 = null, + ids_index: usize = 0, - const IndexType = enum { created, kind, pubkey, tag }; + const IndexType = enum { created, kind, pubkey, tag, ids }; pub fn init(store: *Store, filters: []const nostr.Filter, limit: u32, scan_multiplier: u32) QueryIterator { var iter = QueryIterator{ .store = store, .filters = filters, .limit = limit, - .max_scan = if (scan_multiplier == 0) 0 else limit *| scan_multiplier, + .max_scan = Store.scanCap(limit, scan_multiplier), }; if (filters.len > 0) { const f = filters[0]; + // Direct point lookups by id are inherently bounded (id list is + // capped by max message size), so they bypass the scan cap and + // return matches regardless of age. + if (f.ids()) |id_list| { + if (id_list.len > 0) { + iter.index_type = .ids; + iter.ids = id_list; + return iter; + } + } + if (f.authors()) |authors| { if (authors.len == 1) { iter.index_type = .pubkey; @@ -578,6 +595,8 @@ pub const QueryIterator = struct { pub fn next(self: *QueryIterator) !?[]const u8 { if (self.returned >= self.limit) return null; + if (self.index_type == .ids) return self.nextIds(); + if (self.txn == null) { self.txn = try self.store.lmdb.beginTxn(true); const dbi = switch (self.index_type) { @@ -585,6 +604,7 @@ pub const QueryIterator = struct { .pubkey => self.store.idx_pubkey, .tag => self.store.idx_tag, .created => self.store.idx_created, + .ids => unreachable, }; self.cursor = try self.txn.?.cursor(dbi); } @@ -615,6 +635,7 @@ pub const QueryIterator = struct { break :blk try self.cursor.?.get(.last); } }, + .ids => unreachable, else => try self.cursor.?.get(.last), }; @@ -657,6 +678,29 @@ pub const QueryIterator = struct { } } + fn nextIds(self: *QueryIterator) !?[]const u8 { + if (self.txn == null) self.txn = try self.store.lmdb.beginTxn(true); + const id_list = self.ids orelse return null; + + while (self.ids_index < id_list.len) { + const id = &id_list[self.ids_index]; + self.ids_index += 1; + + const json = try self.txn.?.get(self.store.events, id) orelse continue; + + var event = nostr.Event.parse(json) catch continue; + defer event.deinit(); + + if (nostr.isExpired(&event)) continue; + + if (self.filters.len == 0 or nostr.filtersMatch(self.filters, &event)) { + self.returned += 1; + return json; + } + } + return null; + } + const Entry = @import("lmdb.zig").Entry; fn processEntry(self: *QueryIterator, entry: Entry) !?[]const u8 { const event_id = switch (self.index_type) { @@ -676,6 +720,7 @@ pub const QueryIterator = struct { if (entry.key.len < 40) return null; break :blk entry.key[8..40]; }, + .ids => unreachable, }; const json = try self.txn.?.get(self.store.events, event_id) orelse return null; @@ -703,3 +748,119 @@ pub const QueryIterator = struct { if (self.txn) |*t| t.abort(); } }; + +const testing = std.testing; + +fn testIdHex(n: u32, buf: *[64]u8) void { + @memset(buf, '0'); + _ = std.fmt.bufPrint(buf[56..64], "{x:0>8}", .{n}) catch unreachable; +} + +fn storeTestEvent(s: *Store, alloc: std.mem.Allocator, id_hex: []const u8, pubkey_hex: []const u8, created_at: i64) !void { + const sig_hex = "0" ** 128; + const json = try std.fmt.allocPrint( + alloc, + "{{\"id\":\"{s}\",\"pubkey\":\"{s}\",\"sig\":\"{s}\",\"kind\":1,\"created_at\":{d},\"content\":\"x\",\"tags\":[]}}", + .{ id_hex, pubkey_hex, sig_hex, created_at }, + ); + defer alloc.free(json); + var event = try nostr.Event.parse(json); + defer event.deinit(); + _ = try s.store(&event, json); +} + +test "scanCap disables at 0 and saturates without wrapping" { + try testing.expectEqual(@as(u32, 0), Store.scanCap(1000, 0)); + try testing.expectEqual(@as(u32, 200), Store.scanCap(10, 20)); + try testing.expectEqual(std.math.maxInt(u32), Store.scanCap(std.math.maxInt(u32), 20)); +} + +test "query scan cap stops at limit times multiplier" { + const alloc = testing.allocator; + const io = nostr.io.io(); + const dir = "./.test-scan-cap-a"; + std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + defer std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + + var lmdb = try Lmdb.init(alloc, dir ++ "/db.mdb", 256, .none); + defer lmdb.deinit(); + var s = try Store.init(alloc, &lmdb); + defer s.deinit(); + + const base = nostr.io.timestamp() - 100000; + const filler_pk = "aa" ** 32; + var i: u32 = 0; + while (i < 20) : (i += 1) { + var idb: [64]u8 = undefined; + testIdHex(i + 1, &idb); + try storeTestEvent(&s, alloc, &idb, filler_pk, base + @as(i64, i)); + } + + s.query_scan_multiplier = 3; + var authors = [_][32]u8{ [_]u8{0xcc} ** 32, [_]u8{0xdd} ** 32 }; + const filters = [_]nostr.Filter{.{ .authors_bytes = &authors }}; + var iter = try s.query(&filters, 5); + defer iter.deinit(); + + var count: u32 = 0; + while (try iter.next()) |_| count += 1; + try testing.expectEqual(@as(u32, 0), count); + try testing.expectEqual(@as(u32, 15), iter.scanned); +} + +test "queryFull and ids fast-path bypass the scan cap for old matches" { + const alloc = testing.allocator; + const io = nostr.io.io(); + const dir = "./.test-scan-cap-b"; + std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + defer std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + + var lmdb = try Lmdb.init(alloc, dir ++ "/db.mdb", 256, .none); + defer lmdb.deinit(); + var s = try Store.init(alloc, &lmdb); + defer s.deinit(); + + const base = nostr.io.timestamp() - 100000; + const target_pk = "bb" ** 32; + const filler_pk = "aa" ** 32; + + var target_id_hex: [64]u8 = undefined; + testIdHex(999, &target_id_hex); + try storeTestEvent(&s, alloc, &target_id_hex, target_pk, base); + + var i: u32 = 0; + while (i < 30) : (i += 1) { + var idb: [64]u8 = undefined; + testIdHex(i + 1, &idb); + try storeTestEvent(&s, alloc, &idb, filler_pk, base + 1 + @as(i64, i)); + } + + s.query_scan_multiplier = 2; + + var target_pk_bytes: [32]u8 = undefined; + _ = try std.fmt.hexToBytes(&target_pk_bytes, target_pk); + var authors = [_][32]u8{ target_pk_bytes, [_]u8{0xcc} ** 32 }; + const filters = [_]nostr.Filter{.{ .authors_bytes = &authors }}; + + var capped = try s.query(&filters, 10); + defer capped.deinit(); + var capped_count: u32 = 0; + while (try capped.next()) |_| capped_count += 1; + try testing.expectEqual(@as(u32, 0), capped_count); + + var full = try s.queryFull(&filters, 10); + defer full.deinit(); + var full_count: u32 = 0; + while (try full.next()) |_| full_count += 1; + try testing.expectEqual(@as(u32, 1), full_count); + + var target_id_bytes: [32]u8 = undefined; + _ = try std.fmt.hexToBytes(&target_id_bytes, &target_id_hex); + var ids = [_][32]u8{target_id_bytes}; + const id_filters = [_]nostr.Filter{.{ .ids_bytes = &ids }}; + var by_id = try s.query(&id_filters, 10); + defer by_id.deinit(); + var id_count: u32 = 0; + while (try by_id.next()) |_| id_count += 1; + try testing.expectEqual(@as(u32, 1), id_count); +} diff --git a/wisp.toml.example b/wisp.toml.example index 9387f47..9e4d101 100644 --- a/wisp.toml.example +++ b/wisp.toml.example @@ -38,6 +38,8 @@ max_event_tags = 2000 max_content_length = 102400 query_limit_default = 500 query_limit_max = 5000 +# Caps entries scanned per query at limit * multiplier; 0 disables (default 20) +query_scan_multiplier = 20 # max_event_age = 94608000 # max_future_seconds = 900 # min_pow_difficulty = 0 From ab3491636ec8a74b5ebefbd393e2484843812de3 Mon Sep 17 00:00:00 2001 From: "William K. Santiago" Date: Fri, 17 Jul 2026 12:18:45 -0400 Subject: [PATCH 3/5] Scope query iterator test txns to one read per thread --- src/store.zig | 50 +++++++++++++++++++++++++++++--------------------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/src/store.zig b/src/store.zig index 06a5e3e..589de8d 100644 --- a/src/store.zig +++ b/src/store.zig @@ -842,25 +842,33 @@ test "queryFull and ids fast-path bypass the scan cap for old matches" { var authors = [_][32]u8{ target_pk_bytes, [_]u8{0xcc} ** 32 }; const filters = [_]nostr.Filter{.{ .authors_bytes = &authors }}; - var capped = try s.query(&filters, 10); - defer capped.deinit(); - var capped_count: u32 = 0; - while (try capped.next()) |_| capped_count += 1; - try testing.expectEqual(@as(u32, 0), capped_count); - - var full = try s.queryFull(&filters, 10); - defer full.deinit(); - var full_count: u32 = 0; - while (try full.next()) |_| full_count += 1; - try testing.expectEqual(@as(u32, 1), full_count); - - var target_id_bytes: [32]u8 = undefined; - _ = try std.fmt.hexToBytes(&target_id_bytes, &target_id_hex); - var ids = [_][32]u8{target_id_bytes}; - const id_filters = [_]nostr.Filter{.{ .ids_bytes = &ids }}; - var by_id = try s.query(&id_filters, 10); - defer by_id.deinit(); - var id_count: u32 = 0; - while (try by_id.next()) |_| id_count += 1; - try testing.expectEqual(@as(u32, 1), id_count); + // LMDB is opened without MDB_NOTLS, so a thread may hold only one read txn + // at a time; scope each iterator so its txn is closed before the next opens. + { + var capped = try s.query(&filters, 10); + defer capped.deinit(); + var capped_count: u32 = 0; + while (try capped.next()) |_| capped_count += 1; + try testing.expectEqual(@as(u32, 0), capped_count); + } + + { + var full = try s.queryFull(&filters, 10); + defer full.deinit(); + var full_count: u32 = 0; + while (try full.next()) |_| full_count += 1; + try testing.expectEqual(@as(u32, 1), full_count); + } + + { + var target_id_bytes: [32]u8 = undefined; + _ = try std.fmt.hexToBytes(&target_id_bytes, &target_id_hex); + var ids = [_][32]u8{target_id_bytes}; + const id_filters = [_]nostr.Filter{.{ .ids_bytes = &ids }}; + var by_id = try s.query(&id_filters, 10); + defer by_id.deinit(); + var id_count: u32 = 0; + while (try by_id.next()) |_| id_count += 1; + try testing.expectEqual(@as(u32, 1), id_count); + } } From 33b386239178cacd738ac1379988f5dc82d380f6 Mon Sep 17 00:00:00 2001 From: "William K. Santiago" Date: Fri, 17 Jul 2026 12:46:15 -0400 Subject: [PATCH 4/5] Return NEG-ERR when negentropy enumeration is scan-cap truncated --- src/handler.zig | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/handler.zig b/src/handler.zig index 60a43bd..d647c45 100644 --- a/src/handler.zig +++ b/src/handler.zig @@ -712,8 +712,9 @@ pub const Handler = struct { if (self.shutdown.load(.acquire)) return; // Serving-side enumeration uses the capped query() by design: this path is - // network-reachable, so reconciliation stays DoS-safe at the cost of - // possibly under-enumerating on a pathologically large DB. + // network-reachable, so reconciliation stays DoS-safe. If the scan cap + // truncates enumeration it is detected below and surfaced as NEG-ERR + // rather than silently under-enumerating. var iter = self.store.query(&[_]nostr.Filter{filter}, self.config.negentropy_max_sync_events) catch { conn.removeNegSession(sub_id); self.sendNegErr(conn, sub_id, "error: query failed"); @@ -734,6 +735,16 @@ pub const Handler = struct { } } + // The scan cap may stop enumeration before all matching stored events are + // seen. Sealing a partial set would make reconciliation report events as + // missing that the relay actually holds, so surface truncation as an error + // rather than silently sealing an incomplete set. + if (iter.max_scan != 0 and iter.scanned >= iter.max_scan) { + conn.removeNegSession(sub_id); + self.sendNegErr(conn, sub_id, "error: result set too large to reconcile"); + return; + } + session.storage.seal(); session.sealed = true; From 8f06d47cd49760d2c5b9b26c634d9c5f3ac395be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kyle=20=F0=9F=90=86?= Date: Fri, 17 Jul 2026 18:10:07 -0400 Subject: [PATCH 5/5] Preserve newest-first ids fast-path ordering and precise NEG-ERR truncation --- src/handler.zig | 2 +- src/store.zig | 177 +++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 154 insertions(+), 25 deletions(-) diff --git a/src/handler.zig b/src/handler.zig index d647c45..25d95b7 100644 --- a/src/handler.zig +++ b/src/handler.zig @@ -739,7 +739,7 @@ pub const Handler = struct { // seen. Sealing a partial set would make reconciliation report events as // missing that the relay actually holds, so surface truncation as an error // rather than silently sealing an incomplete set. - if (iter.max_scan != 0 and iter.scanned >= iter.max_scan) { + if (iter.truncated) { conn.removeNegSession(sub_id); self.sendNegErr(conn, sub_id, "error: result set too large to reconcile"); return; diff --git a/src/store.zig b/src/store.zig index 589de8d..abbdad3 100644 --- a/src/store.zig +++ b/src/store.zig @@ -10,9 +10,9 @@ pub const Store = struct { lmdb: *Lmdb, allocator: std.mem.Allocator, query_cache: QueryCache, - // Caps QueryIterator scanning at `limit * query_scan_multiplier` entries. Set - // from config on the serving path; 0 disables. Defaults to 20 (matches - // queryMultiKind) so tooling that builds a Store directly stays bounded. + // Caps QueryIterator and queryMultiKind scanning at `limit * query_scan_multiplier` + // entries. Set from config on the serving path; 0 disables. Defaults to 20 so + // tooling that builds a Store directly stays bounded. query_scan_multiplier: u32 = 20, events: Dbi, @@ -520,7 +520,16 @@ pub const QueryIterator = struct { prefix_len: usize = 0, skip_filter: bool = false, ids: ?[][32]u8 = null, - ids_index: usize = 0, + // Materialized id fast-path hits, sorted newest-first, built lazily on the + // first nextIds call and freed in deinit. + ids_hits: ?std.ArrayListUnmanaged(IdHit) = null, + ids_hits_index: usize = 0, + // Set true only when scanning stopped because it hit the scan cap with more + // in-prefix entries still available, so callers can tell a truncated scan + // apart from a match set that ended exactly at the cap. + truncated: bool = false, + + const IdHit = struct { json: []const u8, created_at: i64 }; const IndexType = enum { created, kind, pubkey, tag, ids }; @@ -537,12 +546,16 @@ pub const QueryIterator = struct { // Direct point lookups by id are inherently bounded (id list is // capped by max message size), so they bypass the scan cap and - // return matches regardless of age. - if (f.ids()) |id_list| { - if (id_list.len > 0) { - iter.index_type = .ids; - iter.ids = id_list; - return iter; + // return matches regardless of age. Only safe as a single-filter + // fast-path: with multiple filters the match set is a union across + // all of them, and this path only enumerates filter[0]'s ids. + if (filters.len == 1) { + if (f.ids()) |id_list| { + if (id_list.len > 0) { + iter.index_type = .ids; + iter.ids = id_list; + return iter; + } } } @@ -660,7 +673,19 @@ pub const QueryIterator = struct { } while (true) { - if (self.max_scan != 0 and self.scanned >= self.max_scan) return null; + if (self.max_scan != 0 and self.scanned >= self.max_scan) { + // Distinguish hitting the cap with more work remaining from a + // match set that ended exactly at the cap: peek the next entry + // and only flag truncation if an in-prefix entry is still there. + if (try self.cursor.?.get(.prev)) |entry| { + if (self.prefix_len == 0 or (entry.key.len >= self.prefix_len and + std.mem.eql(u8, entry.key[0..self.prefix_len], self.prefix[0..self.prefix_len]))) + { + self.truncated = true; + } + } + return null; + } const entry = try self.cursor.?.get(.prev) orelse return null; self.scanned += 1; @@ -679,26 +704,45 @@ pub const QueryIterator = struct { } fn nextIds(self: *QueryIterator) !?[]const u8 { - if (self.txn == null) self.txn = try self.store.lmdb.beginTxn(true); - const id_list = self.ids orelse return null; + if (self.ids_hits == null) { + if (self.txn == null) self.txn = try self.store.lmdb.beginTxn(true); + const id_list = self.ids orelse return null; - while (self.ids_index < id_list.len) { - const id = &id_list[self.ids_index]; - self.ids_index += 1; + var hits: std.ArrayListUnmanaged(IdHit) = .empty; + errdefer hits.deinit(self.store.allocator); - const json = try self.txn.?.get(self.store.events, id) orelse continue; + // NIP-01 requires `limit` to return the most recent matches. Materialize + // every hit (json pointers stay valid while the read txn is held) and + // sort newest-first so the limit clamp in next() keeps the newest ones. + for (id_list) |*id| { + const json = try self.txn.?.get(self.store.events, id) orelse continue; - var event = nostr.Event.parse(json) catch continue; - defer event.deinit(); + var event = nostr.Event.parse(json) catch continue; + defer event.deinit(); - if (nostr.isExpired(&event)) continue; + if (nostr.isExpired(&event)) continue; - if (self.filters.len == 0 or nostr.filtersMatch(self.filters, &event)) { - self.returned += 1; - return json; + if (self.filters.len == 0 or nostr.filtersMatch(self.filters, &event)) { + try hits.append(self.store.allocator, .{ .json = json, .created_at = event.createdAt() }); + } } + + std.sort.pdq(IdHit, hits.items, {}, struct { + fn lessThan(_: void, a: IdHit, b: IdHit) bool { + return a.created_at > b.created_at; + } + }.lessThan); + + self.ids_hits = hits; } - return null; + + const hits = &self.ids_hits.?; + if (self.ids_hits_index >= hits.items.len) return null; + + const json = hits.items[self.ids_hits_index].json; + self.ids_hits_index += 1; + self.returned += 1; + return json; } const Entry = @import("lmdb.zig").Entry; @@ -744,6 +788,7 @@ pub const QueryIterator = struct { } pub fn deinit(self: *QueryIterator) void { + if (self.ids_hits) |*h| h.deinit(self.store.allocator); if (self.cursor) |*cur| cur.close(); if (self.txn) |*t| t.abort(); } @@ -806,6 +851,90 @@ test "query scan cap stops at limit times multiplier" { while (try iter.next()) |_| count += 1; try testing.expectEqual(@as(u32, 0), count); try testing.expectEqual(@as(u32, 15), iter.scanned); + // Cap fired with more entries (20 stored, 15 scanned) still available. + try testing.expect(iter.truncated); +} + +test "query scan cap not flagged truncated when match set ends at the cap" { + const alloc = testing.allocator; + const io = nostr.io.io(); + const dir = "./.test-scan-cap-exhausted"; + std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + defer std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + + var lmdb = try Lmdb.init(alloc, dir ++ "/db.mdb", 256, .none); + defer lmdb.deinit(); + var s = try Store.init(alloc, &lmdb); + defer s.deinit(); + + const base = nostr.io.timestamp() - 100000; + const filler_pk = "aa" ** 32; + // Exactly max_scan (limit 5 * multiplier 3 = 15) entries, none matching. + var i: u32 = 0; + while (i < 15) : (i += 1) { + var idb: [64]u8 = undefined; + testIdHex(i + 1, &idb); + try storeTestEvent(&s, alloc, &idb, filler_pk, base + @as(i64, i)); + } + + s.query_scan_multiplier = 3; + var authors = [_][32]u8{ [_]u8{0xcc} ** 32, [_]u8{0xdd} ** 32 }; + const filters = [_]nostr.Filter{.{ .authors_bytes = &authors }}; + var iter = try s.query(&filters, 5); + defer iter.deinit(); + + var count: u32 = 0; + while (try iter.next()) |_| count += 1; + try testing.expectEqual(@as(u32, 0), count); + try testing.expectEqual(@as(u32, 15), iter.scanned); + // Scanned exactly the cap but nothing remained, so enumeration was complete. + try testing.expect(!iter.truncated); +} + +test "ids fast-path returns newest matches first under limit" { + const alloc = testing.allocator; + const io = nostr.io.io(); + const dir = "./.test-ids-order"; + std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + defer std.Io.Dir.cwd().deleteTree(io, dir) catch {}; + + var lmdb = try Lmdb.init(alloc, dir ++ "/db.mdb", 256, .none); + defer lmdb.deinit(); + var s = try Store.init(alloc, &lmdb); + defer s.deinit(); + + const base = nostr.io.timestamp() - 100000; + const pk = "cc" ** 32; + + // created_at rises with id index, so the id array below (oldest-first) does + // not match age order; the fast-path must still return newest-first. + const n: u32 = 6; + var ids: [6][32]u8 = undefined; + var i: u32 = 0; + while (i < n) : (i += 1) { + var idb: [64]u8 = undefined; + testIdHex(i + 1, &idb); + try storeTestEvent(&s, alloc, &idb, pk, base + @as(i64, i)); + _ = try std.fmt.hexToBytes(&ids[i], &idb); + } + + const id_filters = [_]nostr.Filter{.{ .ids_bytes = &ids }}; + var iter = try s.query(&id_filters, 3); + defer iter.deinit(); + + var prev: i64 = std.math.maxInt(i64); + var count: u32 = 0; + while (try iter.next()) |json| { + var ev = try nostr.Event.parse(json); + defer ev.deinit(); + const ts = ev.createdAt(); + try testing.expect(ts <= prev); + prev = ts; + count += 1; + } + try testing.expectEqual(@as(u32, 3), count); + // Newest three are base+5, base+4, base+3; the last returned is base+3. + try testing.expectEqual(base + 3, prev); } test "queryFull and ids fast-path bypass the scan cap for old matches" {