mokhtar
6a0630c288
Gates / frontend (push) Successful in 2m6s
Gates / test (push) Successful in 2m57s
Gates / test-aarch64 (push) Successful in 8m31s
Gates / package (push) Successful in 4m19s
Gates / container (push) Failing after 2s
CI / gates (push) Failing after 26m21s
648 lines
25 KiB
Zig
648 lines
25 KiB
Zig
//! `GET /api/overview`: everything the Overview page draws, for one period, in
|
||
//! one response.
|
||
//!
|
||
//! It replaces the five per-panel endpoints milestone 30 shipped. Those cost
|
||
//! five scans of every raw row in the window and, being five requests, could
|
||
//! only promise a shared *window* — queries logged between two of them moved
|
||
//! one panel and not the other. One request over one read transaction promises
|
||
//! a shared *snapshot*: the totals, the four breakdowns and the coverage
|
||
//! watermark beside them all describe one database state, so the breakdowns sum
|
||
//! to the totals for a reason and not by luck.
|
||
//!
|
||
//! Buckets are aligned to the UTC grid, not to the moment of the request. Every
|
||
//! width divides a day, so flooring the current time to a multiple of the width
|
||
//! puts each bucket on the same boundary a human reads off a clock, and two
|
||
//! requests a second apart return the same bucket starts. The last bucket is
|
||
//! the one in progress; it fills as the period runs.
|
||
//!
|
||
//! The aggregate runs on the web task's own query-log connection (m7 ruling
|
||
//! 21), which every connection task shares. SQLite's serialized mode makes one
|
||
//! call safe; it does not make a transaction safe, so `WebState.querylog_lock`
|
||
//! covers the whole read and a second BEGIN can never land inside the first.
|
||
//! The transaction is deferred, not `db.Tx`'s BEGIN IMMEDIATE, which would
|
||
//! stall the logger and retention behind an HTTP response.
|
||
//!
|
||
//! The lock is released before the response is written: the body is already
|
||
//! built in the request arena, and holding a database lock across a socket
|
||
//! write would let one slow client serialize every other reader.
|
||
|
||
const std = @import("std");
|
||
|
||
const coverage = @import("../coverage.zig");
|
||
const db = @import("../../storage/db.zig");
|
||
const http_util = @import("../http_util.zig");
|
||
const queries_repo = @import("../../storage/repositories/queries_repo.zig");
|
||
const server = @import("../server.zig");
|
||
|
||
const log = std.log.scoped(.web_overview);
|
||
|
||
/// The four periods ruling 13 defines. The tag names are the wire spellings.
|
||
pub const Period = enum {
|
||
@"1h",
|
||
@"24h",
|
||
@"7d",
|
||
@"30d",
|
||
|
||
pub fn parse(text: []const u8) ?Period {
|
||
return std.meta.stringToEnum(Period, text);
|
||
}
|
||
|
||
/// Ruling 13: 1h→60×1m, 24h→48×30m, 7d→168×1h, 30d→120×6h.
|
||
pub fn bucketSeconds(self: Period) u32 {
|
||
return switch (self) {
|
||
.@"1h" => 60,
|
||
.@"24h" => 30 * 60,
|
||
.@"7d" => 60 * 60,
|
||
.@"30d" => 6 * 60 * 60,
|
||
};
|
||
}
|
||
|
||
pub fn bucketCount(self: Period) u32 {
|
||
return switch (self) {
|
||
.@"1h" => 60,
|
||
.@"24h" => 48,
|
||
.@"7d" => 168,
|
||
.@"30d" => 120,
|
||
};
|
||
}
|
||
|
||
pub fn label(self: Period) []const u8 {
|
||
return @tagName(self);
|
||
}
|
||
};
|
||
|
||
pub const default_period: Period = .@"24h";
|
||
|
||
/// The widest period's bucket count. Nothing here allocates by it any more —
|
||
/// the repository returns arena slices — but it is the bound the response size
|
||
/// argument rests on, and the assertion below is what keeps it true.
|
||
pub const max_buckets = 168;
|
||
|
||
comptime {
|
||
std.debug.assert(std.enums.values(Period).len == server.OverviewCache.slot_count);
|
||
for (std.enums.values(Period)) |period| {
|
||
std.debug.assert(period.bucketCount() <= max_buckets);
|
||
// The UTC alignment argument holds only while every width divides a day.
|
||
std.debug.assert(86_400 % period.bucketSeconds() == 0);
|
||
}
|
||
}
|
||
|
||
pub const Window = struct {
|
||
/// Inclusive, on the bucket grid.
|
||
since: i64,
|
||
/// Exclusive: the end of the bucket that `now` falls in.
|
||
until: i64,
|
||
bucket_seconds: u32,
|
||
bucket_count: u32,
|
||
};
|
||
|
||
pub fn window(period: Period, now_unix: i64) Window {
|
||
const width: i64 = period.bucketSeconds();
|
||
const count: i64 = period.bucketCount();
|
||
const until = @divFloor(now_unix, width) * width + width;
|
||
return .{
|
||
.since = until - width * count,
|
||
.until = until,
|
||
.bucket_seconds = period.bucketSeconds(),
|
||
.bucket_count = period.bucketCount(),
|
||
};
|
||
}
|
||
|
||
pub const Totals = struct {
|
||
queries: u64,
|
||
blocked: u64,
|
||
/// Distinct client addresses in the window.
|
||
clients: u64,
|
||
avg_response_time_us: ?i64,
|
||
};
|
||
|
||
pub const Body = struct {
|
||
period: []const u8,
|
||
since: i64,
|
||
until: i64,
|
||
bucket_seconds: u32,
|
||
totals: Totals,
|
||
buckets: []const queries_repo.Bucket,
|
||
clients: []const queries_repo.ClientSeries,
|
||
other: []const u64,
|
||
types: []const queries_repo.TypeCount,
|
||
routes: []const queries_repo.RouteCount,
|
||
/// Judged against `since`, which is the window this body reports on — so a
|
||
/// dashboard can say "history starts here" instead of charting a pruned
|
||
/// stretch as a quiet one.
|
||
coverage: coverage.Coverage,
|
||
};
|
||
|
||
pub fn handle(
|
||
state: *server.WebState,
|
||
io: std.Io,
|
||
request: *http_util.Request,
|
||
) http_util.HandlerError!void {
|
||
const period = periodParam(request.query) catch return badPeriod(request);
|
||
const database = state.querylog_db orelse return unavailable(request);
|
||
const span = window(period, std.Io.Clock.real.now(io).toSeconds());
|
||
|
||
const body = cachedBody(state, io, database, request.arena, period, span) catch |err| {
|
||
return internal(request, err);
|
||
};
|
||
return http_util.respondBytes(request, .ok, body, http_util.content_type_json, &.{});
|
||
}
|
||
|
||
/// The whole cache decision, start to finish, under one hold of
|
||
/// `querylog_lock`. Returns bytes owned by `arena`, so the caller writes the
|
||
/// socket with the lock already released.
|
||
///
|
||
/// `data_version` is sampled inside the lock and the rebuild is published under
|
||
/// that same sample: a commit landing on another connection while this task
|
||
/// builds moves the pragma, so the entry it installs is keyed to a version the
|
||
/// next request will not ask for and that request rebuilds. Stale bytes under a
|
||
/// current key are therefore not reachable. A failed build or a failed commit
|
||
/// publishes nothing and leaves whatever the slot already held.
|
||
fn cachedBody(
|
||
state: *server.WebState,
|
||
io: std.Io,
|
||
database: *db.Db,
|
||
arena: std.mem.Allocator,
|
||
period: Period,
|
||
span: Window,
|
||
) db.Error![]const u8 {
|
||
state.querylog_lock.lockUncancelable(io);
|
||
defer state.querylog_lock.unlock(io);
|
||
|
||
const index = @intFromEnum(period);
|
||
const data_version = try database.queryInt("PRAGMA data_version");
|
||
|
||
if (state.overview_cache.get(index, span.until, data_version)) |cached| {
|
||
// The copy is what makes a later rebuild's free-and-replace safe: the
|
||
// response is written after the lock is gone, and by then these bytes
|
||
// may belong to nobody.
|
||
return arena.dupe(u8, cached);
|
||
}
|
||
|
||
const body = try buildBody(state, io, database, arena, period, span);
|
||
const owned = try state.gpa.dupe(u8, body);
|
||
state.overview_cache.put(state.gpa, index, span.until, data_version, owned);
|
||
return body;
|
||
}
|
||
|
||
/// One read transaction, one snapshot, one serialized body in `arena`. The
|
||
/// caller holds `querylog_lock` and keeps holding it.
|
||
fn buildBody(
|
||
state: *server.WebState,
|
||
io: std.Io,
|
||
database: *db.Db,
|
||
arena: std.mem.Allocator,
|
||
period: Period,
|
||
span: Window,
|
||
) db.Error![]const u8 {
|
||
var scope = try server.QuerylogRead.openLocked(state, io, database);
|
||
errdefer scope.abort();
|
||
const data = try queries_repo.overview(
|
||
database,
|
||
arena,
|
||
span.since,
|
||
span.bucket_seconds,
|
||
span.bucket_count,
|
||
);
|
||
const window_coverage = try coverage.read(database, span.since);
|
||
try scope.commit();
|
||
|
||
var allocating: std.Io.Writer.Allocating = .init(arena);
|
||
errdefer allocating.deinit();
|
||
std.json.Stringify.value(Body{
|
||
.period = period.label(),
|
||
.since = span.since,
|
||
.until = span.until,
|
||
.bucket_seconds = span.bucket_seconds,
|
||
.totals = .{
|
||
.queries = data.totals.queries,
|
||
.blocked = data.totals.blocked,
|
||
.clients = data.totals.distinct_clients,
|
||
.avg_response_time_us = data.totals.avg_response_time_us,
|
||
},
|
||
.buckets = data.buckets,
|
||
.clients = data.clients.clients,
|
||
.other = data.clients.other,
|
||
.types = data.types,
|
||
.routes = data.routes,
|
||
.coverage = window_coverage,
|
||
}, .{}, &allocating.writer) catch return error.OutOfMemory;
|
||
return allocating.written();
|
||
}
|
||
|
||
pub const PeriodError = error{BadPeriod};
|
||
|
||
/// An absent `period` is the default; anything else it cannot read is a 400,
|
||
/// never a silent fallback — a typo must not return a window nobody asked for.
|
||
fn periodParam(query: []const u8) PeriodError!Period {
|
||
var buf: [8]u8 = undefined;
|
||
const found = http_util.queryValue(query, "period", &buf) catch return error.BadPeriod;
|
||
const text = found orelse return default_period;
|
||
return Period.parse(text) orelse error.BadPeriod;
|
||
}
|
||
|
||
fn badPeriod(request: *http_util.Request) http_util.HandlerError!void {
|
||
return http_util.respondError(request, .bad_request, "period must be one of 1h, 24h, 7d, 30d");
|
||
}
|
||
|
||
fn unavailable(request: *http_util.Request) http_util.HandlerError!void {
|
||
return http_util.respondError(request, .service_unavailable, "query log unavailable");
|
||
}
|
||
|
||
/// The one thing this file logs. A failed aggregate is a fault in the box, not
|
||
/// a property of the request, and the client is told nothing beyond "internal
|
||
/// error" (ruling 8, PLAN §19).
|
||
fn internal(request: *http_util.Request, err: db.Error) http_util.HandlerError!void {
|
||
log.warn("overview failed: {s}", .{@errorName(err)});
|
||
return http_util.respondError(request, .internal_server_error, "internal error");
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// tests
|
||
// ---------------------------------------------------------------------------
|
||
|
||
const querylog_schema = @import("../../storage/querylog_schema.zig");
|
||
const testing = std.testing;
|
||
|
||
test "the period grammar accepts exactly the four spellings" {
|
||
try testing.expectEqual(Period.@"1h", Period.parse("1h").?);
|
||
try testing.expectEqual(Period.@"24h", Period.parse("24h").?);
|
||
try testing.expectEqual(Period.@"7d", Period.parse("7d").?);
|
||
try testing.expectEqual(Period.@"30d", Period.parse("30d").?);
|
||
try testing.expectEqual(@as(?Period, null), Period.parse("12h"));
|
||
try testing.expectEqual(@as(?Period, null), Period.parse("1H"));
|
||
try testing.expectEqual(@as(?Period, null), Period.parse(""));
|
||
}
|
||
|
||
test "an absent period defaults and a bad one is rejected" {
|
||
try testing.expectEqual(default_period, try periodParam(""));
|
||
try testing.expectEqual(default_period, try periodParam("limit=5"));
|
||
try testing.expectEqual(Period.@"7d", try periodParam("period=7d"));
|
||
try testing.expectError(error.BadPeriod, periodParam("period=12h"));
|
||
try testing.expectError(error.BadPeriod, periodParam("period=%2"));
|
||
// Longer than any spelling: rejected rather than truncated to "1h".
|
||
try testing.expectError(error.BadPeriod, periodParam("period=1hhhhhhhhhh"));
|
||
}
|
||
|
||
test "each period spans its own bucket width times its count" {
|
||
for (std.enums.values(Period)) |period| {
|
||
const span = window(period, 1_700_000_000);
|
||
const width: i64 = period.bucketSeconds();
|
||
try testing.expectEqual(width * @as(i64, period.bucketCount()), span.until - span.since);
|
||
}
|
||
}
|
||
|
||
test "the window sits on the UTC grid and ends with the bucket in progress" {
|
||
// 2023-11-14T22:13:20Z, which is not on any bucket boundary.
|
||
const now: i64 = 1_700_000_000;
|
||
const span = window(.@"24h", now);
|
||
|
||
try testing.expectEqual(@as(i64, 0), @rem(span.since, 1800));
|
||
try testing.expectEqual(@as(i64, 0), @rem(span.until, 1800));
|
||
try testing.expect(span.until > now);
|
||
try testing.expect(span.until - now <= 1800);
|
||
try testing.expectEqual(@as(u32, 48), span.bucket_count);
|
||
}
|
||
|
||
test "two requests inside one bucket see the same window" {
|
||
// A bucket boundary, so the offsets below stay inside one minute.
|
||
const boundary: i64 = 1_700_000_000 - @rem(1_700_000_000, 60);
|
||
const first = window(.@"1h", boundary);
|
||
const second = window(.@"1h", boundary + 59);
|
||
try testing.expectEqual(first.since, second.since);
|
||
try testing.expectEqual(first.until, second.until);
|
||
|
||
const next = window(.@"1h", boundary + 60);
|
||
try testing.expectEqual(first.until + 60, next.until);
|
||
}
|
||
|
||
test "a timestamp exactly on a boundary starts a new bucket" {
|
||
const span = window(.@"7d", 1_700_000_000 - 1_700_000_000 % 3600);
|
||
try testing.expectEqual(@as(i64, 0), @rem(span.since, 3600));
|
||
try testing.expectEqual(@as(u32, 168), span.bucket_count);
|
||
}
|
||
|
||
test "every period's window is a window the repository will serve" {
|
||
// The projection path needs the window start on the 30-minute grid and the
|
||
// width a whole number of grains; `window` is the only producer of either.
|
||
for (std.enums.values(Period)) |period| {
|
||
const span = window(period, 1_700_000_123);
|
||
if (span.bucket_seconds < 1800) continue;
|
||
try testing.expectEqual(@as(i64, 0), @mod(span.since, 1800));
|
||
try testing.expectEqual(@as(u32, 0), span.bucket_seconds % 1800);
|
||
}
|
||
}
|
||
|
||
fn openLog() !db.Db {
|
||
var database = try db.Db.open(":memory:", .{ .mode = .memory });
|
||
errdefer database.close();
|
||
try db.applyPragmas(&database, .{});
|
||
try database.exec(querylog_schema.ddl);
|
||
return database;
|
||
}
|
||
|
||
fn writeRow(writer: *queries_repo.BatchWriter, timestamp: i64, blocked: bool, cached: ?bool) !void {
|
||
const rows = [_]queries_repo.Row{.{
|
||
.timestamp = timestamp,
|
||
.domain = "example.com",
|
||
.client_ip = "192.0.2.10",
|
||
.qtype = 1,
|
||
.qclass = 1,
|
||
.rcode = 0,
|
||
.blocked = blocked,
|
||
.response_time_us = 1000,
|
||
.cache_hit = cached,
|
||
.upstream = null,
|
||
.group_id = 1,
|
||
.group_name = "default",
|
||
.policy_action = if (blocked) .block else .allow,
|
||
.policy_reason = if (blocked) .blocklist_domain else .no_match,
|
||
.matched = null,
|
||
.source_id = null,
|
||
.source_name = null,
|
||
.cname_target = null,
|
||
.safe_search_target = null,
|
||
.route_kind = if (blocked) .blocked else .upstream,
|
||
.forward_zone = null,
|
||
}};
|
||
try writer.writeBatch(&rows);
|
||
}
|
||
|
||
test "one overview answers totals and buckets that agree over the same window" {
|
||
var database = try openLog();
|
||
defer database.close();
|
||
|
||
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
|
||
defer arena_state.deinit();
|
||
|
||
const span = window(.@"1h", 1_700_000_000);
|
||
|
||
var writer = try queries_repo.BatchWriter.init(testing.allocator, &database);
|
||
defer writer.deinit();
|
||
// One row in the first bucket, two in the last, one just outside.
|
||
try writeRow(&writer, span.since, false, false);
|
||
try writeRow(&writer, span.until - 1, true, false);
|
||
try writeRow(&writer, span.until - 2, false, true);
|
||
try writeRow(&writer, span.since - 1, false, false);
|
||
|
||
const data = try queries_repo.overview(
|
||
&database,
|
||
arena_state.allocator(),
|
||
span.since,
|
||
span.bucket_seconds,
|
||
span.bucket_count,
|
||
);
|
||
|
||
try testing.expectEqual(@as(u64, 3), data.totals.queries);
|
||
try testing.expectEqual(@as(u64, 1), data.totals.blocked);
|
||
try testing.expectEqual(@as(u64, 1), data.totals.distinct_clients);
|
||
try testing.expectEqual(@as(?i64, 1000), data.totals.avg_response_time_us);
|
||
|
||
try testing.expectEqual(@as(usize, 60), data.buckets.len);
|
||
var summed: u64 = 0;
|
||
var blocked: u64 = 0;
|
||
for (data.buckets) |bucket| {
|
||
summed += bucket.queries;
|
||
blocked += bucket.blocked;
|
||
}
|
||
try testing.expectEqual(data.totals.queries, summed);
|
||
try testing.expectEqual(data.totals.blocked, blocked);
|
||
|
||
try testing.expectEqual(span.since, data.buckets[0].ts);
|
||
try testing.expectEqual(@as(u64, 1), data.buckets[0].queries);
|
||
try testing.expectEqual(@as(u64, 2), data.buckets[59].queries);
|
||
try testing.expectEqual(span.until - span.bucket_seconds, data.buckets[59].ts);
|
||
}
|
||
|
||
test "an empty window reports zeros with a null mean" {
|
||
var database = try openLog();
|
||
defer database.close();
|
||
|
||
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
|
||
defer arena_state.deinit();
|
||
|
||
const span = window(.@"30d", 1_700_000_000);
|
||
const data = try queries_repo.overview(
|
||
&database,
|
||
arena_state.allocator(),
|
||
span.since,
|
||
span.bucket_seconds,
|
||
span.bucket_count,
|
||
);
|
||
|
||
try testing.expectEqual(@as(u64, 0), data.totals.queries);
|
||
try testing.expectEqual(@as(?i64, null), data.totals.avg_response_time_us);
|
||
try testing.expectEqual(@as(usize, 120), data.buckets.len);
|
||
for (data.buckets) |bucket| try testing.expectEqual(@as(u64, 0), bucket.queries);
|
||
// `other` is bucket-count sized even here: a chart must never have to
|
||
// invent the residual series.
|
||
try testing.expectEqual(@as(usize, 120), data.clients.other.len);
|
||
}
|
||
|
||
test "the cache serves one period's bytes and rebuilds when the key moves" {
|
||
const gpa = testing.allocator;
|
||
var cache: server.OverviewCache = .{};
|
||
defer cache.deinit(gpa);
|
||
|
||
try testing.expectEqual(@as(?[]const u8, null), cache.get(0, 100, 7));
|
||
|
||
cache.put(gpa, 0, 100, 7, try gpa.dupe(u8, "first"));
|
||
try testing.expectEqualStrings("first", cache.get(0, 100, 7).?);
|
||
// A different period, a rolled window and a bumped data version are three
|
||
// different keys, and none of them hits.
|
||
try testing.expectEqual(@as(?[]const u8, null), cache.get(1, 100, 7));
|
||
try testing.expectEqual(@as(?[]const u8, null), cache.get(0, 101, 7));
|
||
try testing.expectEqual(@as(?[]const u8, null), cache.get(0, 100, 8));
|
||
|
||
// A rebuild replaces the entry and frees the old body; the leak checker in
|
||
// `testing.allocator` is the assertion.
|
||
cache.put(gpa, 0, 100, 8, try gpa.dupe(u8, "second"));
|
||
try testing.expectEqualStrings("second", cache.get(0, 100, 8).?);
|
||
}
|
||
|
||
/// Two connections onto one file, which is the only arrangement in which
|
||
/// `PRAGMA data_version` moves at all: it reports commits by *other*
|
||
/// connections, so an in-memory database — where there is no other connection —
|
||
/// could never witness the invalidation these tests are about.
|
||
const CacheFixture = struct {
|
||
threaded: std.Io.Threaded,
|
||
tmp: std.testing.TmpDir,
|
||
/// The web task's connection, the one the cache is keyed on.
|
||
reader: db.Db,
|
||
/// Stands in for the logger and for retention.
|
||
writer: db.Db,
|
||
state: server.WebState,
|
||
arena_state: std.heap.ArenaAllocator,
|
||
|
||
fn init(self: *CacheFixture, gpa: std.mem.Allocator) !void {
|
||
self.threaded = .init(gpa, .{});
|
||
errdefer self.threaded.deinit();
|
||
self.tmp = std.testing.tmpDir(.{});
|
||
errdefer self.tmp.cleanup();
|
||
|
||
var path_buf: [256]u8 = undefined;
|
||
const path = try std.fmt.bufPrintZ(
|
||
&path_buf,
|
||
".zig-cache/tmp/{s}/querylog.db",
|
||
.{self.tmp.sub_path},
|
||
);
|
||
|
||
self.writer = try db.Db.open(path, .{ .mode = .read_write_create });
|
||
errdefer self.writer.close();
|
||
try db.applyPragmas(&self.writer, .{});
|
||
try self.writer.exec(querylog_schema.ddl);
|
||
// The DDL stamps `created_at` from the wall clock, and the watermark
|
||
// with it. These tests work over a fixed 2023 window, so a 2026
|
||
// watermark would report every one of them as uncovered and the prune
|
||
// below would not move it.
|
||
try self.writer.exec(
|
||
"UPDATE querylog_meta SET created_at = 1600000000, available_since = 1600000000 WHERE id = 1",
|
||
);
|
||
|
||
self.reader = try db.Db.open(path, .{ .mode = .read_write_existing });
|
||
errdefer self.reader.close();
|
||
try db.applyPragmas(&self.reader, .{});
|
||
|
||
self.state = .{ .gpa = gpa, .querylog_db = &self.reader };
|
||
self.arena_state = .init(gpa);
|
||
}
|
||
|
||
fn deinit(self: *CacheFixture) void {
|
||
self.arena_state.deinit();
|
||
self.state.overview_cache.deinit(self.state.gpa);
|
||
self.reader.close();
|
||
self.writer.close();
|
||
self.tmp.cleanup();
|
||
self.threaded.deinit();
|
||
}
|
||
|
||
fn io(self: *CacheFixture) std.Io {
|
||
return self.threaded.io();
|
||
}
|
||
|
||
fn body(self: *CacheFixture, period: Period, span: Window) db.Error![]const u8 {
|
||
return cachedBody(
|
||
&self.state,
|
||
self.io(),
|
||
&self.reader,
|
||
self.arena_state.allocator(),
|
||
period,
|
||
span,
|
||
);
|
||
}
|
||
|
||
/// One row through the writer connection, which commits and so moves the
|
||
/// reader's `PRAGMA data_version`.
|
||
fn log(self: *CacheFixture, timestamp: i64) !void {
|
||
var batch = try queries_repo.BatchWriter.init(self.state.gpa, &self.writer);
|
||
defer batch.deinit();
|
||
try writeRow(&batch, timestamp, false, false);
|
||
}
|
||
};
|
||
|
||
test "a cache hit answers without opening a read transaction" {
|
||
var fx: CacheFixture = undefined;
|
||
try fx.init(testing.allocator);
|
||
defer fx.deinit();
|
||
|
||
const span = window(.@"1h", 1_700_000_000);
|
||
try fx.log(span.since + 10);
|
||
|
||
const first = try fx.body(.@"1h", span);
|
||
try testing.expect(first.len > 0);
|
||
|
||
// A hit never reaches the database, so a fault armed on the next commit is
|
||
// never spent — and the bytes are the stored ones, not a rebuild's.
|
||
db.read_tx_faults.failNextCommit();
|
||
const second = try fx.body(.@"1h", span);
|
||
try testing.expectEqualStrings(first, second);
|
||
try testing.expect(first.ptr != second.ptr);
|
||
|
||
// Spend the armed fault so it cannot leak into a later test. A miss does
|
||
// reach the database, so this one trips.
|
||
db.read_tx_faults.beginCapture();
|
||
defer _ = db.read_tx_faults.endCapture();
|
||
try testing.expectError(error.Internal, fx.body(.@"24h", window(.@"24h", 1_700_000_000)));
|
||
}
|
||
|
||
test "a commit on another connection invalidates the cached body" {
|
||
var fx: CacheFixture = undefined;
|
||
try fx.init(testing.allocator);
|
||
defer fx.deinit();
|
||
|
||
const span = window(.@"1h", 1_700_000_000);
|
||
try fx.log(span.since + 10);
|
||
const before = try fx.body(.@"1h", span);
|
||
|
||
try fx.log(span.since + 20);
|
||
const after = try fx.body(.@"1h", span);
|
||
try testing.expect(!std.mem.eql(u8, before, after));
|
||
try testing.expect(std.mem.containsAtLeast(u8, after, 1, "\"queries\":2"));
|
||
}
|
||
|
||
test "a retention prune through another connection replaces the body and the watermark" {
|
||
var fx: CacheFixture = undefined;
|
||
try fx.init(testing.allocator);
|
||
defer fx.deinit();
|
||
|
||
const span = window(.@"1h", 1_700_000_000);
|
||
try fx.log(span.since + 10);
|
||
const before = try fx.body(.@"1h", span);
|
||
try testing.expect(std.mem.containsAtLeast(u8, before, 1, "\"queries\":1"));
|
||
|
||
// Past the whole window: the row goes and the watermark advances, and both
|
||
// halves of the response must move together.
|
||
_ = try queries_repo.pruneOlderThan(&fx.writer, span.until);
|
||
const after = try fx.body(.@"1h", span);
|
||
try testing.expect(std.mem.containsAtLeast(u8, after, 1, "\"queries\":0"));
|
||
|
||
var watermark_buf: [64]u8 = undefined;
|
||
const watermark = try std.fmt.bufPrint(
|
||
&watermark_buf,
|
||
"\"available_since\":{d}",
|
||
.{span.until},
|
||
);
|
||
try testing.expect(std.mem.containsAtLeast(u8, after, 1, watermark));
|
||
}
|
||
|
||
test "a window roll rebuilds even with the data unchanged" {
|
||
var fx: CacheFixture = undefined;
|
||
try fx.init(testing.allocator);
|
||
defer fx.deinit();
|
||
|
||
const now: i64 = 1_700_000_000;
|
||
const first = try fx.body(.@"1h", window(.@"1h", now));
|
||
// One bucket later: same data, a different window, and so a different body.
|
||
const rolled = try fx.body(.@"1h", window(.@"1h", now + 60));
|
||
try testing.expect(!std.mem.eql(u8, first, rolled));
|
||
|
||
// The slot now holds the rolled window; asking for the earlier one again
|
||
// rebuilds rather than answering from a key that no longer matches.
|
||
const again = try fx.body(.@"1h", window(.@"1h", now));
|
||
try testing.expectEqualStrings(first, again);
|
||
}
|
||
|
||
test "a failed commit installs nothing and leaves the stored entry alone" {
|
||
var fx: CacheFixture = undefined;
|
||
try fx.init(testing.allocator);
|
||
defer fx.deinit();
|
||
|
||
const span = window(.@"1h", 1_700_000_000);
|
||
try fx.log(span.since + 10);
|
||
const stored = try fx.body(.@"1h", span);
|
||
|
||
// A commit that fails on a rebuild: the key has moved, so this is a miss.
|
||
try fx.log(span.since + 20);
|
||
db.read_tx_faults.failNextCommit();
|
||
db.read_tx_faults.beginCapture();
|
||
try testing.expectError(error.Internal, fx.body(.@"1h", span));
|
||
try testing.expectEqual(@as(usize, 1), db.read_tx_faults.endCapture());
|
||
|
||
// Nothing was published under the new key: the next request rebuilds and
|
||
// sees the second row, rather than being served the failed read's work or
|
||
// the first row's body under a key that now describes two.
|
||
const rebuilt = try fx.body(.@"1h", span);
|
||
try testing.expect(!std.mem.eql(u8, stored, rebuilt));
|
||
try testing.expect(std.mem.containsAtLeast(u8, rebuilt, 1, "\"queries\":2"));
|
||
}
|