//! The blocklist manager (PLAN §4): the compiled files under //! `/blocklists/`, the refresh that produces them, the metadata //! columns it writes back, the snapshot built from them and the swap that //! publishes it. //! //! This is the only file in this milestone that touches both the database and //! the filesystem. Everything it composes — the parsers, the compiler, the //! domain sets, the rule sets and the snapshot — is pure and testable without //! either. //! //! **The swap is an `std.Io.RwLock`, not a lock-free pointer.** PLAN §7.3 says //! "readers lock-free"; this is a deliberate deviation. Freeing the old //! snapshot without a lock needs epoch-based reclamation or hazard pointers: a //! class of code that is very hard to get right and impossible to test //! convincingly, bought for a household resolver whose target is 100 qps. A //! shared lock held for the microseconds of one `evaluate` costs an uncontended //! atomic pair; the writer takes the exclusive lock only on a swap, which //! happens on refresh. The old snapshot is freed *after* `unlock` returns, and //! the `Handle` API makes "do not retain the pointer" the only shape a caller //! can write. //! //! The same lock guards the status table, which is written from the refresh //! task and read by the API. Both critical sections are short and hold no //! socket and no file, so the uncancelable lock forms are used: a lock this //! code takes is always released within a few instructions. //! //! Two more locks serialize the writers, and they divide the work by how long //! it takes. //! //! `writer_lock` covers what a writer does to the *published* state: build a //! snapshot, install the compiled files, write the runtime columns, record a //! status. Two concurrent reloads would otherwise compute the same generation //! and each destroy a snapshot the other had just published. Every section it //! guards is bounded by local work — a read of the compiled files at worst — //! so a web mutation that ends in `reload` never waits out a download. //! //! `refresh_lock` covers what a writer does *before* it has anything to //! publish: the download of one source, at up to 300 s each, and the compile //! that follows it. It also covers blocklist-directory maintenance, because //! those stages are the only writers of `.raw.tmp` / `.list.tmp` / //! `.wild.tmp` / `.allow.tmp` and `pruneOrphans` must not sweep the //! temporaries of a refresh that is still running. Two concurrent refreshes //! would share the fetcher's buffers and, for one source, the same temporary //! paths. //! //! **Lock ordering: `refresh_lock` is never acquired while `writer_lock` is //! held.** A path that needs both takes `refresh_lock` first. The public entry //! points take what they need; the `*Locked` bodies assume it and never take //! it again, because neither mutex is reentrant. const std = @import("std"); const Allocator = std.mem.Allocator; const model = @import("../config/model.zig"); const safe_url = @import("../safe_url.zig"); const db = @import("../storage/db.zig"); const clients_repo = @import("../storage/repositories/clients_repo.zig"); const groups_repo = @import("../storage/repositories/groups_repo.zig"); const rules_repo = @import("../storage/repositories/rules_repo.zig"); const sources_repo = @import("../storage/repositories/sources_repo.zig"); const disk_monitor = @import("../storage/disk_monitor.zig"); const compiler = @import("compiler.zig"); const fetcher = @import("fetcher.zig"); const matcher = @import("matcher.zig"); const parsers = @import("parsers.zig"); const log = std.log.scoped(.blocklist_manager); const Sha256 = std.crypto.hash.sha2.Sha256; /// `SourceStatus.last_error` is fixed-size so the failure path allocates /// nothing. pub const max_error_len: usize = 128; /// `SourceStatus.url` is fixed-size so a copied status borrows nothing. A /// blocklist url longer than this is truncated in the status only; the row /// keeps it whole. /// /// The log form of a url is bounded separately by `safe_url.max_len`. The two /// numbers agree today and answer different questions; neither follows the /// other. pub const max_url_len: usize = 255; /// A compiled body larger than this is refused at load. A source that reaches /// it produced more than `fetcher.max_body_bytes` of names, which cannot /// happen from a download this fetcher performed. pub const max_compiled_bytes: usize = 128 * 1024 * 1024; /// Buffer size for every file stream this file opens. One buffer is live per /// stage, and the stages do not overlap. const io_buf_len: usize = 64 * 1024; /// Holds the sniff sample: `parsers.sample_lines` lines of at most /// `compiler.max_line_len` bytes, each with its newline. A fixed byte window /// would be spent by a handful of legal 4096-byte comment lines and the format /// would then be decided by almost no data. const sample_buf_len: usize = parsers.sample_lines * (compiler.max_line_len + 1); /// `` is at most 20 characters and the longest suffix is `.allow.tmp`. const name_buf_len: usize = 48; /// How one blocklist source is named in a log line: by its row id and its name, /// which are its own identity, and by its redacted url, which says where it /// points and nothing more. /// /// The url used to carry the identity here on its own. It cannot: `safe_url` /// drops the path, because a path segment is a place an operator's token lives, /// and two sources on one host are told apart by exactly that path. The id and /// the name are on the row every one of these lines already holds, they are /// what the API and the web UI show, and neither can leak what the url holds. /// The name is escaped for the same reason the url is — both are database text /// and a newline in either would forge a log line. It carries its own quotes, /// out of `safe_url.quoteText`, because a quote this format string added would /// be a quote the name could close: `ads' (https://decoy.example) --` would then /// read as a source pointing somewhere it does not. const SourceLabel = struct { id: i64, name: []const u8, url: []const u8, fn of(row: sources_repo.SourceRow) SourceLabel { return .{ .id = row.id, .name = row.name, .url = row.url }; } pub fn format(self: SourceLabel, w: *std.Io.Writer) std.Io.Writer.Error!void { try w.print("source {d} {f} {f}", .{ self.id, safe_url.quoteText(self.name), safe_url.redactQuoted(self.url), }); } }; pub const Paths = struct { /// ``, owned by the caller and left open for the manager's life. dir: std.Io.Dir, subdir: []const u8 = "blocklists", }; /// Where one source stands. `.never_fetched` is the state of a source that has /// no compiled files and no stored checksum, which is a fresh install rather /// than a failure. `.no_valid_entries` is a download that compiled cleanly and /// yielded nothing usable — an error page or a compressed body, not a /// blocklist. pub const State = enum { ok, never_fetched, fetch_failed, compile_failed, no_valid_entries, load_failed, /// Whether this state was recorded by a refresh rather than by a load. /// A load outcome never overwrites one: the files a reload just read are /// exactly the files the failed refresh could not replace, and the operator /// still has to see why the update did not land. `SourceStatus.loaded` /// carries the other half — whether the source is filtering at all. pub fn isRefreshFailure(self: State) bool { return switch (self) { .fetch_failed, .compile_failed, .no_valid_entries => true, .ok, .never_fetched, .load_failed => false, }; } }; /// A status is a value with no borrowed memory, so a copy handed to the API /// outlives every reload. The url is held inline for that reason. pub const SourceStatus = struct { id: i64, state: State = .never_fetched, /// Whether this source's compiled files were read into the most recent /// snapshot build — that is, whether it is filtering right now. `state` /// describes the most recent attempt to *produce* those files, which is a /// different fact: a source whose refresh failed keeps serving what the /// refresh did not replace, and reads `.fetch_failed` with `loaded` set. loaded: bool = false, last_attempt: i64 = 0, last_success: i64 = 0, counts: compiler.Counts = .{}, /// A display copy of the source url, truncated at `max_url_len`. The whole /// url is in the `blocklist_sources` row this status shares an `id` with. url: [max_url_len]u8 = @splat(0), url_len: u8 = 0, last_error: [max_error_len]u8 = @splat(0), last_error_len: u8 = 0, pub fn errorText(self: *const SourceStatus) []const u8 { return self.last_error[0..self.last_error_len]; } pub fn urlText(self: *const SourceStatus) []const u8 { return self.url[0..self.url_len]; } fn setUrl(self: *SourceStatus, url: []const u8) void { const kept = @min(url.len, max_url_len); @memcpy(self.url[0..kept], url[0..kept]); @memset(self.url[kept..], 0); self.url_len = @intCast(kept); } fn fail(self: *SourceStatus, state: State, text: []const u8) void { self.state = state; const kept = @min(text.len, max_error_len); @memcpy(self.last_error[0..kept], text[0..kept]); @memset(self.last_error[kept..], 0); self.last_error_len = @intCast(kept); } fn succeed(self: *SourceStatus, at: i64, counts: compiler.Counts) void { self.state = .ok; self.counts = counts; self.last_success = at; self.last_error = @splat(0); self.last_error_len = 0; } }; /// The header every compiled file carries, ahead of the body. The `sha256` /// covers the `.list` body, then the `.wild` body, then the `.allow` body, and /// **not** the header, so it stays stable across a refetch of unchanged content /// while `fetched_at` moves. /// /// Each body is followed by a separator byte, so the digest identifies which /// body a name sits in rather than only which names were written. pub const Header = struct { url: []const u8, format: parsers.Format, fetched_at: i64, counts: compiler.Counts, /// 64 lowercase hex characters. checksum: []const u8, pub fn write(self: Header, w: *std.Io.Writer) std.Io.Writer.Error!void { try w.writeAll("# nxdns blocklist\n"); try w.print("# url {s}\n", .{self.url}); try w.print("# format {s}\n", .{@tagName(self.format)}); try w.print("# fetched_at {d}\n", .{self.fetched_at}); try w.print("# domains {d}\n", .{self.counts.domains}); try w.print("# wildcards {d}\n", .{self.counts.wildcards}); try w.print("# exceptions {d}\n", .{self.counts.exceptions}); try w.print("# skipped_regex {d}\n", .{self.counts.skipped_regex}); try w.print("# skipped_unsupported {d}\n", .{self.counts.skipped_unsupported}); try w.print("# invalid {d}\n", .{self.counts.invalid}); try w.print("# sha256 {s}\n", .{self.checksum}); } }; /// The body of a compiled file: everything after the leading `#` lines. A file /// with no header is all body, which is what makes a hand-written fixture a /// legal compiled file. pub fn stripHeader(bytes: []const u8) []const u8 { var rest = bytes; while (rest.len != 0 and rest[0] == '#') { const newline = std.mem.indexOfScalar(u8, rest, '\n') orelse return rest[rest.len..]; rest = rest[newline + 1 ..]; } return rest; } pub const Manager = struct { gpa: Allocator, database: *db.Db, paths: Paths, fetcher: *fetcher.Fetcher, update: model.BlocklistUpdate, /// Bounds one download. `std.http.Client` has no per-request deadline, so /// the fetch runs under `io.concurrent` against a sleep of this length. total_budget: std.Io.Clock.Duration, lock: std.Io.RwLock, /// Serializes everything that changes the published state — the snapshot, /// the compiled files, the runtime columns, the status table — against /// every other writer. Never taken by a reader. /// /// Lock ordering: `refresh_lock` is never acquired while this is held. A /// caller that needs both takes `refresh_lock` first. writer_lock: std.Io.Mutex, /// Serializes refresh passes against each other, and against the /// blocklist-directory maintenance in `pruneOrphans`. Held across a /// download and a compile, which `writer_lock` deliberately is not, so an /// unrelated `reload` never waits out a 300-second fetch. /// /// Lock ordering: this is taken first, and never while `writer_lock` is /// held. refresh_lock: std.Io.Mutex, current: ?*matcher.Snapshot, generation: u64, statuses: []SourceStatus, /// Owns the `statuses` table. The entries themselves borrow nothing. status_arena: std.heap.ArenaAllocator, /// The §11.6 disk gate (ruling 17). Set by the composition root after /// `init` and before `runScheduler` starts; null disables gating, which is /// what every test and `nxdns check` want. Only the scheduler consults it — /// see `refreshGated`. monitor: ?*disk_monitor.Monitor = null, /// Scheduled refresh passes skipped by the disk gate. The `/api/health` /// rollup reads it through `refreshesGated`. refreshes_gated: std.atomic.Value(u64) = .init(0), pub const Error = error{ OutOfMemory, Canceled, /// A filesystem operation on the blocklist directory failed. The /// concrete cause is logged at `warn` where it happens: this taxonomy /// would otherwise carry two dozen members no caller can act on /// differently. FileSystem, /// The `groups` table changed between listing the groups and reading /// their ids. Retrying the reload is the answer, and the caller is the /// only one that can decide to. GroupSetChanged, } || db.Error || matcher.Snapshot.Error; /// The result is not copyable afterwards: `status_arena` and `lock` are /// addressed through `self`. pub fn init( gpa: Allocator, database: *db.Db, paths: Paths, fetcher_ptr: *fetcher.Fetcher, update: model.BlocklistUpdate, total_budget: std.Io.Clock.Duration, ) Error!Manager { return .{ .gpa = gpa, .database = database, .paths = paths, .fetcher = fetcher_ptr, .update = update, .total_budget = total_budget, .lock = .init, .writer_lock = .init, .refresh_lock = .init, .current = null, .generation = 0, .statuses = &.{}, .status_arena = .init(gpa), }; } pub fn deinit(self: *Manager, io: std.Io) void { self.lock.lockUncancelable(io); const old = self.current; self.current = null; self.statuses = &.{}; self.lock.unlock(io); if (old) |snapshot| destroySnapshot(self.gpa, snapshot); self.status_arena.deinit(); self.* = undefined; } /// Reader side of the swap. The handle holds a shared lock: release it, and /// do not retain `snapshot` afterwards. pub const Handle = struct { snapshot: *const matcher.Snapshot, manager: *Manager, pub fn release(self: Handle, io: std.Io) void { self.manager.lock.unlockShared(io); } }; /// `null` before the first successful `reload`. The caller answers /// SERVFAIL, or forwards unfiltered, on its own policy — this file does not /// decide that. pub fn acquire(self: *Manager, io: std.Io) ?Handle { self.lock.lockSharedUncancelable(io); const snapshot = self.current orelse { self.lock.unlockShared(io); return null; }; return .{ .snapshot = snapshot, .manager = self }; } /// Copies the status table for the API and for `nxdns check`. Returns the /// number of entries written, which is `min(out.len, source count)`. /// /// The copies are self-contained: `SourceStatus` holds its url and its /// error text inline, so the caller may keep them for as long as it likes /// and a concurrent reload cannot pull memory out from under them. pub fn statusSnapshot(self: *Manager, io: std.Io, out: []SourceStatus) usize { self.lock.lockSharedUncancelable(io); defer self.lock.unlockShared(io); const kept = @min(out.len, self.statuses.len); @memcpy(out[0..kept], self.statuses[0..kept]); return kept; } // ----------------------------------------------------------------------- // reload // ----------------------------------------------------------------------- /// Reads the database and every compiled file, builds a snapshot and swaps /// it in. /// /// A source whose compiled files are missing, unreadable or checksum /// mismatched is marked `.load_failed` and left out of the snapshot rather /// than failing the whole reload: one bad file must not cost the operator /// every other list. `runScheduler` refreshes exactly those sources, so the /// state is recorded, surfaced and repaired, never silently accepted. /// /// A body that is present and checksum-clean but malformed fails the build /// (`error.NotSorted`), and the previously published snapshot keeps /// serving: nothing is swapped until the new snapshot exists. The status /// table keeps describing that snapshot too — the table is rebuilt off to /// the side and the load findings are written into it there, so a reload /// that never publishes changes neither. pub fn reload(self: *Manager, io: std.Io) Error!void { self.writer_lock.lockUncancelable(io); defer self.writer_lock.unlock(io); return self.reloadLocked(io); } fn reloadLocked(self: *Manager, io: std.Io) Error!void { var rows = try sources_repo.listSourceRows(self.database, self.gpa); defer rows.deinit(self.gpa); defer sources_repo.freeSourceRows(self.gpa, rows.items); // The table this reload will publish, built where no reader can see it. // It is installed in the swap below or freed unpublished, so a reload // that fails leaves the previous table describing the previous // snapshot — including the entry of a source deleted from the database, // which that snapshot still enforces. var candidate: ?StatusTable = try self.buildStatusTable(io, rows.items); errdefer if (candidate) |*table| table.deinit(); var dir = try self.openDir(io, .{}); defer dir.close(io); const sources = try self.gpa.alloc(model.BlocklistSource, rows.items.len); 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); } // 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 // serves is worse than one describing the previous one. const outcomes = try self.gpa.alloc(LoadOutcome, rows.items.len); 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; 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; }, } } var groups = try groups_repo.listGroups(self.database, self.gpa); defer groups.deinit(self.gpa); defer groups_repo.freeGroups(self.gpa, groups.items); const group_ids = try self.groupIds(groups.items); defer self.gpa.free(group_ids); var group_sources = try groups_repo.listGroupSources(self.database, self.gpa); defer group_sources.deinit(self.gpa); defer groups_repo.freeGroupSources(self.gpa, group_sources.items); var rule_rows = try rules_repo.listRules(self.database, self.gpa); defer rule_rows.deinit(self.gpa); defer rules_repo.freeRules(self.gpa, rule_rows.items); var clients = try clients_repo.listClients(self.database, self.gpa); defer clients.deinit(self.gpa); defer clients_repo.freeClients(self.gpa, clients.items); var prefixes = try clients_repo.listClientPrefixes(self.database, self.gpa); 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, .{ .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, }); // Read before the swap: once `current` points at it, this snapshot // belongs to the readers and to whichever writer replaces it next. const memory_bytes = snapshot.memoryBytes(); // The snapshot, the status table and the load facts land together, so a // reader never sees a status table describing anything but the // published snapshot. self.lock.lockUncancelable(io); const old = self.current; self.current = snapshot; self.generation = generation; applyLoadOutcomes(candidate.?.items, rows.items, outcomes); self.installStatuses(candidate.?); candidate = null; self.lock.unlock(io); // After `unlock`: no reader can still hold the old snapshot here, and // `writer_lock` keeps every other writer out of this sequence. if (old) |previous| destroySnapshot(self.gpa, previous); log.info("blocklist snapshot generation {d}: {d} of {d} sources loaded, {d} bytes", .{ generation, loaded, rows.items.len, memory_bytes, }); } /// What one enabled source contributes to the snapshot being built. Nothing /// here touches the status table: the outcome is data until the swap /// commits it. fn loadSource( self: *Manager, io: std.Io, dir: std.Io.Dir, row: sources_repo.SourceRow, bodies: *std.ArrayList([]u8), ) Error!LoadOutcome { const stored = row.checksum orelse // No stored checksum means no successful compile has ever // happened. A fresh install is here on every source. return .{ .failed = .{ .state = .never_fetched, .text = "" } }; var list_buf: [name_buf_len]u8 = undefined; var wild_buf: [name_buf_len]u8 = undefined; var allow_buf: [name_buf_len]u8 = undefined; const list_name = compiledName(&list_buf, row.id, ".list"); const wild_name = compiledName(&wild_buf, row.id, ".wild"); const allow_name = compiledName(&allow_buf, row.id, ".allow"); // Reserved before the reads, so no buffer can be orphaned by a failing // append: `bodies` owns each one from the moment it is read. try bodies.ensureUnusedCapacity(self.gpa, 3); // `error.Canceled` is the one-shot signal that this task is being torn // down, and it is consumed by whoever catches it. Recording it as a // load failure would spend it on a status row that reads "Canceled", // publish a snapshot with this source missing, and let the shutdown // carry on as if nothing had asked it to stop. const list_bytes = dir.readFileAlloc(io, list_name, self.gpa, .limited(max_compiled_bytes)) catch |err| { if (err == error.OutOfMemory) return error.OutOfMemory; if (err == error.Canceled) return error.Canceled; return loadFailure(row, list_name, err); }; bodies.appendAssumeCapacity(list_bytes); const wild_bytes = dir.readFileAlloc(io, wild_name, self.gpa, .limited(max_compiled_bytes)) catch |err| { if (err == error.OutOfMemory) return error.OutOfMemory; if (err == error.Canceled) return error.Canceled; return loadFailure(row, wild_name, err); }; bodies.appendAssumeCapacity(wild_bytes); // A missing `.allow` file is an empty allow body, not a failure. The // digest still covers three bodies, the third of them empty, so a // source whose list carries no `@@` line matches whether its empty // `.allow` file survived or not. const allow_bytes: []const u8 = blk: { const read = dir.readFileAlloc(io, allow_name, self.gpa, .limited(max_compiled_bytes)) catch |err| { if (err == error.OutOfMemory) return error.OutOfMemory; if (err == error.Canceled) return error.Canceled; if (err == error.FileNotFound) break :blk ""; return loadFailure(row, allow_name, err); }; bodies.appendAssumeCapacity(read); break :blk read; }; const list_body = stripHeader(list_bytes); const wild_body = stripHeader(wild_bytes); const allow_body = stripHeader(allow_bytes); // The checksum covers the three bodies together, so a crash between the // `replace` calls — a new `.list` beside an old `.wild` — is caught // here and refreshed, not served as a half-updated list. if (!std.mem.eql(u8, stored, &bodyChecksum(list_body, wild_body, allow_body))) { log.warn("blocklist {f}: compiled files do not match the stored checksum", .{SourceLabel.of(row)}); return .{ .failed = .{ .state = .load_failed, .text = "ChecksumMismatch" } }; } return .{ .loaded = .{ .list_body = list_body, .wild_body = wild_body, .allow_body = allow_body, } }; } // ----------------------------------------------------------------------- // refresh // ----------------------------------------------------------------------- /// 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 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 { self.refresh_lock.lockUncancelable(io); defer self.refresh_lock.unlock(io); return self.refreshSourceLocked(io, row); } /// 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 { // 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 // still being enforced. var status: SourceStatus = self.priorStatus(io, row.id) orelse .{ .id = row.id }; status.setUrl(row.url); status.last_attempt = std.Io.Clock.real.now(io).toSeconds(); var dir = try self.openDir(io, .{}); defer dir.close(io); var raw_buf: [name_buf_len]u8 = undefined; var list_tmp_buf: [name_buf_len]u8 = undefined; var wild_tmp_buf: [name_buf_len]u8 = undefined; var allow_tmp_buf: [name_buf_len]u8 = undefined; const raw_name = compiledName(&raw_buf, row.id, ".raw.tmp"); const tmp: TempNames = .{ .list = compiledName(&list_tmp_buf, row.id, ".list.tmp"), .wild = compiledName(&wild_tmp_buf, row.id, ".wild.tmp"), .allow = compiledName(&allow_tmp_buf, row.id, ".allow.tmp"), }; // Installed before the calls that create these files, not after: an // `error.Canceled` or `error.OutOfMemory` returned straight out of // `download` or `compileTo` would outrun a later `defer` and leave a // temporary behind. Deleting a name that was never created is a no-op. defer self.deleteQuietly(io, dir, raw_name); defer self.deleteQuietly(io, dir, tmp.list); defer self.deleteQuietly(io, dir, tmp.wild); defer self.deleteQuietly(io, dir, tmp.allow); // The half that takes the time: one download of up to `total_budget` // and one compile of everything it returned. `refresh_lock` alone is // held here, so a rule save, a settings change or any other web // mutation that ends in `reload` runs beside it instead of behind it. const prepared = try self.prepareRefresh(io, dir, row, &status, raw_name, tmp); // The half that publishes. The compiled files, the runtime columns and // the status entry land under one `writer_lock`, so a reload never // reads new files beside a status entry describing the previous ones. self.writer_lock.lockUncancelable(io); defer self.writer_lock.unlock(io); const replaced = try self.publishRefresh(io, dir, row, &status, prepared, tmp); self.commitStatus(io, status); return replaced; } /// Every enabled source, one at a time, then one `reload`. A failing source /// never stops the pass: it would hide every source behind it. /// /// This returns an error only when nothing could be done at all — out of /// memory, an unreachable database, an unusable blocklist directory. A /// source that failed to download or compile is a successful pass with a /// non-`ok` status. pub fn refreshAll(self: *Manager, io: std.Io) Error!void { self.refresh_lock.lockUncancelable(io); defer self.refresh_lock.unlock(io); var rows = try sources_repo.listSourceRows(self.database, self.gpa); defer rows.deinit(self.gpa); defer sources_repo.freeSourceRows(self.gpa, rows.items); try self.syncStatuses(io, rows.items); for (rows.items) |row| { if (!row.enabled) continue; _ = try self.refreshSourceLocked(io, row); } // `reload` takes `writer_lock`, which the pass has been careful not to // hold: the order is `refresh_lock` first, always. return self.reload(io); } /// The three temporary files one refresh compiles into, before the header /// is prepended and each is renamed over the file it replaces. const TempNames = struct { list: []const u8, wild: []const u8, allow: []const u8, }; /// What the fetch-and-compile half of a refresh produced. `.failed` needs /// no publish and has already recorded why in the status entry. const Prepared = union(enum) { failed, compiled: struct { format: parsers.Format, result: compiler.Result, }, }; /// Downloads one source and compiles it into the temporary files. /// /// Assumes `refresh_lock` and must not be called with `writer_lock` held: /// this is the part that takes seconds, and nothing here touches the /// published state. fn prepareRefresh( self: *Manager, io: std.Io, dir: std.Io.Dir, row: sources_repo.SourceRow, status: *SourceStatus, raw_name: []const u8, tmp: TempNames, ) Error!Prepared { self.download(io, dir, raw_name, row) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Canceled => return error.Canceled, else => { self.reportFetchFailure(row, status, err); return .failed; }, }; const format = self.detectFormat(io, dir, raw_name) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Canceled => return error.Canceled, else => { self.reportCompileFailure(row, status, err); return .failed; }, }; const result = self.compileTo(io, dir, raw_name, format, tmp) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Canceled => return error.Canceled, else => { self.reportCompileFailure(row, status, err); return .failed; }, }; if (rejectedWithoutEntries(result.counts)) { self.reportEmptyCompile(row, status, result.counts); return .failed; } return .{ .compiled = .{ .format = format, .result = result } }; } /// 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. fn publishRefresh( self: *Manager, io: std.Io, dir: std.Io.Dir, row: sources_repo.SourceRow, status: *SourceStatus, prepared: Prepared, tmp: TempNames, ) Error!bool { const compiled = switch (prepared) { .failed => return false, .compiled => |value| value, }; const now = std.Io.Clock.real.now(io).toSeconds(); // Recompiling identical content into new files would invalidate the // snapshot for nothing. The files on disk must actually carry that // content: if a reload found them corrupt and excluded the source, a // re-download of unchanged upstream bytes is the one chance to repair // them, and skipping the rewrite here would leave filtering off for // good. if (row.checksum) |stored| { if (std.mem.eql(u8, stored, &compiled.result.checksum) and self.diskBodiesMatch(io, dir, row.id, stored)) { // Every count comes from the compile that just ran, not from // the row. The two skip counts have to: a skipped line lands in // no body, so a list that changed only its regex or // browser-syntax lines reaches here with a stale row. The three // written counts equal the row's anyway once the digest is // framed, so reading them from the compile costs nothing and // leaves no field whose freshness rests on an argument about // what the checksum covers. try sources_repo.updateSourceStats(self.database, row.id, .{ .last_updated = now, .domain_count = compiled.result.counts.domains, .wildcard_count = compiled.result.counts.wildcards, .exception_count = compiled.result.counts.exceptions, .skipped_regex_count = compiled.result.counts.skipped_regex, .skipped_unsupported_count = compiled.result.counts.skipped_unsupported, .checksum = stored, }); status.succeed(now, compiled.result.counts); return false; } } const header: Header = .{ .url = row.url, .format = compiled.format, .fetched_at = now, .counts = compiled.result.counts, .checksum = &compiled.result.checksum, }; self.publish(io, dir, row.id, header, tmp) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, error.Canceled => return error.Canceled, else => { self.reportCompileFailure(row, status, err); return false; }, }; try sources_repo.updateSourceStats(self.database, row.id, .{ .last_updated = now, .domain_count = compiled.result.counts.domains, .wildcard_count = compiled.result.counts.wildcards, .exception_count = compiled.result.counts.exceptions, .skipped_regex_count = compiled.result.counts.skipped_regex, .skipped_unsupported_count = compiled.result.counts.skipped_unsupported, .checksum = &compiled.result.checksum, }); status.succeed(now, compiled.result.counts); return true; } /// The body goes to a temporary file, never to memory: `max_body_bytes` is /// 64 MiB and the memory budget has no room for it beside two snapshots. /// /// It takes the whole row rather than the url alone because its two log /// lines name the source by its id and name, which only the row carries. fn download( self: *Manager, io: std.Io, dir: std.Io.Dir, raw_name: []const u8, row: sources_repo.SourceRow, ) !void { const file = try dir.createFile(io, raw_name, .{ .permissions = .fromMode(0o600) }); defer file.close(io); const buffer = try self.gpa.alloc(u8, io_buf_len); defer self.gpa.free(buffer); var fw = file.writer(io, buffer); const result = self.fetchWithin(io, row.url, &fw.interface) catch |err| { // `fetcher.Error.Unexpected` is what a failing sink surfaces as; // the concrete cause is on this writer, which the fetcher does not // own. if (fw.err) |cause| return cause; if (err == error.HttpStatus) { if (self.fetcher.last_status) |status| { log.warn("blocklist {f}: http status {d}", .{ SourceLabel.of(row), @intFromEnum(status) }); } } return err; }; try fw.interface.flush(); // The compile reads this file back; the bytes must be there, not in a // buffer this function is about to drop. try file.sync(io); log.debug("blocklist {f}: downloaded {d} bytes", .{ SourceLabel.of(row), result.bytes_read }); } /// `std.http.Client` has no per-request deadline, so the whole exchange /// races a sleep and the loser is canceled — milestone 3's pattern. fn fetchWithin( self: *Manager, io: std.Io, url: []const u8, w: *std.Io.Writer, ) fetcher.Error!fetcher.Result { var outcomes: [2]Outcome = undefined; var race: std.Io.Select(Outcome) = .init(io, &outcomes); defer race.cancelDiscard(); race.concurrent(.fetch, fetcher.Fetcher.fetch, .{ self.fetcher, io, url, w }) catch |err| switch (err) { error.ConcurrencyUnavailable => return error.SystemResources, }; race.concurrent(.expiry, expire, .{ io, self.total_budget }) catch |err| switch (err) { error.ConcurrencyUnavailable => return error.SystemResources, }; switch (try race.await()) { .fetch => |result| return result, .expiry => |result| { // A canceled sleep means this task is being torn down, not that // the download is slow. try result; return error.Timeout; }, } } fn detectFormat( self: *Manager, io: std.Io, dir: std.Io.Dir, raw_name: []const u8, ) !parsers.Format { const file = try dir.openFile(io, raw_name, .{}); defer file.close(io); const buffers = try self.gpa.alloc(u8, io_buf_len + sample_buf_len); defer self.gpa.free(buffers); var fr = file.reader(io, buffers[0..io_buf_len]); var sample: std.Io.Writer = .fixed(buffers[io_buf_len..]); collectSample(&fr.interface, &sample) catch |err| switch (err) { error.ReadFailed => return fr.err orelse err, // `sample_buf_len` holds every line `collectSample` can emit, so a // full buffer means the sample is complete. error.WriteFailed => {}, }; return parsers.detectFormat(sample.buffered()); } /// Compiles into three plain temporary files. The compiled bodies cannot go /// straight into the final files: the header carries counts that only exist /// once the whole input has been compiled, and the loader requires the /// header first. fn compileTo( self: *Manager, io: std.Io, dir: std.Io.Dir, raw_name: []const u8, format: parsers.Format, tmp: TempNames, ) !compiler.Result { const raw = try dir.openFile(io, raw_name, .{}); defer raw.close(io); const list_file = try dir.createFile(io, tmp.list, .{ .permissions = .fromMode(0o600) }); defer list_file.close(io); const wild_file = try dir.createFile(io, tmp.wild, .{ .permissions = .fromMode(0o600) }); defer wild_file.close(io); const allow_file = try dir.createFile(io, tmp.allow, .{ .permissions = .fromMode(0o600) }); defer allow_file.close(io); const buffers = try self.gpa.alloc(u8, 4 * io_buf_len); defer self.gpa.free(buffers); var fr = raw.reader(io, buffers[0..io_buf_len]); var list_w = list_file.writer(io, buffers[io_buf_len .. 2 * io_buf_len]); var wild_w = wild_file.writer(io, buffers[2 * io_buf_len .. 3 * io_buf_len]); var allow_w = allow_file.writer(io, buffers[3 * io_buf_len ..]); const result = compiler.compile( self.gpa, &fr.interface, format, &list_w.interface, &wild_w.interface, &allow_w.interface, ) catch |err| switch (err) { // `compiler.Error` names the direction; the concrete cause is on // the stream that failed. error.ReadFailed => return fr.err orelse err, error.WriteFailed => return list_w.err orelse (wild_w.err orelse (allow_w.err orelse err)), else => return err, }; try list_w.interface.flush(); try wild_w.interface.flush(); try allow_w.interface.flush(); try list_file.sync(io); try wild_file.sync(io); try allow_file.sync(io); return result; } /// Writes header + body into each final file through `createFileAtomic` + /// `replace`, so a crash mid-write can never leave a half-list that would /// load as a valid, shorter blocklist. fn publish( self: *Manager, io: std.Io, dir: std.Io.Dir, id: i64, header: Header, tmp: TempNames, ) !void { const buffers = try self.gpa.alloc(u8, 2 * io_buf_len); defer self.gpa.free(buffers); var list_buf: [name_buf_len]u8 = undefined; var wild_buf: [name_buf_len]u8 = undefined; var allow_buf: [name_buf_len]u8 = undefined; try publishOne(io, dir, compiledName(&list_buf, id, ".list"), tmp.list, header, buffers); try publishOne(io, dir, compiledName(&wild_buf, id, ".wild"), tmp.wild, header, buffers); try publishOne(io, dir, compiledName(&allow_buf, id, ".allow"), tmp.allow, header, buffers); } fn publishOne( io: std.Io, dir: std.Io.Dir, dest: []const u8, body_name: []const u8, header: Header, buffers: []u8, ) !void { const body = try dir.openFile(io, body_name, .{}); defer body.close(io); var af = try dir.createFileAtomic(io, dest, .{ .permissions = .fromMode(0o600), .replace = true, }); defer af.deinit(io); var fr = body.reader(io, buffers[0..io_buf_len]); var fw = af.file.writer(io, buffers[io_buf_len..]); header.write(&fw.interface) catch return fw.err orelse error.WriteFailed; _ = fr.interface.streamRemaining(&fw.interface) catch return fr.err orelse (fw.err orelse error.WriteFailed); fw.interface.flush() catch return fw.err orelse error.WriteFailed; // Before `replace`, which closes the file: the rename must publish // durable bytes, not an empty file with the content still in the page // cache. try af.file.sync(io); try af.replace(io); } /// Whether the compiled files on disk hash to `expected`. A missing, /// unreadable or corrupt file answers false, which sends the caller down /// the rewrite path — the only path that can repair it. /// /// A missing `.allow` file is the one exception, and it is the same one /// `loadSource` makes: it reads as an empty allow body, which is what a list /// with no `@@` line compiles to anyway. Answering false there would rewrite /// such a list on every refresh for no change in content. fn diskBodiesMatch(self: *Manager, io: std.Io, dir: std.Io.Dir, id: i64, expected: []const u8) bool { var list_buf: [name_buf_len]u8 = undefined; var wild_buf: [name_buf_len]u8 = undefined; var allow_buf: [name_buf_len]u8 = undefined; const limit: std.Io.Limit = .limited(max_compiled_bytes); const list_bytes = dir.readFileAlloc(io, compiledName(&list_buf, id, ".list"), self.gpa, limit) catch return false; defer self.gpa.free(list_bytes); const wild_bytes = dir.readFileAlloc(io, compiledName(&wild_buf, id, ".wild"), self.gpa, limit) catch return false; defer self.gpa.free(wild_bytes); const allow_bytes = dir.readFileAlloc(io, compiledName(&allow_buf, id, ".allow"), self.gpa, limit) catch |err| if (err == error.FileNotFound) @as([]u8, &.{}) else return false; defer self.gpa.free(allow_bytes); return compiledBodiesMatch(list_bytes, wild_bytes, allow_bytes, expected); } fn reportFetchFailure( self: *Manager, row: sources_repo.SourceRow, status: *SourceStatus, err: anyerror, ) void { _ = self; log.warn("blocklist {f}: download failed: {s}", .{ SourceLabel.of(row), @errorName(err) }); status.fail(.fetch_failed, @errorName(err)); } fn reportCompileFailure( self: *Manager, row: sources_repo.SourceRow, status: *SourceStatus, err: anyerror, ) void { _ = self; log.warn("blocklist {f}: compile failed: {s}", .{ SourceLabel.of(row), @errorName(err) }); status.fail(.compile_failed, @errorName(err)); } fn reportEmptyCompile( self: *Manager, row: sources_repo.SourceRow, status: *SourceStatus, counts: compiler.Counts, ) void { _ = self; var buf: [max_error_len]u8 = undefined; const text = std.fmt.bufPrint( &buf, "NoValidEntries invalid={d} unsupported={d} long_lines={d}", .{ counts.invalid, counts.skipped_unsupported, counts.long_lines }, ) catch "NoValidEntries"; log.warn("blocklist {f}: {s}", .{ SourceLabel.of(row), text }); status.fail(.no_valid_entries, text); } // ----------------------------------------------------------------------- // scheduling // ----------------------------------------------------------------------- /// Loads at startup, refreshes only what needs it, then sleeps /// `update.interval_hours` between full passes. Returns on /// `error.Canceled`. /// /// A cold restart must not re-download every list and a boot loop must not /// become a download loop, so the startup pass refreshes a source only when /// it has no usable compiled files or its `last_updated` is older than the /// interval. /// /// `update.enabled == false` stops after the startup pass; manual refresh /// through `refreshAll` still works. pub fn runScheduler(self: *Manager, io: std.Io) std.Io.Cancelable!void { // Ahead of the pass, not after it. This is the sweep that collects what // a killed process left behind: a `.raw.tmp` as large as the body the // dead refresh was writing, and the compiled files of a source deleted // while the server was down. Both are bytes the pass below is about to // ask the same filesystem for. try self.sweepOrphans(io); self.startupPass(io) catch |err| switch (err) { error.Canceled => return error.Canceled, else => log.warn("blocklist startup pass failed: {s}", .{@errorName(err)}), }; if (!self.update.enabled) return; // `boot` rather than `awake`: a box that suspends overnight should // still see its daily interval elapse. const interval: std.Io.Clock.Duration = .{ .raw = .fromSeconds(model.updateIntervalSeconds(self.update)), .clock = .boot, }; while (true) { try interval.sleep(io); // Ahead of the gate as well as ahead of the pass: the sweep only // unlinks, so it is the one thing here that can give a critically // full disk room back, and gating it would keep the residue that // helped fill the disk in the first place. try self.sweepOrphans(io); if (self.refreshGated()) continue; self.refreshAll(io) catch |err| switch (err) { error.Canceled => return error.Canceled, else => log.warn("blocklist refresh pass failed: {s}", .{@errorName(err)}), }; } } /// `pruneOrphans` with its failure absorbed. Leftover bytes under /// `/blocklists/` are not an outage, and a sweep that could not /// read the directory must not cost the household the refresh pass behind /// it — let alone the server. Cancellation is the one outcome that /// propagates, because it means shutdown. /// /// Taken from outside every `*Locked` body: `pruneOrphans` takes both /// writer mutexes itself and neither is reentrant. fn sweepOrphans(self: *Manager, io: std.Io) std.Io.Cancelable!void { self.pruneOrphans(io) catch |err| switch (err) { error.Canceled => return error.Canceled, else => log.warn("pruning orphaned blocklist files failed: {s}", .{@errorName(err)}), }; } /// The §11.6 gate, consulted by scheduled passes only (ruling 17). A /// download writes tens of megabytes into the blocklist directory and the /// compile writes as much again, which is exactly the "non-essential write" /// a critically full disk must not take. /// /// `reload` and `refreshAll` are deliberately not gated: both are operator /// actions (the composition root's startup load, `POST /// /api/blocklists/update`), /// and an operator who asks for a refresh on a full disk has asked for it. /// /// Counting happens here, so a caller cannot skip a pass without recording /// it. One `warn` line per skipped pass — at a 24-hour interval that is one /// line a day, and the disk monitor already logs the state change itself. fn refreshGated(self: *Manager) bool { const monitor = self.monitor orelse return false; if (monitor.writesAllowed()) return false; _ = self.refreshes_gated.fetchAdd(1, .monotonic); log.warn("free space is critical; skipping the scheduled blocklist refresh", .{}); return true; } /// Scheduled refresh passes the disk gate has skipped. pub fn refreshesGated(self: *const Manager) u64 { return self.refreshes_gated.load(.monotonic); } fn startupPass(self: *Manager, io: std.Io) Error!void { self.refresh_lock.lockUncancelable(io); defer self.refresh_lock.unlock(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.reload(io); if (self.refreshGated()) return; var rows = try sources_repo.listSourceRows(self.database, self.gpa); defer rows.deinit(self.gpa); defer sources_repo.freeSourceRows(self.gpa, rows.items); const now = std.Io.Clock.real.now(io).toSeconds(); var refreshed = false; for (rows.items) |row| { if (!row.enabled) continue; if (!self.needsRefresh(io, row, now)) continue; if (try self.refreshSourceLocked(io, row)) refreshed = true; } if (refreshed) try self.reload(io); } fn needsRefresh(self: *Manager, io: std.Io, row: sources_repo.SourceRow, now: i64) bool { self.lock.lockSharedUncancelable(io); const state: State = state: { for (self.statuses) |status| { if (status.id == row.id) break :state status.state; } break :state .never_fetched; }; self.lock.unlockShared(io); if (state != .ok) return true; const last = row.last_updated orelse return true; // The Pi has no RTC, so a fetch stamped while the clock ran ahead of // real time (a pre-NTP boot, a restored image) leaves a `last_updated` // in the future. Plain interval arithmetic would then suspend every // refresh until real time caught up with the poison stamp, and the // reconcile engine preserves runtime columns faithfully, so nothing // else would ever clear it. A stamp from the future is not evidence of // a recent fetch. if (last > now) return true; return now - last >= model.updateIntervalSeconds(self.update); } // ----------------------------------------------------------------------- // orphans // ----------------------------------------------------------------------- /// Deletes the compiled files and the leftover temporaries whose id is no /// longer a `blocklist_sources` row. Files of a live source are left alone, /// whatever their state. /// /// Three callers, and between them they cover every way an orphan is made: /// `runScheduler` sweeps once before its startup pass — the residue of a /// process that was killed mid-refresh, and of a source deleted while the /// server was down — and again before each scheduled pass; the /// `DELETE /api/blocklists/{id}` handler sweeps as soon as it has removed /// the row, so the directory follows the table an operator can see instead /// of waiting out `blocklist_update.interval_hours`. /// /// It is safe to call on a fresh install: `openDir` creates /// `/blocklists/` if nothing has yet, and an empty directory /// sweeps to nothing. pub fn pruneOrphans(self: *Manager, io: std.Io) Error!void { // `refresh_lock` first, and for the reason it exists: the download and // the compile are the only writers of `.raw.tmp`, `.list.tmp`, // `.wild.tmp` and `.allow.tmp`, and they hold it for as long as they // run. Without it here, a source deleted through the API would sweep // the temporaries of a refresh still writing them — the row is gone, so // nothing else in this function would spare them — and the pass would // fail on a raw file that vanished under it. // // `writer_lock` second, in the one order this file ever takes them, // because the rows this reads and the compiled files it deletes are // what a reload is reading. self.refresh_lock.lockUncancelable(io); defer self.refresh_lock.unlock(io); self.writer_lock.lockUncancelable(io); defer self.writer_lock.unlock(io); var rows = try sources_repo.listSourceRows(self.database, self.gpa); defer rows.deinit(self.gpa); defer sources_repo.freeSourceRows(self.gpa, rows.items); var dir = try self.openDir(io, .{ .iterate = true }); defer dir.close(io); // The names are collected first: `Entry.name` is invalidated by the // next `next`, and deleting under an open cursor is not defined. var doomed: std.ArrayList([]u8) = .empty; defer { for (doomed.items) |item| self.gpa.free(item); doomed.deinit(self.gpa); } var it = dir.iterate(); while (true) { const entry = it.next(io) catch |err| switch (err) { error.Canceled => return error.Canceled, else => { log.warn("pruning blocklists: reading the directory failed: {s}", .{@errorName(err)}); return error.FileSystem; }, } orelse break; if (entry.kind != .file) continue; const id = sourceFileId(entry.name) orelse continue; if (containsId(rows.items, id)) continue; try doomed.append(self.gpa, try self.gpa.dupe(u8, entry.name)); } for (doomed.items) |name| { self.deleteQuietly(io, dir, name); log.info("pruned orphaned blocklist file {s}", .{name}); } } // ----------------------------------------------------------------------- // internals // ----------------------------------------------------------------------- /// Builds the status table for `rows`, carrying every existing entry over /// by row id so a recorded failure survives. Nothing is published: the /// caller either installs the result or frees it, which is what lets /// `reloadLocked` decide only once its snapshot exists. fn buildStatusTable( self: *Manager, io: std.Io, rows: []const sources_repo.SourceRow, ) Error!StatusTable { var fresh: std.heap.ArenaAllocator = .init(self.gpa); errdefer fresh.deinit(); const arena = fresh.allocator(); // The published table is copied under the lock. Reading it unlocked // would race the writer that replaces it — and the arena its entries // live in is freed by whoever installs what this builds. const previous = previous: { self.lock.lockSharedUncancelable(io); defer self.lock.unlockShared(io); break :previous try arena.dupe(SourceStatus, self.statuses); }; const table = try arena.alloc(SourceStatus, rows.len); mergeStatuses(table, rows, previous); return .{ .arena = fresh, .items = table }; } /// Publishes a built table and frees the one it replaces. The caller holds /// the exclusive lock, so no reader is inside the old table. fn installStatuses(self: *Manager, table: StatusTable) void { self.status_arena.deinit(); self.status_arena = table.arena; self.statuses = table.items; } /// Rebuilds and publishes the status table from the current source set. /// /// `refreshAll` calls this before it refreshes anything: a source added /// since the last reload needs an entry to record its outcome in, and the /// API has to see the pass advance while it runs. `reloadLocked` does not /// call it — a reload publishes its table together with the snapshot that /// table describes. fn syncStatuses(self: *Manager, io: std.Io, rows: []const sources_repo.SourceRow) Error!void { const table = try self.buildStatusTable(io, rows); self.lock.lockUncancelable(io); defer self.lock.unlock(io); self.installStatuses(table); } /// The recorded entry for one source, or `null` when the table has none. fn priorStatus(self: *Manager, io: std.Io, id: i64) ?SourceStatus { self.lock.lockSharedUncancelable(io); defer self.lock.unlockShared(io); for (self.statuses) |entry| { if (entry.id == id) return entry; } return null; } fn commitStatus(self: *Manager, io: std.Io, status: SourceStatus) void { self.lock.lockUncancelable(io); defer self.lock.unlock(io); for (self.statuses) |*entry| { if (entry.id != status.id) continue; // `loaded` is the reload's fact, not the refresh's: the files this // refresh wrote are not in a snapshot until the next reload reads // them. const loaded = entry.loaded; entry.* = status; entry.loaded = loaded; return; } // No entry of that id: the source was added or deleted between the // table this pass started from and this commit. The outcome is lost // either way — the next reload rebuilds the table from the rows — but // a failure that disappears without a line is the one thing milestone // 5 says never happens. log.warn("blocklist source {d}: no status entry to record the refresh outcome in", .{status.id}); } /// One id per group, in `listGroups` order. fn groupIds(self: *Manager, groups: []const model.Group) Error![]i64 { const out = try self.gpa.alloc(i64, groups.len); errdefer self.gpa.free(out); for (out, groups) |*slot, group| { slot.* = try groups_repo.groupId(self.database, group.name) orelse return error.GroupSetChanged; } return out; } fn openDir(self: *Manager, io: std.Io, options: std.Io.Dir.OpenOptions) Error!std.Io.Dir { _ = self.paths.dir.createDirPathStatus(io, self.paths.subdir, .fromMode(0o700)) catch |err| switch (err) { error.Canceled => return error.Canceled, else => { log.warn("creating {s} failed: {s}", .{ self.paths.subdir, @errorName(err) }); return error.FileSystem; }, }; return self.paths.dir.openDir(io, self.paths.subdir, options) catch |err| switch (err) { error.Canceled => return error.Canceled, else => { log.warn("opening {s} failed: {s}", .{ self.paths.subdir, @errorName(err) }); return error.FileSystem; }, }; } /// A temporary that cannot be removed is not a failure of the operation /// that made it, but it is not nothing either: it is left visible. fn deleteQuietly(self: *Manager, io: std.Io, dir: std.Io.Dir, name: []const u8) void { _ = self; dir.deleteFile(io, name) catch |err| switch (err) { error.FileNotFound => {}, else => log.warn("deleting {s} failed: {s}", .{ name, @errorName(err) }), }; } }; /// A status table and the arena holding it. Until `installStatuses` takes it, /// it is a candidate nobody can see, and `deinit` frees it whole. const StatusTable = struct { arena: std.heap.ArenaAllocator, items: []SourceStatus, fn deinit(self: *StatusTable) void { self.arena.deinit(); self.items = &.{}; } }; /// Fills `table` with one entry per row, carrying an entry of the same row id /// over from `previous`. A source deleted since `previous` was built is gone; a /// source added since starts blank. `previous` is only read, so the caller's /// published table is untouched by this. fn mergeStatuses( table: []SourceStatus, rows: []const sources_repo.SourceRow, previous: []const SourceStatus, ) void { for (table, rows) |*status, row| { status.* = .{ .id = row.id }; for (previous) |prior| { if (prior.id != row.id) continue; status.* = prior; break; } // After the carry-over: a url edited on the row wins over the one the // prior entry recorded. status.setUrl(row.url); } } /// What one `reload` found for one source. The texts are static, so an outcome /// borrows nothing and stays valid until the swap that commits it. 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, /// It is not in the snapshot, for this reason. failed: struct { state: State, text: []const u8 }, }; fn loadFailure(row: sources_repo.SourceRow, file_name: []const u8, err: anyerror) LoadOutcome { log.warn("blocklist {f}: reading {s} failed: {s}", .{ SourceLabel.of(row), file_name, @errorName(err) }); return .{ .failed = .{ .state = .load_failed, .text = @errorName(err) } }; } /// Writes one reload's findings into the status table. The caller holds the /// exclusive lock: this runs inside the swap so the table and the published /// snapshot describe the same thing. /// /// `rows` and `outcomes` are parallel. A source with no entry is one the table /// was not synced for, which cannot happen from `reloadLocked` and is skipped /// rather than asserted, because the table is rebuilt by row id. fn applyLoadOutcomes( statuses: []SourceStatus, rows: []const sources_repo.SourceRow, outcomes: []const LoadOutcome, ) void { for (rows, outcomes) |row, outcome| { const entry = entryFor(statuses, row.id) orelse continue; switch (outcome) { // A source disabled since it last loaded is no longer filtering, // and its recorded state describes files nothing reads. .disabled => entry.loaded = false, .loaded => { entry.loaded = true; // Two states survive a successful load. `.ok`, because a // refresh in this process already filled the counters the // compile produced and the five database columns are a subset // of them. And any refresh failure, because the files that just // loaded are exactly the ones the failed refresh could not // replace, so the operator must still see why. if (entry.state == .ok or entry.state.isRefreshFailure()) continue; entry.succeed(row.last_updated orelse 0, .{ .domains = countOf(row.domain_count), .wildcards = countOf(row.wildcard_count), .exceptions = countOf(row.exception_count), .skipped_regex = countOf(row.skipped_regex_count), .skipped_unsupported = countOf(row.skipped_unsupported_count), }); }, .failed => |reason| { entry.loaded = false; // The state follows only when nothing more informative is // there: a refresh failure already says why the files are // missing or stale, and `loaded` already says they are not // filtering. if (entry.state.isRefreshFailure()) continue; entry.fail(reason.state, reason.text); }, } } } fn entryFor(statuses: []SourceStatus, id: i64) ?*SourceStatus { for (statuses) |*entry| { if (entry.id == id) return entry; } return null; } const Outcome = union(enum) { fetch: fetcher.Error!fetcher.Result, expiry: std.Io.Cancelable!void, }; fn expire(io: std.Io, duration: std.Io.Clock.Duration) std.Io.Cancelable!void { return duration.sleep(io); } /// A counter column read back from the database. It is `NOT NULL DEFAULT 0` and /// only this file writes it, so a value outside `u32` means the row was edited /// behind nxdns's back; the status reports 0 rather than trapping. fn countOf(value: i64) u32 { return std.math.cast(u32, value) orelse 0; } fn destroySnapshot(gpa: Allocator, snapshot: *matcher.Snapshot) void { snapshot.deinit(); gpa.destroy(snapshot); } /// Copies the lines `parsers.detectFormat` would count — neither blank nor a /// comment — until it has `parsers.sample_lines` of them, and writes them to /// `w` newline-separated. Sampling by line rather than by a byte window is what /// keeps a run of long comment lines from deciding the format: `detectFormat` /// reads exactly these lines and ignores everything this drops. /// /// A line over `compiler.max_line_len` is skipped, as the compiler skips it. fn collectSample(r: *std.Io.Reader, w: *std.Io.Writer) error{ ReadFailed, WriteFailed }!void { var considered: usize = 0; while (considered < parsers.sample_lines) { const event = (try parsers.nextBoundedLine(r, compiler.max_line_len)) orelse return; const raw = switch (event) { .long_line => continue, .line => |line| line, }; const line = std.mem.trim(u8, raw, &std.ascii.whitespace); if (line.len == 0) continue; if (parsers.isComment(line)) continue; considered += 1; try w.writeAll(line); try w.writeByte('\n'); } } /// Whether three compiled files carry the bodies `expected` was taken over. fn compiledBodiesMatch( list_bytes: []const u8, wild_bytes: []const u8, allow_bytes: []const u8, expected: []const u8, ) bool { return std.mem.eql(u8, expected, &bodyChecksum( stripHeader(list_bytes), stripHeader(wild_bytes), stripHeader(allow_bytes), )); } /// A compile that produced no entry at all while rejecting lines is an error /// page, a compressed body or a format the sniff got wrong — not a blocklist. /// Publishing it would replace a working list with nothing and report `ok`. An /// input that rejected nothing is an empty list, which is legal. /// /// A list of nothing but exceptions is loadable: an allow-only list published /// beside a blocking one is a shape operators use, and it produces entries. fn rejectedWithoutEntries(counts: compiler.Counts) bool { if (counts.domains != 0 or counts.wildcards != 0 or counts.exceptions != 0) return false; return counts.invalid != 0 or counts.skipped_unsupported != 0 or counts.long_lines != 0; } /// The digest the `.list`, `.wild` and `.allow` bodies share, in that order, /// each followed by `compiler.body_separator`. /// /// Must stay byte-for-byte what `compiler.compile` produces, separators /// included: this is the other half of the same digest, and the two are /// compared against each other on every refresh. fn bodyChecksum(list_body: []const u8, wild_body: []const u8, allow_body: []const u8) [64]u8 { var hasher = Sha256.init(.{}); hasher.update(list_body); hasher.update(compiler.body_separator); hasher.update(wild_body); hasher.update(compiler.body_separator); hasher.update(allow_body); hasher.update(compiler.body_separator); var digest: [Sha256.digest_length]u8 = undefined; hasher.final(&digest); return std.fmt.bytesToHex(digest, .lower); } fn compiledName(buf: *[name_buf_len]u8, id: i64, suffix: []const u8) []const u8 { // An `i64` prints in at most 20 characters and the longest suffix is ten, // so `name_buf_len` cannot be exceeded. return std.fmt.bufPrint(buf, "{d}{s}", .{ id, suffix }) catch unreachable; } /// Every name `compiledName` can produce, longest suffix first so `.list.tmp` /// is never read as `.list`. const source_file_suffixes = [_][]const u8{ ".allow.tmp", ".list.tmp", ".wild.tmp", ".raw.tmp", ".allow", ".list", ".wild", }; /// The source id a file under the blocklist directory belongs to, or null when /// the name is not one of ours. /// /// The four temporaries count. A refresh that dies between writing one and /// renaming it leaves a file no later refresh reuses and no `defer` reaches, so /// excluding them from the sweep means nothing ever removes them. Matching them /// is safe because `pruneOrphans` holds `refresh_lock` for its whole body: /// every path that creates a temporary runs under that same lock, so no refresh /// is in flight while the sweep reads the directory, and a temporary the sweep /// sees belonging to a source that still has a row is kept regardless. fn sourceFileId(file_name: []const u8) ?i64 { for (source_file_suffixes) |suffix| { if (!std.mem.endsWith(u8, file_name, suffix)) continue; const stem = file_name[0 .. file_name.len - suffix.len]; return std.fmt.parseInt(i64, stem, 10) catch null; } return null; } fn containsId(rows: []const sources_repo.SourceRow, id: i64) bool { for (rows) |row| { if (row.id == id) return true; } return false; } // --------------------------------------------------------------------------- // tests // --------------------------------------------------------------------------- // // Everything here runs against a `:memory:` database. Only the cancellation // cases below reach a file, and they reach a `testing.tmpDir` — real HTTP and // real swaps under load are the integration suite's (S9). const testing = std.testing; const migrations = @import("../storage/migrations.zig"); fn openMigrated() !db.Db { var database = try db.Db.open(":memory:", .{ .mode = .memory }); errdefer database.close(); try db.applyPragmas(&database, .{}); _ = try migrations.migrate(&database); return database; } fn testManager(database: *db.Db, fetcher_ptr: *fetcher.Fetcher) !Manager { return Manager.init( testing.allocator, database, // `acquire` answers before any directory is touched and the header // helpers are pure, so this directory is never opened unless a test // replaces it with one of its own. .{ .dir = std.Io.Dir.cwd() }, fetcher_ptr, .{}, .{ .raw = .fromSeconds(30), .clock = .awake }, ); } test "init leaves the manager with no snapshot and no statuses" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var database = try openMigrated(); defer database.close(); var f: fetcher.Fetcher = undefined; var manager = try testManager(&database, &f); defer manager.deinit(io); try testing.expectEqual(@as(u64, 0), manager.generation); try testing.expectEqual(@as(usize, 0), manager.statuses.len); try testing.expect(manager.current == null); } test "acquire before any reload returns null and holds no lock" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var database = try openMigrated(); defer database.close(); var f: fetcher.Fetcher = undefined; var manager = try testManager(&database, &f); defer manager.deinit(io); try testing.expect(manager.acquire(io) == null); // A retained shared lock would make this exclusive lock block forever. try testing.expect(manager.lock.tryLock(io)); manager.lock.unlock(io); } test "needsRefresh treats a last_updated in the future as due" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var database = try openMigrated(); defer database.close(); var f: fetcher.Fetcher = undefined; var manager = try testManager(&database, &f); defer manager.deinit(io); // `.ok` is the only state that consults the clock at all; every other one // is already due, so the arithmetic below would be unreachable without it. var statuses = [_]SourceStatus{.{ .id = 1, .state = .ok }}; manager.statuses = &statuses; defer manager.statuses = &.{}; const row = testRow(1, true); const stamp = row.last_updated.?; const interval = model.updateIntervalSeconds(manager.update); // The ordinary cases still hold: fresh is not due, stale is. try testing.expect(!manager.needsRefresh(io, row, stamp + 1)); try testing.expect(manager.needsRefresh(io, row, stamp + interval)); // The Pi has no RTC. A fetch stamped while the clock ran ahead of real // time leaves `now - last` negative, which reads as "fetched moments ago" // and suspends every refresh until real time catches the poison stamp — // for a whole day here, and for as long as the clock was wrong in general. try testing.expect(manager.needsRefresh(io, row, stamp - 1)); try testing.expect(manager.needsRefresh(io, row, stamp - 86_400)); } test "the disk gate skips a scheduled refresh only while writes are critical" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var database = try openMigrated(); defer database.close(); var f: fetcher.Fetcher = undefined; var manager = try testManager(&database, &f); defer manager.deinit(io); // No monitor: every pass runs, which is what the tests and `check` rely on. try testing.expect(!manager.refreshGated()); try testing.expectEqual(@as(u64, 0), manager.refreshesGated()); var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null); manager.monitor = &monitor; // `.ok` and `.warn` both allow writes: only `critical` stops them. try testing.expect(!manager.refreshGated()); monitor.state_raw.store(@intFromEnum(disk_monitor.State.warn), .monotonic); try testing.expect(!manager.refreshGated()); try testing.expectEqual(@as(u64, 0), manager.refreshesGated()); monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic); try testing.expect(manager.refreshGated()); try testing.expect(manager.refreshGated()); try testing.expectEqual(@as(u64, 2), manager.refreshesGated()); // Free space recovers and the schedule resumes; the counter keeps its total. monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic); try testing.expect(!manager.refreshGated()); try testing.expectEqual(@as(u64, 2), manager.refreshesGated()); } test "statusSnapshot on an empty manager copies nothing" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var database = try openMigrated(); defer database.close(); var f: fetcher.Fetcher = undefined; var manager = try testManager(&database, &f); defer manager.deinit(io); var out: [4]SourceStatus = undefined; try testing.expectEqual(@as(usize, 0), manager.statusSnapshot(io, &out)); } // --------------------------------------------------------------------------- // a canceled read of a compiled file // --------------------------------------------------------------------------- /// The one file open a `cancelingIo` turns into `error.Canceled`, and the io it /// hands every other open to. /// /// A `std.Io` carries its implementation's `userdata`, so a patched vtable entry /// cannot smuggle a receiver of its own through it and has to read its /// configuration from here. The test runner runs the tests of one binary in /// sequence, so one instance is enough. var canceling_read: struct { inner: std.Io = undefined, /// The file-name suffix whose open is canceled. suffix: []const u8 = "", } = .{}; /// `inner` with the open of every file whose name ends in `suffix` replaced by /// `error.Canceled`. `vtable` is the caller's storage for the patched copy and /// must outlive the returned io. fn cancelingIo(inner: std.Io, suffix: []const u8, vtable: *std.Io.VTable) std.Io { canceling_read = .{ .inner = inner, .suffix = suffix }; vtable.* = inner.vtable.*; vtable.dirOpenFile = cancelingOpenFile; return .{ .userdata = inner.userdata, .vtable = vtable }; } fn cancelingOpenFile( userdata: ?*anyopaque, dir: std.Io.Dir, sub_path: []const u8, options: std.Io.Dir.OpenFileOptions, ) std.Io.File.OpenError!std.Io.File { if (std.mem.endsWith(u8, sub_path, canceling_read.suffix)) return error.Canceled; return canceling_read.inner.vtable.dirOpenFile(userdata, dir, sub_path, options); } test "a canceled compiled-file read cancels the reload instead of recording it" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var database = try openMigrated(); defer database.close(); var f: fetcher.Fetcher = undefined; var mgr = try testManager(&database, &f); defer mgr.deinit(io); mgr.paths = .{ .dir = tmp.dir }; const url = "https://lists.example/hosts.txt"; try sources_repo.insertBlocklistSource(&database, .{ .url = url, .name = "example" }, .{}); var rows = try sources_repo.listSourceRows(&database, testing.allocator); defer rows.deinit(testing.allocator); defer sources_repo.freeSourceRows(testing.allocator, rows.items); const id = rows.items[0].id; const list_body = "aaa.example.com\n"; const wild_body = ""; const allow_body = ""; var dir = try tmp.dir.createDirPathOpen(io, "blocklists", .{}); defer dir.close(io); var list_buf: [name_buf_len]u8 = undefined; var wild_buf: [name_buf_len]u8 = undefined; var allow_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 }); // Present rather than absent, so the third read is a real one: `loadSource` // treats a missing `.allow` as an empty body and would never open it. try dir.writeFile(io, .{ .sub_path = compiledName(&allow_buf, id, ".allow"), .data = allow_body }); try sources_repo.updateSourceStats(&database, id, .{ .last_updated = 1_700_000_000, .domain_count = 1, .wildcard_count = 0, .skipped_regex_count = 0, .skipped_unsupported_count = 0, .exception_count = 0, .checksum = &bodyChecksum(list_body, wild_body, allow_body), }); // The baseline every assertion below is against: one clean reload, one // status entry that says so. try mgr.reload(io); var out: [4]SourceStatus = undefined; try testing.expectEqual(@as(usize, 1), mgr.statusSnapshot(io, &out)); try testing.expectEqual(State.ok, out[0].state); try testing.expect(out[0].loaded); const published = mgr.generation; // Every catch site, in the order `loadSource` reads the three files. A // cancellation is consumed by whoever catches it, so folding it into a load // failure would spend the shutdown signal and leave a status row reading // "Canceled" behind. The `.allow` read is the one that can get this wrong // twice over: it also has to keep `FileNotFound` apart from a cancellation. for ([_][]const u8{ ".list", ".wild", ".allow" }) |suffix| { var vtable: std.Io.VTable = undefined; const canceling = cancelingIo(io, suffix, &vtable); try testing.expectError(error.Canceled, mgr.reload(canceling)); try testing.expectEqual(@as(usize, 1), mgr.statusSnapshot(io, &out)); try testing.expectEqual(State.ok, out[0].state); try testing.expectEqual(@as(usize, 0), out[0].errorText().len); try testing.expect(out[0].loaded); // Nothing was published either: the snapshot the reload never built // cannot have replaced the one still serving. try testing.expectEqual(published, 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(); const io = threaded.io(); var database = try openMigrated(); defer database.close(); var f: fetcher.Fetcher = undefined; var mgr = try testManager(&database, &f); defer mgr.deinit(io); const rows = [_]sources_repo.SourceRow{testRow(1, true)}; try mgr.syncStatuses(io, &rows); // The unknown id: a source inserted through the API after this pass built // its table, or deleted before the pass reached its commit. The outcome // has nowhere to go, and the warning is the only trace it leaves. The log // sink cannot be installed under the test runner — it would eat the // harness's own output — so what is asserted here is that the miss is // survivable and changes nothing. var stranger: SourceStatus = .{ .id = 42 }; stranger.fail(.fetch_failed, "Timeout"); mgr.commitStatus(io, stranger); var out: [4]SourceStatus = undefined; try testing.expectEqual(@as(usize, 1), mgr.statusSnapshot(io, &out)); try testing.expectEqual(@as(i64, 1), out[0].id); try testing.expectEqual(State.never_fetched, out[0].state); // The same commit against an id the table does know still lands. var known: SourceStatus = .{ .id = 1 }; known.fail(.fetch_failed, "Timeout"); mgr.commitStatus(io, known); _ = mgr.statusSnapshot(io, &out); try testing.expectEqual(State.fetch_failed, out[0].state); try testing.expectEqualStrings("Timeout", out[0].errorText()); } test "the header writer produces the documented text" { var buf: [512]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); const header: Header = .{ .url = "https://lists.example/hosts.txt", .format = .hosts, .fetched_at = 1_700_000_000, .counts = .{ .domains = 12, .wildcards = 3, .exceptions = 7, .skipped_regex = 2, .skipped_unsupported = 1, .invalid = 5, .long_lines = 9, .duplicates = 4, }, .checksum = "0" ** 64, }; try header.write(&w); try testing.expectEqualStrings( \\# nxdns blocklist \\# url https://lists.example/hosts.txt \\# format hosts \\# fetched_at 1700000000 \\# domains 12 \\# wildcards 3 \\# exceptions 7 \\# skipped_regex 2 \\# skipped_unsupported 1 \\# invalid 5 \\ ++ "# sha256 " ++ "0" ** 64 ++ "\n", w.buffered()); } test "the log label names a source without printing what its url carries" { // Every `log.warn` in this file formats its subject through `SourceLabel`, // so this is the text of those lines. A `std.log` line is not observable // from a unit test under the default runner; the label is. var buf: [1024]u8 = undefined; const row: sources_repo.SourceRow = .{ .id = 3, .url = "https://lists.example/download/token/hunter2/hosts.txt?apikey=s3cr3t", .name = "ads", .enabled = true, .last_updated = null, .domain_count = 0, .wildcard_count = 0, .skipped_regex_count = 0, .skipped_unsupported_count = 0, .checksum = null, }; const printed = try std.fmt.bufPrint(&buf, "blocklist {f}: download failed: {s}", .{ SourceLabel.of(row), @errorName(error.ConnectFailed), }); try testing.expectEqualStrings( "blocklist source 3 'ads' 'https://lists.example': download failed: ConnectFailed", printed, ); try testing.expect(!std.mem.containsAtLeast(u8, printed, 1, "hunter2")); try testing.expect(!std.mem.containsAtLeast(u8, printed, 1, "s3cr3t")); // The row is database text, and a path that writes it does not have to // validate as strictly as the config validator. Neither column may end the // line and start one of the operator's choosing. var forged = row; forged.name = "ads\n2026-01-01 ERROR forged"; forged.url = "https://lists.example\n2026-01-01 ERROR forged/hosts.txt"; const escaped = try std.fmt.bufPrint(&buf, "blocklist {f}", .{SourceLabel.of(forged)}); try testing.expectEqualStrings( "blocklist source 3 'ads\\n2026-01-01 ERROR forged'" ++ " 'https://lists.example\\n2026-01-01 ERROR forged'", escaped, ); try testing.expect(!std.mem.containsAtLeast(u8, escaped, 1, "\n")); // A name is operator-supplied and reaches the row through the API, so it // can close the quote this label puts around it and open a decoy that reads // as the url of a second source. The quote it would close is escaped, and // the escape is unambiguous because a `\` is escaped too. var decoy = row; decoy.name = "ads' (https://decoy.example) --"; decoy.url = "https://lists.example/hosts.txt"; const quoted = try std.fmt.bufPrint(&buf, "blocklist {f}", .{SourceLabel.of(decoy)}); try testing.expectEqualStrings( "blocklist source 3 'ads\\' (https://decoy.example) --' 'https://lists.example'", quoted, ); } test "stripHeader returns the body of a compiled file" { const file = "# nxdns blocklist\n" ++ "# url https://lists.example/hosts.txt\n" ++ "ads.example.com\ntracker.example.net\n"; try testing.expectEqualStrings("ads.example.com\ntracker.example.net\n", stripHeader(file)); } test "stripHeader returns everything for a file with no header" { try testing.expectEqualStrings("a.example.com\n", stripHeader("a.example.com\n")); } test "stripHeader returns an empty body for a header-only file" { try testing.expectEqualStrings("", stripHeader("# nxdns blocklist\n# sha256 x\n")); } test "stripHeader tolerates an unterminated header line" { try testing.expectEqualStrings("", stripHeader("# nxdns blocklist")); } test "SourceStatus truncates a long error at max_error_len" { var status: SourceStatus = .{ .id = 1 }; const long = "E" ** (max_error_len + 40); status.fail(.fetch_failed, long); try testing.expectEqual(State.fetch_failed, status.state); try testing.expectEqual(@as(u8, max_error_len), status.last_error_len); try testing.expectEqualStrings("E" ** max_error_len, status.errorText()); } test "a success clears the recorded error" { var status: SourceStatus = .{ .id = 7 }; status.fail(.compile_failed, "TooManyDomains"); status.succeed(1_700_000_000, .{ .domains = 3, .wildcards = 1 }); try testing.expectEqual(State.ok, status.state); try testing.expectEqual(@as(i64, 1_700_000_000), status.last_success); try testing.expectEqual(@as(u32, 3), status.counts.domains); try testing.expectEqualStrings("", status.errorText()); } comptime { // The two tests below spell every suffix out instead of looping over // `source_file_suffixes`: a test that reads the table moves with it, so a // name dropped from the table would take the assertion that covers it along. // An eighth suffix breaks the build here until both are extended. std.debug.assert(source_file_suffixes.len == 7); } test "compiledName spells every file name of a source" { var buf: [name_buf_len]u8 = undefined; try testing.expectEqualStrings("42.list", compiledName(&buf, 42, ".list")); try testing.expectEqualStrings("42.wild", compiledName(&buf, 42, ".wild")); try testing.expectEqualStrings("42.allow", compiledName(&buf, 42, ".allow")); try testing.expectEqualStrings("42.raw.tmp", compiledName(&buf, 42, ".raw.tmp")); try testing.expectEqualStrings("42.list.tmp", compiledName(&buf, 42, ".list.tmp")); try testing.expectEqualStrings("42.wild.tmp", compiledName(&buf, 42, ".wild.tmp")); try testing.expectEqualStrings("42.allow.tmp", compiledName(&buf, 42, ".allow.tmp")); } test "sourceFileId matches every name a refresh writes, including the temporaries" { try testing.expectEqual(@as(?i64, 7), sourceFileId("7.list")); try testing.expectEqual(@as(?i64, 7), sourceFileId("7.wild")); try testing.expectEqual(@as(?i64, 7), sourceFileId("7.allow")); // A temporary left by a refresh that died belongs to its source id, so the // sweep can tell whether that source still has a row. try testing.expectEqual(@as(?i64, 7), sourceFileId("7.raw.tmp")); try testing.expectEqual(@as(?i64, 7), sourceFileId("7.list.tmp")); try testing.expectEqual(@as(?i64, 7), sourceFileId("7.wild.tmp")); try testing.expectEqual(@as(?i64, 7), sourceFileId("7.allow.tmp")); try testing.expectEqual(@as(?i64, null), sourceFileId("notes.list")); try testing.expectEqual(@as(?i64, null), sourceFileId("notes.allow")); try testing.expectEqual(@as(?i64, null), sourceFileId("notes.raw.tmp")); try testing.expectEqual(@as(?i64, null), sourceFileId("notes.allow.tmp")); try testing.expectEqual(@as(?i64, null), sourceFileId("7.tmp")); try testing.expectEqual(@as(?i64, null), sourceFileId("7.raw")); try testing.expectEqual(@as(?i64, null), sourceFileId("7.allowed")); try testing.expectEqual(@as(?i64, null), sourceFileId("README")); } test "every name compiledName writes is a name the sweep can attribute" { // A round-trip over the table, not a coverage check: this loop reads the // same array the code reads, so it cannot notice a missing entry. The two // tests above are what pins the set. var buf: [name_buf_len]u8 = undefined; for (source_file_suffixes) |suffix| { try testing.expectEqual(@as(?i64, 42), sourceFileId(compiledName(&buf, 42, suffix))); } } test "a failed refresh keeps the fields of the compiled files still serving" { var status: SourceStatus = .{ .id = 3 }; status.succeed(1_700_000_000, .{ .domains = 5, .wildcards = 2 }); // What `refreshSourceLocked` starts from, and what a download failure does // to it. var next = status; next.last_attempt = 1_700_003_600; next.fail(.fetch_failed, "Timeout"); try testing.expectEqual(State.fetch_failed, next.state); try testing.expectEqualStrings("Timeout", next.errorText()); try testing.expectEqual(@as(i64, 1_700_000_000), next.last_success); try testing.expectEqual(@as(i64, 1_700_003_600), next.last_attempt); try testing.expectEqual(@as(u32, 5), next.counts.domains); try testing.expectEqual(@as(u32, 2), next.counts.wildcards); } test "a load outcome never overwrites a refresh failure" { try testing.expect(State.fetch_failed.isRefreshFailure()); try testing.expect(State.compile_failed.isRefreshFailure()); try testing.expect(State.no_valid_entries.isRefreshFailure()); // The three a load produces. `applyLoadOutcomes` may write over these, // because nothing more informative is there. try testing.expect(!State.ok.isRefreshFailure()); try testing.expect(!State.never_fetched.isRefreshFailure()); try testing.expect(!State.load_failed.isRefreshFailure()); } fn testRow(id: i64, enabled: bool) sources_repo.SourceRow { return .{ .id = id, .url = "https://lists.example/hosts.txt", .name = "example", .enabled = enabled, .last_updated = 1_700_000_000, .domain_count = 9, .wildcard_count = 4, .exception_count = 2, .skipped_regex_count = 1, .skipped_unsupported_count = 5, .checksum = "0" ** 64, }; } test "a candidate table carries prior entries over and leaves the published one alone" { var published = [_]SourceStatus{ .{ .id = 1 }, .{ .id = 2 } }; published[0].setUrl("https://lists.example/one.txt"); published[0].fail(.fetch_failed, "HttpStatus"); published[0].loaded = true; published[1].setUrl("https://lists.example/two.txt"); published[1].succeed(1_700_000_000, .{ .domains = 4 }); // Source 2 was deleted and source 3 added; source 1 kept its id and got a // new url. const rows = [_]sources_repo.SourceRow{ blk: { var row = testRow(1, true); row.url = "https://lists.example/moved.txt"; break :blk row; }, testRow(3, true), }; var candidate: [2]SourceStatus = undefined; mergeStatuses(&candidate, &rows, &published); try testing.expectEqual(@as(i64, 1), candidate[0].id); try testing.expectEqual(State.fetch_failed, candidate[0].state); try testing.expectEqualStrings("HttpStatus", candidate[0].errorText()); try testing.expect(candidate[0].loaded); try testing.expectEqualStrings("https://lists.example/moved.txt", candidate[0].urlText()); try testing.expectEqual(@as(i64, 3), candidate[1].id); try testing.expectEqual(State.never_fetched, candidate[1].state); try testing.expect(!candidate[1].loaded); // The published table is untouched, so a reload that fails before the swap // leaves it describing the snapshot that is still serving — including the // entry of the deleted source, which that snapshot still enforces. try testing.expectEqual(@as(usize, 2), published.len); try testing.expectEqual(@as(i64, 2), published[1].id); try testing.expectEqual(State.ok, published[1].state); try testing.expectEqualStrings("https://lists.example/one.txt", published[0].urlText()); } test "a disabled source stops being loaded" { var statuses = [_]SourceStatus{.{ .id = 1 }}; statuses[0].succeed(1_700_000_000, .{ .domains = 9 }); statuses[0].loaded = true; const rows = [_]sources_repo.SourceRow{testRow(1, false)}; applyLoadOutcomes(&statuses, &rows, &.{.disabled}); // Nothing enforces it any more, and the state that described the files it // used to serve is left as the record of how it last stood. try testing.expect(!statuses[0].loaded); try testing.expectEqual(State.ok, statuses[0].state); } test "a source that failed to refresh keeps its failure while its old files serve" { // What `refreshSourceLocked` records, then what the `reload` that follows // it in `refreshAll` finds: the previous files still load. var statuses = [_]SourceStatus{.{ .id = 1 }}; statuses[0].succeed(1_700_000_000, .{ .domains = 9 }); 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 }}); try testing.expect(statuses[0].loaded); try testing.expectEqual(State.fetch_failed, statuses[0].state); try testing.expectEqualStrings("HttpStatus", statuses[0].errorText()); try testing.expectEqual(@as(u32, 9), statuses[0].counts.domains); } test "a load failure is recorded when no refresh failure explains it" { var statuses = [_]SourceStatus{ .{ .id = 1 }, .{ .id = 2 } }; statuses[0].succeed(1_700_000_000, .{ .domains = 9 }); statuses[0].loaded = true; statuses[1].fail(.compile_failed, "TooManyDomains"); statuses[1].loaded = true; const rows = [_]sources_repo.SourceRow{ testRow(1, true), testRow(2, true) }; const reason: LoadOutcome = .{ .failed = .{ .state = .load_failed, .text = "ChecksumMismatch" } }; applyLoadOutcomes(&statuses, &rows, &.{ reason, reason }); try testing.expect(!statuses[0].loaded); try testing.expectEqual(State.load_failed, statuses[0].state); try testing.expectEqualStrings("ChecksumMismatch", statuses[0].errorText()); // The compile failure is why the files are unusable; it outranks the // symptom the loader saw. try testing.expect(!statuses[1].loaded); try testing.expectEqual(State.compile_failed, statuses[1].state); try testing.expectEqualStrings("TooManyDomains", statuses[1].errorText()); } 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 }}); try testing.expect(statuses[0].loaded); try testing.expectEqual(State.ok, statuses[0].state); try testing.expectEqual(@as(i64, 1_700_000_000), statuses[0].last_success); try testing.expectEqual(@as(u32, 9), statuses[0].counts.domains); try testing.expectEqual(@as(u32, 4), statuses[0].counts.wildcards); try testing.expectEqual(@as(u32, 1), statuses[0].counts.skipped_regex); // Rehydration: a restart reads this from the row and nowhere else, because // no path reparses a compiled file's header. try testing.expectEqual(@as(u32, 5), statuses[0].counts.skipped_unsupported); } test "a status borrows nothing, so a copy outlives the table it came from" { var status: SourceStatus = .{ .id = 5 }; status.setUrl("https://lists.example/hosts.txt"); status.fail(.load_failed, "ChecksumMismatch"); const copy = status; // The source of the original is overwritten, as a reload overwrites the // table: a copy that borrowed would read the new bytes or freed memory. status.setUrl("https://other.example/other.txt"); status.fail(.fetch_failed, "Timeout"); try testing.expectEqualStrings("https://lists.example/hosts.txt", copy.urlText()); try testing.expectEqualStrings("ChecksumMismatch", copy.errorText()); try testing.expectEqual(State.load_failed, copy.state); } test "SourceStatus truncates a long url at max_url_len" { var status: SourceStatus = .{ .id = 6 }; status.setUrl("https://lists.example/" ++ "p" ** max_url_len); try testing.expectEqual(@as(u8, max_url_len), status.url_len); try testing.expectEqualStrings( ("https://lists.example/" ++ "p" ** max_url_len)[0..max_url_len], status.urlText(), ); // A shorter url must not leave the tail of the longer one behind it. status.setUrl("https://a.example/x"); try testing.expectEqualStrings("https://a.example/x", status.urlText()); } test "compiledBodiesMatch verifies the bodies, not the presence of the files" { const list_body = "a.example.com\nb.example.com\n"; const wild_body = "c.example.com\n"; const allow_body = "d.example.com\n"; const expected = bodyChecksum(list_body, wild_body, allow_body); const header = "# nxdns blocklist\n" ++ "# url https://lists.example/hosts.txt\n"; try testing.expect(compiledBodiesMatch( header ++ list_body, header ++ wild_body, header ++ allow_body, &expected, )); // The corruption a reload reports as `ChecksumMismatch`: the file is there, // its body is not what the checksum was taken over. An allow body that lost // its entry counts, because a dropped exception silently restores a block. try testing.expect(!compiledBodiesMatch( header ++ "a.example.com\nb.exa", header ++ wild_body, header ++ allow_body, &expected, )); try testing.expect(!compiledBodiesMatch(header ++ list_body, header ++ wild_body, "", &expected)); try testing.expect(!compiledBodiesMatch("", "", "", &expected)); } test "bodyChecksum separates the three bodies" { const list_body = "a.example.com\nb.example.com\n"; const wild_body = "c.example.com\n"; var hasher = Sha256.init(.{}); hasher.update(list_body); hasher.update(compiler.body_separator); hasher.update(wild_body); hasher.update(compiler.body_separator); hasher.update(compiler.body_separator); var digest: [Sha256.digest_length]u8 = undefined; hasher.final(&digest); const expected = std.fmt.bytesToHex(digest, .lower); try testing.expectEqualStrings(&expected, &bodyChecksum(list_body, wild_body, "")); // No `.allow` file: what `loadSource` and `diskBodiesMatch` pass for one. // It is an empty body, and an empty body still gets its separator. try testing.expect(compiledBodiesMatch(list_body, wild_body, "", &expected)); // The framing itself: the same bytes in a different body is a different // digest. Unframed these two are equal, and a stale `.list` survives an // upstream that switched the name to a wildcard. try testing.expect(!std.mem.eql( u8, &bodyChecksum("a.example\n", "", ""), &bodyChecksum("", "a.example\n", ""), )); } test "rejectedWithoutEntries fails a compile that produced nothing usable" { // An html error page: every line is rejected, nothing is written. try testing.expect(rejectedWithoutEntries(.{ .invalid = 12, .skipped_unsupported = 3 })); // A compressed body: one long binary run with no newline in it. try testing.expect(rejectedWithoutEntries(.{ .long_lines = 1 })); // An empty list rejects nothing and is legal. try testing.expect(!rejectedWithoutEntries(.{})); // A real list rejects lines and still produces entries. try testing.expect(!rejectedWithoutEntries(.{ .domains = 1000, .invalid = 40 })); try testing.expect(!rejectedWithoutEntries(.{ .wildcards = 7, .skipped_unsupported = 90 })); } fn sampleOf(input: []const u8, out: []u8) ![]const u8 { var r: std.Io.Reader = .fixed(input); var w: std.Io.Writer = .fixed(out); try collectSample(&r, &w); return w.buffered(); } test "collectSample skips comments instead of spending the sample on them" { const gpa = testing.allocator; const long_comment = "# " ++ "c" ** (compiler.max_line_len - 2) ++ "\n"; var input: std.ArrayList(u8) = .empty; defer input.deinit(gpa); // Sixteen of these fill a 64 KiB window on their own. for (0..20) |_| try input.appendSlice(gpa, long_comment); try input.appendSlice(gpa, "0.0.0.0 ads.example.com\n0.0.0.0 tracker.example.net\n"); const out = try gpa.alloc(u8, sample_buf_len); defer gpa.free(out); const sample = try sampleOf(input.items, out); try testing.expectEqualStrings( "0.0.0.0 ads.example.com\n0.0.0.0 tracker.example.net\n", sample, ); try testing.expectEqual(parsers.Format.hosts, parsers.detectFormat(sample)); } test "collectSample stops at sample_lines counted lines" { const gpa = testing.allocator; var input: std.ArrayList(u8) = .empty; defer input.deinit(gpa); var line_buf: [64]u8 = undefined; for (0..parsers.sample_lines + 10) |i| { try input.appendSlice(gpa, try std.fmt.bufPrint(&line_buf, "0.0.0.0 host{d}.example.com\n", .{i})); } const out = try gpa.alloc(u8, sample_buf_len); defer gpa.free(out); const sample = try sampleOf(input.items, out); var lines = std.mem.tokenizeScalar(u8, sample, '\n'); var count: usize = 0; while (lines.next()) |_| count += 1; try testing.expectEqual(parsers.sample_lines, count); } test "collectSample keeps the abp marker a long comment run would have hidden" { const gpa = testing.allocator; const long_comment = "! " ++ "c" ** (compiler.max_line_len - 2) ++ "\n"; var input: std.ArrayList(u8) = .empty; defer input.deinit(gpa); for (0..20) |_| try input.appendSlice(gpa, long_comment); try input.appendSlice(gpa, "||ads.example.com^\n"); const out = try gpa.alloc(u8, sample_buf_len); defer gpa.free(out); const sample = try sampleOf(input.items, out); try testing.expectEqualStrings("||ads.example.com^\n", sample); try testing.expectEqual(parsers.Format.abp, parsers.detectFormat(sample)); } test "collectSample skips a line over max_line_len" { const gpa = testing.allocator; var input: std.ArrayList(u8) = .empty; defer input.deinit(gpa); try input.appendNTimes(gpa, 'x', 8 * compiler.max_line_len); try input.appendSlice(gpa, "\nads.example.com\n"); const out = try gpa.alloc(u8, sample_buf_len); defer gpa.free(out); // A `Reader.fixed` holds the whole input, so the over-long line comes back // rather than being refused. The compiler would skip it, so the sniff does. const sample = try sampleOf(input.items, out); try testing.expectEqualStrings("ads.example.com\n", sample); } test "collectSample steps over a line that does not fit the reader buffer" { const gpa = testing.allocator; var input: std.ArrayList(u8) = .empty; defer input.deinit(gpa); try input.appendNTimes(gpa, 'x', 4 * compiler.max_line_len); try input.appendSlice(gpa, "\nads.example.com\n"); // A reader buffer smaller than the long line makes `takeDelimiter` report // `error.StreamTooLong` and leave the stream where it was, which is the // path that loops forever without the discard. var backing: std.Io.Reader = .fixed(input.items); var reader_buf: [compiler.max_line_len]u8 = undefined; var limited = backing.limited(.unlimited, &reader_buf); const out = try gpa.alloc(u8, sample_buf_len); defer gpa.free(out); var w: std.Io.Writer = .fixed(out); try collectSample(&limited.interface, &w); try testing.expectEqualStrings("ads.example.com\n", w.buffered()); } test "bodyChecksum covers the list body, then the wild body, then the allow body" { const all = bodyChecksum("a.example.com\n", "b.example.com\n", "c.example.com\n"); var hasher = Sha256.init(.{}); hasher.update("a.example.com\n"); hasher.update(compiler.body_separator); hasher.update("b.example.com\n"); hasher.update(compiler.body_separator); hasher.update("c.example.com\n"); hasher.update(compiler.body_separator); var digest: [Sha256.digest_length]u8 = undefined; hasher.final(&digest); try testing.expectEqualStrings(&std.fmt.bytesToHex(digest, .lower), &all); // Order matters: the three parts are not interchangeable. try testing.expect(!std.mem.eql( u8, &all, &bodyChecksum("b.example.com\n", "a.example.com\n", "c.example.com\n"), )); try testing.expect(!std.mem.eql( u8, &all, &bodyChecksum("a.example.com\n", "c.example.com\n", "b.example.com\n"), )); }