From 0cafd770c16303ac3520fd66cb0577063c7e466f Mon Sep 17 00:00:00 2001 From: m5r Date: Wed, 21 Oct 2026 21:41:47 +0200 Subject: [PATCH] filter: a reload builds the snapshot one source at a time, and the startup pass reloads only when a source changed or was not enforced --- CHANGELOG.md | 1 + src/filter/filter_integration_test.zig | 99 +++++-- src/filter/manager.zig | 349 ++++++++++++++++++++----- src/filter/matcher.zig | 336 ++++++++++++++++-------- 4 files changed, 588 insertions(+), 197 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 844fb8b..3ba8c58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ Sections are written by hand. Nothing here is generated from commit messages: th ### Changed +- **Blocklists take less memory, and a refresh peaks lower.** A list that names a domain both on its own and with its subdomains, as Adblock-style lists do for every entry, now stores that name once instead of twice. A reload reads one list's compiled files at a time instead of all of them at once, and the startup pass no longer rebuilds the blocklists when no list was due for a refresh. - **Each database connection keeps a page cache sized for its job** instead of SQLite's 2 MiB default on all seven: 256 KiB for the configuration database and for retention, 1 MiB for the query-history writer and reader. - **The documentation is written for people running nxdns.** It starts from Get started, which downloads a release instead of building from source, and drops the API reference, the troubleshooting page and the performance pages. It is published at , and `INSTALL.md` in the tarball links there. diff --git a/src/filter/filter_integration_test.zig b/src/filter/filter_integration_test.zig index 1c9ed44..ac3db81 100644 --- a/src/filter/filter_integration_test.zig +++ b/src/filter/filter_integration_test.zig @@ -915,7 +915,7 @@ test "3: a damaged compiled file never replaces a serving snapshot with a worse /// The whole refresh path for one source served by `fixture`, with the status /// table synced first so `refreshSource` has somewhere to record its outcome. -fn refreshOnce(env: *Env, url: []const u8) !bool { +fn refreshOnce(env: *Env, url: []const u8) !manager.RefreshOutcome { try env.mgr.reload(env.io()); var rows = try listRows(&env.database); defer rows.deinit(); @@ -940,7 +940,7 @@ test "4: a 200 response is fetched, compiled, recorded and served" { const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try env.mgr.reload(io); var dir = try env.blocklistDir(); @@ -997,7 +997,7 @@ test "5: a redirect is followed to the same result" { const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try env.mgr.reload(io); try testing.expectEqual(manager.State.ok, (try env.status(id)).state); @@ -1040,7 +1040,7 @@ test "6: a 404 leaves the compiled files and the snapshot untouched" { const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try env.mgr.reload(io); var dir = try env.blocklistDir(); @@ -1049,7 +1049,7 @@ test "6: a 404 leaves the compiled files and the snapshot untouched" { defer before.deinit(gpa); fixture.setRoute(.not_found); - try testing.expect(!try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.failed, try refreshOnce(env, url)); const failed = try env.status(id); try testing.expectEqual(manager.State.fetch_failed, failed.state); @@ -1093,7 +1093,7 @@ test "6b: a local failure reports itself, not the previous source's network faul const id = try seedSource(&env.database, url); // First a real network failure, so the fetcher is holding a record. - try testing.expect(!try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.failed, try refreshOnce(env, url)); try expectFailureText( "HttpStatus (receive, cause HttpStatus, http 404, 0 bytes, ", (try env.status(id)).errorText(), @@ -1107,7 +1107,7 @@ test "6b: a local failure reports itself, not the previous source's network faul try dir.setPermissions(io, .fromMode(0o500)); defer dir.setPermissions(io, .fromMode(0o700)) catch {}; - try testing.expect(!try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.failed, try refreshOnce(env, url)); const failed = try env.status(id); try testing.expectEqual(manager.State.fetch_failed, failed.state); @@ -1135,7 +1135,7 @@ test "7: a body over the cap fails the refresh and leaves no temporary file" { const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(!try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.failed, try refreshOnce(env, url)); const failed = try env.status(id); try testing.expectEqual(manager.State.fetch_failed, failed.state); @@ -1170,7 +1170,7 @@ test "8: refetching identical content skips the rewrite and still moves last_upd const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); var dir = try env.blocklistDir(); defer dir.close(io); @@ -1194,7 +1194,7 @@ test "8: refetching identical content skips the rewrite and still moves last_upd }, }); - try testing.expect(!try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.unchanged, try refreshOnce(env, url)); // `replace` renames a new file over the old one, so an untouched file keeps // both its inode and its modification time. @@ -1464,7 +1464,7 @@ const RefreshAllTask = struct { const RefreshSourceTask = struct { mgr: *manager.Manager, row: sources_repo.SourceRow, - result: manager.Manager.Error!bool = false, + result: manager.Manager.Error!manager.RefreshOutcome = .failed, fn run(self: *RefreshSourceTask, io: std.Io) void { self.result = self.mgr.refreshSource(io, self.row); @@ -1589,7 +1589,7 @@ test "10d: a source deleted mid-refresh does not take the refresh's temporary fi try tasks.await(io); // True only if the compile found the raw file the download had written. - try testing.expect(try task.result); + try testing.expectEqual(manager.RefreshOutcome.replaced, try task.result); // The sweep did run, once the refresh was out of the way: the compiled // files of a source with no row are gone, and so is every temporary. @@ -1626,7 +1626,7 @@ test "10e: a reconcile then a restart reuses the compiled files and downloads no // One real download, so the compiled artifacts exist and are named after // the row id the rest of this test is about. - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try testing.expectEqual(@as(u32, 1), fixture.accepted.load(.monotonic)); var dir = try env.blocklistDir(); @@ -1755,6 +1755,63 @@ fn localTable(gpa: std.mem.Allocator, database: *db.Db) !records.Records { return records.Records.build(gpa, rows.items); } +test "10f: a source the boot load could not read is enforced once its refresh finds the same files" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + const env = try Env.create(gpa); + defer env.destroy(); + const io = env.io(); + + var fixture = try HttpFixture.init(io, http_body); + defer fixture.deinit(io); + var group: std.Io.Group = .init; + defer group.cancel(io); + try group.concurrent(io, HttpFixture.serve, .{ &fixture, io }); + + var url_buf: [64]u8 = undefined; + const url = try fixture.url(&url_buf); + const id = try seedSource(&env.database, url); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); + + // The restart, with the `.list` file missing for the boot load alone. + env.mgr.deinit(io); + env.mgr = try manager.Manager.init( + gpa, + &env.database, + .{ .dir = env.tmp.dir }, + &env.f, + .{ .enabled = false }, + budget, + ); + var dir = try env.blocklistDir(); + defer dir.close(io); + var list_buf: [64]u8 = undefined; + const list_name = try std.fmt.bufPrint(&list_buf, "{d}.list", .{id}); + try dir.rename(list_name, dir, "aside", io); + try env.mgr.reload(io); + try dir.rename("aside", dir, list_name, io); + + try testing.expect(!(try env.status(id)).loaded); + { + const decision, _ = try env.evaluate("ads.example.com"); + try testing.expect(!decision.blocked); + } + + // The startup pass refreshes the source, which is not `ok`, and the + // download compiles to the files already on disk. That alone rewrites + // nothing, so it is the reload after it that puts the source back. + env.mgr.schedule_clock = .shutdown_at_first_park; + try testing.expectError(error.Canceled, env.mgr.runScheduler(io)); + try testing.expectEqual(@as(u32, 2), fixture.accepted.load(.monotonic)); + + const status = try env.status(id); + try testing.expect(status.loaded); + try testing.expectEqual(manager.State.ok, status.state); + const decision, _ = try env.evaluate("ads.example.com"); + try testing.expect(decision.blocked); +} + test "11: a local A record read from the database answers an A query" { if (!build_options.integration) return error.SkipZigTest; @@ -2109,7 +2166,7 @@ test "18: a body streamed in flushed parts survives the fetcher's multi-read pum const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try env.mgr.reload(io); try testing.expect(fixture.flushed_parts.load(.acquire) >= 3); @@ -2181,7 +2238,7 @@ test "19: a downloaded list's exceptions lift its own blocks and nothing else" { const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try env.mgr.reload(io); // The `.allow` body is a third file beside the two, and the row and the @@ -2371,7 +2428,7 @@ test "21: an abp list's unsupported lines are counted, persisted and read back" const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try env.mgr.reload(io); var rows = try listRows(&env.database); @@ -2432,7 +2489,7 @@ test "23: a name moving from the list body to the wild body forces a republish" const url = try fixture.url(&url_buf); _ = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); var first_checksum: [64]u8 = undefined; { @@ -2464,7 +2521,7 @@ test "22: a list that changed only its skipped lines still updates both skip cou const url = try fixture.url(&url_buf); const id = try seedSource(&env.database, url); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); var first_checksum: [64]u8 = undefined; { @@ -2479,7 +2536,7 @@ test "22: a list that changed only its skipped lines still updates both skip cou // The edited list. Two more skipped lines and not one written entry moved, // so the refresh finds its stored checksum and rewrites nothing on disk. fixture.setRoute(.changed); - try testing.expect(!try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.unchanged, try refreshOnce(env, url)); { var rows = try listRows(&env.database); @@ -2548,7 +2605,7 @@ test "27: a failing refresh opens a blocklist.refresh episode a good one closes" const url = try fixture.url(&url_buf); _ = try seedSource(&env.database, url); - try testing.expect(!try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.failed, try refreshOnce(env, url)); try testing.expectEqualStrings("blocklist.refresh", try fx.text( "SELECT code FROM operational_events WHERE resolved_at IS NULL", @@ -2562,7 +2619,7 @@ test "27: a failing refresh opens a blocklist.refresh episode a good one closes" )); fixture.setRoute(.body); - try testing.expect(try refreshOnce(env, url)); + try testing.expectEqual(manager.RefreshOutcome.replaced, try refreshOnce(env, url)); try testing.expectEqual( @as(i64, 0), diff --git a/src/filter/manager.zig b/src/filter/manager.zig index cb42040..30be39f 100644 --- a/src/filter/manager.zig +++ b/src/filter/manager.zig @@ -892,17 +892,32 @@ pub const Manager = struct { defer self.gpa.free(sources); const source_ids = try self.gpa.alloc(i64, rows.items.len); defer self.gpa.free(source_ids); - const compiled = try self.gpa.alloc(?matcher.Snapshot.Compiled, rows.items.len); - defer self.gpa.free(compiled); - - // The file contents outlive the header stripping and are freed once the - // snapshot has copied what it needs into its own arena. - var bodies: std.ArrayList([]u8) = .empty; - defer { - for (bodies.items) |body| self.gpa.free(body); - bodies.deinit(self.gpa); + for (rows.items, sources, source_ids) |row, *source, *source_id| { + source_id.* = row.id; + // `is_suggested` is a UI hint the snapshot never reads. + source.* = .{ .url = row.url, .name = row.name, .enabled = row.enabled }; } + // Query names are attacker-supplied, so a fixed seed would make + // probe-chain flooding computable offline. + var seed_bytes: [8]u8 = undefined; + io.random(&seed_bytes); + + var builder: matcher.Snapshot.Builder = try .init( + self.gpa, + sources, + source_ids, + std.mem.readInt(u64, &seed_bytes, .little), + ); + defer builder.deinit(); + + // One source's file contents at a time: each is freed once the builder + // has copied what it keeps, before the next source is read. A reload + // then holds the old snapshot, the new one and one source's bodies. + var bodies: std.ArrayList([]u8) = .empty; + defer bodies.deinit(self.gpa); + defer for (bodies.items) |body| self.gpa.free(body); + // What this reload found, per source. It is applied to the status table // only if the snapshot it describes is published: everything below here // can still fail, and a status table describing a snapshot nobody @@ -911,24 +926,25 @@ pub const Manager = struct { defer self.gpa.free(outcomes); var loaded: usize = 0; - for (rows.items, sources, source_ids, compiled, outcomes) |row, *source, *source_id, *slot, *outcome| { - source_id.* = row.id; - // `is_suggested` is a UI hint the snapshot never reads. - source.* = .{ .url = row.url, .name = row.name, .enabled = row.enabled }; - outcome.* = if (row.enabled) try self.loadSource(io, dir, row, &bodies) else .disabled; - switch (outcome.*) { - .loaded => |body| { - slot.* = body; + for (rows.items, outcomes, 0..) |row, *outcome, index| { + if (!row.enabled) { + outcome.* = .disabled; + continue; + } + switch (try self.loadSource(io, dir, row, &bodies)) { + .loaded => |compiled| { + try builder.addSource(index, compiled); + outcome.* = .loaded; loaded += 1; }, - // Not loadable and therefore not enforced. Saying so here is - // what keeps `Snapshot.build`'s `MissingCompiledSource` for the - // case it is meant for: a caller that forgot to read a body. - .disabled, .failed => { - slot.* = null; - source.enabled = false; + // Not loadable and therefore not enforced. + .failed => |failure| { + builder.skipSource(index); + outcome.* = .{ .failed = failure }; }, } + for (bodies.items) |body| self.gpa.free(body); + bodies.clearRetainingCapacity(); } var groups = try groups_repo.listGroups(self.database, self.gpa); @@ -954,25 +970,16 @@ pub const Manager = struct { defer prefixes.deinit(self.gpa); defer clients_repo.freeClientPrefixes(self.gpa, prefixes.items); - // Query names are attacker-supplied, so a fixed seed would make - // probe-chain flooding computable offline. - var seed_bytes: [8]u8 = undefined; - io.random(&seed_bytes); - const generation = self.generation + 1; const snapshot = try self.gpa.create(matcher.Snapshot); errdefer self.gpa.destroy(snapshot); - snapshot.* = try matcher.Snapshot.build(self.gpa, .{ + snapshot.* = try builder.finish(.{ .groups = groups.items, .group_ids = group_ids, .group_sources = group_sources.items, - .sources = sources, - .source_ids = source_ids, .rules = rule_rows.items, .clients = clients.items, .prefixes = prefixes.items, - .compiled = compiled, - .seed = std.mem.readInt(u64, &seed_bytes, .little), .generation = generation, }); @@ -1014,7 +1021,7 @@ pub const Manager = struct { dir: std.Io.Dir, row: sources_repo.SourceRow, bodies: *std.ArrayList([]u8), - ) Error!LoadOutcome { + ) Error!SourceLoad { const stored = row.checksum orelse // No stored checksum means no successful compile has ever // happened. A fresh install is here on every source. @@ -1091,15 +1098,14 @@ pub const Manager = struct { /// Downloads, compiles and atomically replaces the compiled files of one /// source, then writes its runtime columns. /// - /// Returns `true` only when the compiled files were replaced. Unchanged - /// content (equal checksum) and every recorded failure return `false`; the - /// reason for a failure is in the status table, not in the return value. + /// The reason for a failure is in the status table, not in the return + /// value. /// /// The status entry is found by row id, so a source added since the last /// `reload` has nowhere to record its outcome. `refreshAll` syncs the table /// before it refreshes anything, which is why `POST /api/blocklists/update` /// and the scheduler both enter through `refreshAll`. - pub fn refreshSource(self: *Manager, io: std.Io, row: sources_repo.SourceRow) Error!bool { + pub fn refreshSource(self: *Manager, io: std.Io, row: sources_repo.SourceRow) Error!RefreshOutcome { self.refresh_lock.lockUncancelable(io); defer self.refresh_lock.unlock(io); // Registered after the unlock, so it runs before it: a pass drains its @@ -1112,7 +1118,7 @@ pub const Manager = struct { /// Assumes `refresh_lock`. Takes `writer_lock` itself, for the publish half /// alone. - fn refreshSourceLocked(self: *Manager, io: std.Io, row: sources_repo.SourceRow) Error!bool { + fn refreshSourceLocked(self: *Manager, io: std.Io, row: sources_repo.SourceRow) Error!RefreshOutcome { // The previous entry describes the compiled files that are still on // disk, and a failed refresh leaves them serving. Starting from a blank // status would erase `last_success` and the counters of a list that is @@ -1156,9 +1162,9 @@ pub const Manager = struct { self.writer_lock.lockUncancelable(io); defer self.writer_lock.unlock(io); - const replaced = try self.publishRefresh(io, dir, row, &status, prepared, tmp); + const outcome = try self.publishRefresh(io, dir, row, &status, prepared, tmp); self.commitStatus(io, status); - return replaced; + return outcome; } /// Every enabled source, one at a time, then one `reload`. A failing source @@ -1260,7 +1266,6 @@ pub const Manager = struct { } /// Installs what `prepareRefresh` produced and writes the runtime columns. - /// Returns `true` only when the compiled files were replaced. /// /// Assumes `writer_lock`. Reading the files on disk belongs here and not in /// the half above: they are published under this lock. @@ -1272,9 +1277,9 @@ pub const Manager = struct { status: *SourceStatus, prepared: Prepared, tmp: TempNames, - ) Error!bool { + ) Error!RefreshOutcome { const compiled = switch (prepared) { - .failed => return false, + .failed => return .failed, .compiled => |value| value, }; @@ -1308,7 +1313,7 @@ pub const Manager = struct { .checksum = stored, }); status.succeed(now, compiled.result.counts); - return false; + return .unchanged; } } @@ -1324,7 +1329,7 @@ pub const Manager = struct { error.Canceled => return error.Canceled, else => { self.reportCompileFailure(row, status, err); - return false; + return .failed; }, }; @@ -1338,7 +1343,7 @@ pub const Manager = struct { .checksum = &compiled.result.checksum, }); status.succeed(now, compiled.result.counts); - return true; + return .replaced; } /// The body goes to a temporary file, never to memory: `max_body_bytes` is @@ -1817,10 +1822,13 @@ pub const Manager = struct { // `refreshSource`. defer self.flushDiagnostics(io); - // Ahead of the gate on purpose: loading the compiled files that already - // exist is a read. A full disk must not cost the household its - // filtering as well as its downloads. - try self.reloadCollecting(io); + // Boot already loaded the compiled files before the listeners started, + // and its status table is what the refresh decisions below read. Only + // a boot load that published nothing is retried here, ahead of the + // gate on purpose: loading the files that already exist is a read, and + // a full disk must not cost the household its filtering as well as its + // downloads. + if (!self.published(io)) try self.reloadCollecting(io); if (self.refreshGated()) return; @@ -1829,13 +1837,36 @@ pub const Manager = struct { defer sources_repo.freeSourceRows(self.gpa, rows.items); const now = std.Io.Clock.real.now(io).toSeconds(); - var refreshed = false; + var stale = false; for (rows.items) |row| { if (!row.enabled) continue; if (!self.needsRefresh(io, row, now)) continue; - if (try self.refreshSourceLocked(io, row)) refreshed = true; + switch (try self.refreshSourceLocked(io, row)) { + .replaced => stale = true, + // The files on disk match their checksum, yet the published + // snapshot may still lack them: a boot load that could not read + // them left the source out, and nothing else would add it back + // before the next scheduled pass. + .unchanged => if (!self.enforced(io, row.id)) { + stale = true; + }, + .failed => {}, + } } - if (refreshed) try self.reloadCollecting(io); + if (stale) try self.reloadCollecting(io); + } + + fn enforced(self: *Manager, io: std.Io, id: i64) bool { + self.lock.lockSharedUncancelable(io); + defer self.lock.unlockShared(io); + const entry = entryFor(self.statuses, id) orelse return false; + return entry.loaded; + } + + fn published(self: *Manager, io: std.Io) bool { + self.lock.lockSharedUncancelable(io); + defer self.lock.unlockShared(io); + return self.generation != 0; } fn needsRefresh(self: *Manager, io: std.Io, row: sources_repo.SourceRow, now: i64) bool { @@ -2146,12 +2177,32 @@ const LoadOutcome = union(enum) { /// The source is switched off. Not in the snapshot, whatever it was before. disabled, /// Its compiled files are in the snapshot. - loaded: matcher.Snapshot.Compiled, + loaded, /// It is not in the snapshot, for this reason. - failed: struct { state: State, text: []const u8 }, + failed: LoadFailure, }; -fn loadFailure(row: sources_repo.SourceRow, file_name: []const u8, err: anyerror) LoadOutcome { +const LoadFailure = struct { state: State, text: []const u8 }; + +/// What one source's refresh did to its compiled files. +pub const RefreshOutcome = enum { + /// Recorded in the status table; the previous files, if any, still serve. + failed, + /// The download compiled to the files already on disk, which were left + /// alone. + unchanged, + /// New compiled files were installed. + replaced, +}; + +/// What reading one enabled source's compiled files produced. The bodies +/// borrow the buffers `loadSource` appended to its caller's list. +const SourceLoad = union(enum) { + loaded: matcher.Snapshot.Compiled, + failed: LoadFailure, +}; + +fn loadFailure(row: sources_repo.SourceRow, file_name: []const u8, err: anyerror) SourceLoad { log.warn("blocklist {f}: reading {s} failed: {s}", .{ SourceLabel.of(row), file_name, @errorName(err) }); return .{ .failed = .{ .state = .load_failed, .text = @errorName(err) } }; } @@ -2608,6 +2659,184 @@ test "a canceled compiled-file read cancels the reload instead of recording it" } } +/// Writes one source's compiled files and the checksum and stamp a refresh +/// would have recorded, so a reload loads it. +fn installCompiledSource( + io: std.Io, + database: *db.Db, + dir: std.Io.Dir, + url: []const u8, + list_body: []const u8, + wild_body: []const u8, + last_updated: i64, +) !void { + try sources_repo.insertBlocklistSource(database, .{ .url = url, .name = url }, .{}); + var rows = try sources_repo.listSourceRows(database, testing.allocator); + defer rows.deinit(testing.allocator); + defer sources_repo.freeSourceRows(testing.allocator, rows.items); + const id = for (rows.items) |row| { + if (std.mem.eql(u8, row.url, url)) break row.id; + } else unreachable; + + var list_buf: [name_buf_len]u8 = undefined; + var wild_buf: [name_buf_len]u8 = undefined; + try dir.writeFile(io, .{ .sub_path = compiledName(&list_buf, id, ".list"), .data = list_body }); + try dir.writeFile(io, .{ .sub_path = compiledName(&wild_buf, id, ".wild"), .data = wild_body }); + try sources_repo.updateSourceStats(database, id, .{ + .last_updated = last_updated, + .domain_count = @intCast(std.mem.count(u8, list_body, "\n")), + .wildcard_count = @intCast(std.mem.count(u8, wild_body, "\n")), + .skipped_regex_count = 0, + .skipped_unsupported_count = 0, + .exception_count = 0, + .checksum = &bodyChecksum(list_body, wild_body, ""), + }); +} + +/// Forwards to `child` and records the most bytes it held at once. +const PeakAllocator = struct { + child: Allocator, + live: usize = 0, + peak: usize = 0, + + fn allocator(self: *PeakAllocator) Allocator { + return .{ .ptr = self, .vtable = &.{ + .alloc = alloc, + .resize = resize, + .remap = remap, + .free = free, + } }; + } + + fn held(self: *PeakAllocator, old_len: usize, new_len: usize) void { + self.live = self.live - old_len + new_len; + self.peak = @max(self.peak, self.live); + } + + fn alloc(ctx: *anyopaque, len: usize, alignment: std.mem.Alignment, ret_addr: usize) ?[*]u8 { + const self: *PeakAllocator = @ptrCast(@alignCast(ctx)); + const ptr = self.child.rawAlloc(len, alignment, ret_addr) orelse return null; + self.held(0, len); + return ptr; + } + + fn resize(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, new_len: usize, ret_addr: usize) bool { + const self: *PeakAllocator = @ptrCast(@alignCast(ctx)); + if (!self.child.rawResize(memory, alignment, new_len, ret_addr)) return false; + self.held(memory.len, new_len); + return true; + } + + fn remap(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, new_len: usize, ret_addr: usize) ?[*]u8 { + const self: *PeakAllocator = @ptrCast(@alignCast(ctx)); + const ptr = self.child.rawRemap(memory, alignment, new_len, ret_addr) orelse return null; + self.held(memory.len, new_len); + return ptr; + } + + fn free(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, ret_addr: usize) void { + const self: *PeakAllocator = @ptrCast(@alignCast(ctx)); + self.child.rawFree(memory, alignment, ret_addr); + self.held(memory.len, 0); + } +}; + +test "a reload holds one source's bodies at a time, not every source's" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var dir = try tmp.dir.createDirPathOpen(io, "blocklists", .{}); + defer dir.close(io); + + var database = try openMigrated(); + defer database.close(); + + // As an ABP list compiles: the same names in `.list` and `.wild`. + var body: std.ArrayList(u8) = .empty; + defer body.deinit(testing.allocator); + for (0..4_000) |i| try body.print(testing.allocator, "n{d:0>5}.example.com\n", .{i}); + + const source_count = 6; + for (0..source_count) |i| { + var url_buf: [64]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "https://lists.example/{d}.txt", .{i}); + try installCompiledSource(io, &database, dir, url, body.items, body.items, 1_700_000_000); + } + const one_source_bytes = 2 * body.items.len; + + var peak: PeakAllocator = .{ .child = testing.allocator }; + var f: fetcher.Fetcher = undefined; + var mgr: Manager = try .init(peak.allocator(), &database, .{ .dir = tmp.dir }, &f, .{}, .{ .raw = .fromSeconds(30), .clock = .awake }); + defer mgr.deinit(io); + + try mgr.reload(io); + try testing.expectEqual(@as(u64, 1), mgr.generation); + + // What stays live is the snapshot and the status table. Everything above + // that was transient, and every source's bodies at once would be six + // times `one_source_bytes`. A file read may grow its buffer by copying, + // which holds one body twice for a moment, hence two sources' worth. + const transient = peak.peak - peak.live; + try testing.expect(transient < 2 * one_source_bytes); +} + +test "the startup pass reuses the boot load when no source is due" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var dir = try tmp.dir.createDirPathOpen(io, "blocklists", .{}); + defer dir.close(io); + + var database = try openMigrated(); + defer database.close(); + const now = std.Io.Clock.real.now(io).toSeconds(); + try installCompiledSource(io, &database, dir, "https://lists.example/a.txt", "a.example\n", "", now); + + // Never called: a source refreshed `now` is not due. + var f: fetcher.Fetcher = undefined; + var mgr = try testManager(&database, &f); + defer mgr.deinit(io); + mgr.paths = .{ .dir = tmp.dir }; + + try mgr.reload(io); + try mgr.startupPass(io); + try testing.expectEqual(@as(u64, 1), mgr.generation); + + const handle = mgr.acquire(io).?; + defer handle.release(io); + try testing.expectEqual(@as(usize, 1), handle.snapshot.sources.len); +} + +test "the startup pass loads the snapshot a failed boot load left unpublished" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var dir = try tmp.dir.createDirPathOpen(io, "blocklists", .{}); + defer dir.close(io); + + var database = try openMigrated(); + defer database.close(); + const now = std.Io.Clock.real.now(io).toSeconds(); + try installCompiledSource(io, &database, dir, "https://lists.example/a.txt", "a.example\n", "", now); + + var f: fetcher.Fetcher = undefined; + var mgr = try testManager(&database, &f); + defer mgr.deinit(io); + mgr.paths = .{ .dir = tmp.dir }; + + try mgr.startupPass(io); + try testing.expectEqual(@as(u64, 1), mgr.generation); +} + test "committing a status for an id the table has no entry for is not silent" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); @@ -3123,8 +3352,7 @@ test "a source that failed to refresh keeps its failure while its old files serv statuses[0].fail(.fetch_failed, "HttpStatus"); const rows = [_]sources_repo.SourceRow{testRow(1, true)}; - const body: matcher.Snapshot.Compiled = .{ .list_body = "", .wild_body = "" }; - applyLoadOutcomes(&statuses, &rows, &.{.{ .loaded = body }}); + applyLoadOutcomes(&statuses, &rows, &.{.loaded}); try testing.expect(statuses[0].loaded); try testing.expectEqual(State.fetch_failed, statuses[0].state); @@ -3157,8 +3385,7 @@ test "a load failure is recorded when no refresh failure explains it" { test "a load of a source this process never refreshed takes the row counters" { var statuses = [_]SourceStatus{.{ .id = 1 }}; const rows = [_]sources_repo.SourceRow{testRow(1, true)}; - const body: matcher.Snapshot.Compiled = .{ .list_body = "", .wild_body = "" }; - applyLoadOutcomes(&statuses, &rows, &.{.{ .loaded = body }}); + applyLoadOutcomes(&statuses, &rows, &.{.loaded}); try testing.expect(statuses[0].loaded); try testing.expectEqual(State.ok, statuses[0].state); diff --git a/src/filter/matcher.zig b/src/filter/matcher.zig index b1b81d3..62e8b42 100644 --- a/src/filter/matcher.zig +++ b/src/filter/matcher.zig @@ -183,113 +183,233 @@ pub const Snapshot = struct { BadClientPrefix, } || rules.Error; - /// Builds an immutable snapshot. Every string is copied into the arena, so - /// the caller may free the repository lists immediately afterwards. - /// Disabled sources are skipped entirely — they cost no memory. + /// Builds an immutable snapshot from bodies the caller already holds, every + /// one at once. A caller reading bodies from disk uses `Builder` instead, + /// so it can free each source's bodies before it reads the next. pub fn build(gpa: Allocator, input: Input) Error!Snapshot { - std.debug.assert(input.group_ids.len == input.groups.len); - std.debug.assert(input.source_ids.len == input.sources.len); - - var arena_state: std.heap.ArenaAllocator = .init(gpa); - errdefer arena_state.deinit(); - const arena = arena_state.allocator(); - - // `position[i]` is where `input.sources[i]` landed in `sources`, or - // null when the source is disabled and therefore not loaded. - const position = try arena.alloc(?u32, input.sources.len); - var enabled: u32 = 0; - for (input.sources, position) |row, *slot| { - if (!row.enabled) { - slot.* = null; - continue; - } - slot.* = enabled; - enabled += 1; - } - - const sources = try arena.alloc(SourceSets, enabled); - for (input.sources, input.source_ids, position, 0..) |row, id, slot, i| { - const at = slot orelse continue; + var builder: Builder = try .init(gpa, input.sources, input.source_ids, input.seed); + defer builder.deinit(); + for (input.sources, 0..) |row, i| { + if (!row.enabled) continue; if (i >= input.compiled.len) return error.MissingCompiledSource; - const bodies = input.compiled[i] orelse return error.MissingCompiledSource; - sources[at] = .{ - .id = id, - .name = try arena.dupe(u8, row.name), - .names = try domain_set.DomainSet.buildMerged(arena, bodies.list_body, bodies.wild_body, input.seed), - .exceptions = try domain_set.DomainSet.build(arena, bodies.allow_body, input.seed), - }; + try builder.addSource(i, input.compiled[i] orelse return error.MissingCompiledSource); } - - const groups = try arena.alloc(Group, input.groups.len); - var group_rules: std.ArrayList(model.Rule) = .empty; - defer group_rules.deinit(gpa); - var group_sources: std.ArrayList(u32) = .empty; - defer group_sources.deinit(gpa); - - for (groups, input.groups, input.group_ids) |*group, row, id| { - group_rules.clearRetainingCapacity(); - for (input.rules) |rule_row| { - if (std.mem.eql(u8, rule_row.group, row.name)) try group_rules.append(gpa, rule_row); - } - - group_sources.clearRetainingCapacity(); - for (input.group_sources) |link| { - if (!std.mem.eql(u8, link.group, row.name)) continue; - const at = try findSource(input.sources, position, link.source_url); - // A link to a disabled source is not an error: the source - // exists, it is simply not loaded. - if (at) |index| try group_sources.append(gpa, index); - } - std.mem.sort(u32, group_sources.items, {}, std.sort.asc(u32)); - - group.* = .{ - .id = id, - .name = try arena.dupe(u8, row.name), - .safe_search = row.safe_search, - // `arena` retains, `gpa` scratches: an arena reclaims only its - // most recent allocation, so a rule build's temporaries taken - // from it would outlive the build and go unreported by - // `memoryBytes`. - .rules = try rules.RuleSet.build(arena, gpa, group_rules.items, input.seed), - .sources = try arena.dupe(u32, dedupSorted(group_sources.items)), - }; - } - - // Every `group_sources` row must name a group that exists, whether or - // not that group also owns rules. - for (input.group_sources) |link| { - if (indexOfGroup(groups, link.group) == null) return error.UnknownGroup; - } - - const clients = try arena.alloc(ClientEntry, input.clients.len); - for (clients, input.clients) |*entry, row| { - const addr = address.NetAddress.parse(row.ip) catch return error.BadClientIp; - entry.* = .{ - .key = addr.key(), - .group = indexOfGroup(groups, row.group) orelse return error.UnknownGroup, - }; - } - - const prefixes = try arena.alloc(PrefixEntry, input.prefixes.len); - for (prefixes, input.prefixes) |*entry, row| { - entry.* = .{ - .prefix = address.Prefix.parse(row.prefix) catch return error.BadClientPrefix, - .group = indexOfGroup(groups, row.group) orelse return error.UnknownGroup, - .priority = row.priority, - }; - } - - return .{ - .arena = arena_state, - .groups = groups, - .sources = sources, - .clients = clients, - .prefixes = prefixes, - .default_group = indexOfGroup(groups, "default") orelse return error.MissingDefaultGroup, + return builder.finish(.{ + .groups = input.groups, + .group_ids = input.group_ids, + .group_sources = input.group_sources, + .rules = input.rules, + .clients = input.clients, + .prefixes = input.prefixes, .generation = input.generation, - }; + }); } + /// Everything but the sources: what `Builder.finish` assigns them to. + pub const Assignments = struct { + groups: []const model.Group, + /// One `groups` row id per `groups[i]`, in the same order. + group_ids: []const i64, + group_sources: []const model.GroupSource, + rules: []const model.Rule, + clients: []const model.Client, + prefixes: []const model.ClientPrefix, + generation: u64, + }; + + /// A snapshot built one source at a time. Each `addSource` copies what the + /// snapshot keeps into its arena, so the caller may free that source's + /// bodies as soon as the call returns: a reload then holds at most one + /// source's bodies beside the snapshot it is building. + /// + /// Every enabled source row is either added or skipped before `finish`. A + /// skipped source is not enforced and costs no memory; an enabled one left + /// pending is `error.MissingCompiledSource`, because a silently unenforced + /// blocklist is exactly the failure PLAN §1.3 exists to prevent. + /// + /// `deinit` is always safe to call, before or after `finish`. + pub const Builder = struct { + gpa: Allocator, + /// Null once `finish` has handed the arena to the snapshot. + arena_state: ?std.heap.ArenaAllocator, + rows: []const model.BlocklistSource, + row_ids: []const i64, + /// One per `rows[i]`. + states: []RowState, + /// Capacity: the enabled rows. Length: the rows added so far. + sources: []SourceSets, + added: u32, + /// The lowest row `addSource` still accepts. + next_row: usize, + seed: u64, + + const RowState = union(enum) { + pending, + skipped, + /// Index into `sources`. + added: u32, + }; + + /// `rows` and `row_ids` are borrowed until `finish`. Disabled rows are + /// skipped here. + pub fn init(gpa: Allocator, rows: []const model.BlocklistSource, row_ids: []const i64, seed: u64) Error!Builder { + std.debug.assert(row_ids.len == rows.len); + var arena_state: std.heap.ArenaAllocator = .init(gpa); + errdefer arena_state.deinit(); + const arena = arena_state.allocator(); + + const states = try arena.alloc(RowState, rows.len); + var enabled: usize = 0; + for (rows, states) |row, *state| { + state.* = if (row.enabled) .pending else .skipped; + if (row.enabled) enabled += 1; + } + const sources = try arena.alloc(SourceSets, enabled); + + return .{ + .gpa = gpa, + .arena_state = arena_state, + .rows = rows, + .row_ids = row_ids, + .states = states, + .sources = sources, + .added = 0, + .next_row = 0, + .seed = seed, + }; + } + + pub fn deinit(self: *Builder) void { + if (self.arena_state) |*arena| arena.deinit(); + self.arena_state = null; + } + + /// Builds `rows[row]`'s sets from `bodies`. Nothing of `bodies` is + /// borrowed once this returns. + /// + /// Rows are added in ascending order. A source's index in the snapshot + /// is its position among the added rows, and that index decides which + /// of several overlapping sources reports a match. + pub fn addSource(self: *Builder, row: usize, bodies: Compiled) Error!void { + std.debug.assert(self.states[row] == .pending); + std.debug.assert(row >= self.next_row); + const arena = self.arena_state.?.allocator(); + self.sources[self.added] = .{ + .id = self.row_ids[row], + .name = try arena.dupe(u8, self.rows[row].name), + .names = try domain_set.DomainSet.buildMerged(arena, bodies.list_body, bodies.wild_body, self.seed), + .exceptions = try domain_set.DomainSet.build(arena, bodies.allow_body, self.seed), + }; + self.states[row] = .{ .added = self.added }; + self.added += 1; + self.next_row = row + 1; + } + + /// An enabled source whose bodies could not be loaded. It is not + /// enforced; the caller reports why. + pub fn skipSource(self: *Builder, row: usize) void { + std.debug.assert(self.states[row] == .pending); + self.states[row] = .skipped; + } + + pub fn finish(self: *Builder, assignments: Assignments) Error!Snapshot { + std.debug.assert(assignments.group_ids.len == assignments.groups.len); + for (self.states) |state| { + if (state == .pending) return error.MissingCompiledSource; + } + const gpa = self.gpa; + const arena = self.arena_state.?.allocator(); + const sources = self.sources[0..self.added]; + + const groups = try arena.alloc(Group, assignments.groups.len); + var group_rules: std.ArrayList(model.Rule) = .empty; + defer group_rules.deinit(gpa); + var group_sources: std.ArrayList(u32) = .empty; + defer group_sources.deinit(gpa); + + for (groups, assignments.groups, assignments.group_ids) |*group, row, id| { + group_rules.clearRetainingCapacity(); + for (assignments.rules) |rule_row| { + if (std.mem.eql(u8, rule_row.group, row.name)) try group_rules.append(gpa, rule_row); + } + + group_sources.clearRetainingCapacity(); + for (assignments.group_sources) |link| { + if (!std.mem.eql(u8, link.group, row.name)) continue; + const at = try self.findSource(link.source_url); + // A link to a disabled or unloaded source is not an error: + // the source exists, it is simply not enforced. + if (at) |index| try group_sources.append(gpa, index); + } + std.mem.sort(u32, group_sources.items, {}, std.sort.asc(u32)); + + group.* = .{ + .id = id, + .name = try arena.dupe(u8, row.name), + .safe_search = row.safe_search, + // `arena` retains, `gpa` scratches: an arena reclaims only + // its most recent allocation, so a rule build's temporaries + // taken from it would outlive the build and go unreported + // by `memoryBytes`. + .rules = try rules.RuleSet.build(arena, gpa, group_rules.items, self.seed), + .sources = try arena.dupe(u32, dedupSorted(group_sources.items)), + }; + } + + // Every `group_sources` row must name a group that exists, whether + // or not that group also owns rules. + for (assignments.group_sources) |link| { + if (indexOfGroup(groups, link.group) == null) return error.UnknownGroup; + } + + const clients = try arena.alloc(ClientEntry, assignments.clients.len); + for (clients, assignments.clients) |*entry, row| { + const addr = address.NetAddress.parse(row.ip) catch return error.BadClientIp; + entry.* = .{ + .key = addr.key(), + .group = indexOfGroup(groups, row.group) orelse return error.UnknownGroup, + }; + } + + const prefixes = try arena.alloc(PrefixEntry, assignments.prefixes.len); + for (prefixes, assignments.prefixes) |*entry, row| { + entry.* = .{ + .prefix = address.Prefix.parse(row.prefix) catch return error.BadClientPrefix, + .group = indexOfGroup(groups, row.group) orelse return error.UnknownGroup, + .priority = row.priority, + }; + } + + const default_group = indexOfGroup(groups, "default") orelse return error.MissingDefaultGroup; + const snapshot: Snapshot = .{ + .arena = self.arena_state.?, + .groups = groups, + .sources = sources, + .clients = clients, + .prefixes = prefixes, + .default_group = default_group, + .generation = assignments.generation, + }; + self.arena_state = null; + return snapshot; + } + + /// The position in `sources`, or null when the named source exists but + /// is not enforced. A URL naming no configured source is an error: the + /// group would silently enforce one list fewer than the operator + /// configured. + fn findSource(self: *const Builder, url: []const u8) error{UnknownSource}!?u32 { + for (self.rows, self.states) |row, state| { + if (!std.mem.eql(u8, row.url, url)) continue; + return switch (state) { + .added => |index| index, + .pending, .skipped => null, + }; + } + return error.UnknownSource; + } + }; + pub fn deinit(self: *Snapshot) void { self.arena.deinit(); self.* = undefined; @@ -472,20 +592,6 @@ fn indexOfGroup(groups: []const Group, group_name: []const u8) ?u32 { return null; } -/// The position in the loaded `sources` array, or null when the named source -/// exists but is disabled. A URL naming no configured source is an error: the -/// group would silently enforce one list fewer than the operator configured. -fn findSource( - rows: []const model.BlocklistSource, - position: []const ?u32, - url: []const u8, -) error{UnknownSource}!?u32 { - for (rows, position) |row, slot| { - if (std.mem.eql(u8, row.url, url)) return slot; - } - return error.UnknownSource; -} - fn dedupSorted(items: []u32) []const u32 { var kept: usize = 0; for (items) |item| {