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

@@ -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 <https://nxdns.sh/docs/>, and `INSTALL.md` in the tarball links there.
 

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),

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);
+        for (rows.items, sources, source_ids) |row, *source, *source_id| {
-        defer self.gpa.free(compiled);
+            source_id.* = row.id;
-
+            // `is_suggested` is a UI hint the snapshot never reads.
-        // The file contents outlive the header stripping and are freed once the
+            source.* = .{ .url = row.url, .name = row.name, .enabled = row.enabled };
-        // 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);
         }
 
+        // 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| {
+        for (rows.items, outcomes, 0..) |row, *outcome, index| {
-            source_id.* = row.id;
+            if (!row.enabled) {
-            // `is_suggested` is a UI hint the snapshot never reads.
+                outcome.* = .disabled;
-            source.* = .{ .url = row.url, .name = row.name, .enabled = row.enabled };
+                continue;
-            outcome.* = if (row.enabled) try self.loadSource(io, dir, row, &bodies) else .disabled;
+            }
-            switch (outcome.*) {
+            switch (try self.loadSource(io, dir, row, &bodies)) {
-                .loaded => |body| {
+                .loaded => |compiled| {
-                    slot.* = body;
+                    try builder.addSource(index, compiled);
+                    outcome.* = .loaded;
                     loaded += 1;
                 },
-                // Not loadable and therefore not enforced. Saying so here is
+                // Not loadable and therefore not enforced.
-                // what keeps `Snapshot.build`'s `MissingCompiledSource` for the
+                .failed => |failure| {
-                // case it is meant for: a caller that forgot to read a body.
+                    builder.skipSource(index);
-                .disabled, .failed => {
+                    outcome.* = .{ .failed = failure };
-                    slot.* = null;
-                    source.enabled = false;
                 },
             }
+            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
+    /// The reason for a failure is in the status table, not in the return
-    /// content (equal checksum) and every recorded failure return `false`; the
+    /// value.
-    /// 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
+        // Boot already loaded the compiled files before the listeners started,
-        // exist is a read. A full disk must not cost the household its
+        // and its status table is what the refresh decisions below read. Only
-        // filtering as well as its downloads.
+        // a boot load that published nothing is retried here, ahead of the
-        try self.reloadCollecting(io);
+        // 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});
-    applyLoadOutcomes(&statuses, &rows, &.{.{ .loaded = body }});
 
     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});
-    applyLoadOutcomes(&statuses, &rows, &.{.{ .loaded = body }});
 
     try testing.expect(statuses[0].loaded);
     try testing.expectEqual(State.ok, statuses[0].state);

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
+    /// Builds an immutable snapshot from bodies the caller already holds, every
-    /// the caller may free the repository lists immediately afterwards.
+    /// one at once. A caller reading bodies from disk uses `Builder` instead,
-    /// Disabled sources are skipped entirely — they cost no memory.
+    /// 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);
+        var builder: Builder = try .init(gpa, input.sources, input.source_ids, input.seed);
-        std.debug.assert(input.source_ids.len == input.sources.len);
+        defer builder.deinit();
-
+        for (input.sources, 0..) |row, i| {
-        var arena_state: std.heap.ArenaAllocator = .init(gpa);
+            if (!row.enabled) continue;
-        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;
             if (i >= input.compiled.len) return error.MissingCompiledSource;
-            const bodies = input.compiled[i] orelse return error.MissingCompiledSource;
+            try builder.addSource(i, 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),
-            };
         }
-
+        return builder.finish(.{
-        const groups = try arena.alloc(Group, input.groups.len);
+            .groups = input.groups,
-        var group_rules: std.ArrayList(model.Rule) = .empty;
+            .group_ids = input.group_ids,
-        defer group_rules.deinit(gpa);
+            .group_sources = input.group_sources,
-        var group_sources: std.ArrayList(u32) = .empty;
+            .rules = input.rules,
-        defer group_sources.deinit(gpa);
+            .clients = input.clients,
-
+            .prefixes = input.prefixes,
-        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,
             .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| {