Compare commits
6
Commits
cc23c97218
..
v0.0.9
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a8fd9fee48
|
||
|
|
f1a85d3dab
|
||
|
|
72dbcbe24f
|
||
|
|
c5875af8c8
|
||
|
|
409384ee9e
|
||
|
|
9eb78f6171
|
@@ -101,6 +101,11 @@ jobs:
|
|||||||
sudo apt-get update -qq
|
sudo apt-get update -qq
|
||||||
sudo apt-get install -qq -y --no-install-recommends qemu-user
|
sudo apt-get install -qq -y --no-install-recommends qemu-user
|
||||||
|
|
||||||
|
# qemu 11.1.0 has a TCG regression: its translation-block optimization
|
||||||
|
# takes the panic branch of Zig's UBSan pointer-overflow check on a
|
||||||
|
# valid in-bounds pointer (11.0.3 and earlier are clean; bisected via
|
||||||
|
# the Arch archive). This image's qemu 8.2 is not affected — do not
|
||||||
|
# upgrade the emulator past 11.0.x until qemu fixes it.
|
||||||
- name: Run test suite under qemu (plain suite, no -Dintegration)
|
- name: Run test suite under qemu (plain suite, no -Dintegration)
|
||||||
run: zig build test-aarch64 -fqemu
|
run: zig build test-aarch64 -fqemu
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ All notable changes to nxdns are recorded here. The format follows [Keep a Chang
|
|||||||
|
|
||||||
Sections are written by hand. Nothing here is generated from commit messages: the point of the file is to say what changed for an operator, which a commit subject rarely does.
|
Sections are written by hand. Nothing here is generated from commit messages: the point of the file is to say what changed for an operator, which a commit subject rarely does.
|
||||||
|
|
||||||
## [Unreleased]
|
## [0.0.9] - 2026-08-22
|
||||||
|
|
||||||
Query provenance: every logged query becomes exactly explainable — what the policy decided, what matched, where the answer came from and what the client saw. The handler records all of it as the reply goes out, `query_log` stores it, and a detail page reads one query back in the order the pipeline decided it. Read the upgrade note below first: it resets your query history.
|
Query provenance: every logged query becomes exactly explainable — what the policy decided, what matched, where the answer came from and what the client saw. The handler records all of it as the reply goes out, `query_log` stores it, and a detail page reads one query back in the order the pipeline decided it. Read the upgrade note below first: it resets your query history.
|
||||||
|
|
||||||
|
|||||||
@@ -8,7 +8,7 @@
|
|||||||
|
|
||||||
import { act, fireEvent, render, screen, waitFor, within } from "@testing-library/react";
|
import { act, fireEvent, render, screen, waitFor, within } from "@testing-library/react";
|
||||||
import { QueryClientProvider } from "@tanstack/react-query";
|
import { QueryClientProvider } from "@tanstack/react-query";
|
||||||
import { RouterProvider, createMemoryHistory } from "@tanstack/react-router";
|
import { RouterContextProvider, RouterProvider, createMemoryHistory } from "@tanstack/react-router";
|
||||||
import { AuthProvider } from "@/auth/store";
|
import { AuthProvider } from "@/auth/store";
|
||||||
import { createQueryClient } from "@/lib/queryClient";
|
import { createQueryClient } from "@/lib/queryClient";
|
||||||
import { createAppRouter } from "@/routes";
|
import { createAppRouter } from "@/routes";
|
||||||
@@ -16,6 +16,17 @@ import type { Client } from "@/lib/types";
|
|||||||
import { provenance, queryRow } from "@/features/provenance/provenanceFixture";
|
import { provenance, queryRow } from "@/features/provenance/provenanceFixture";
|
||||||
import { health } from "@/lib/healthFixture";
|
import { health } from "@/lib/healthFixture";
|
||||||
import { FakeEventSource } from "./fakeEventSource";
|
import { FakeEventSource } from "./fakeEventSource";
|
||||||
|
import LiveActivity from "./LiveActivity";
|
||||||
|
import type { ActivitySearch } from "./search";
|
||||||
|
|
||||||
|
const LIVE_ORIGIN: ActivitySearch = {
|
||||||
|
mode: "live",
|
||||||
|
since: undefined,
|
||||||
|
until: undefined,
|
||||||
|
domain: undefined,
|
||||||
|
client: undefined,
|
||||||
|
blocked: undefined,
|
||||||
|
};
|
||||||
|
|
||||||
function client(ip: string, name: string, learnedName: string): Client {
|
function client(ip: string, name: string, learnedName: string): Client {
|
||||||
return {
|
return {
|
||||||
@@ -327,23 +338,55 @@ test("opening a second row moves the expanded state and the focus with it", asyn
|
|||||||
expect(document.activeElement).toBe(second);
|
expect(document.activeElement).toBe(second);
|
||||||
});
|
});
|
||||||
|
|
||||||
test("an open streamed detail survives the row being evicted from the ring buffer", async () => {
|
/**
|
||||||
await openLive();
|
* The one render that bypasses the route, because the ring capacity is a
|
||||||
|
* parameter of the component and the route deliberately never passes it:
|
||||||
|
* evicting a row at the real 500 means pushing 500 frames through React state,
|
||||||
|
* which proves nothing the fifth frame does not. The real route tree still
|
||||||
|
* backs the links inside the detail.
|
||||||
|
*/
|
||||||
|
function renderLiveWithCapacity(capacity: number) {
|
||||||
|
const queryClient = createQueryClient();
|
||||||
|
const router = createAppRouter(createMemoryHistory({ initialEntries: ["/activity?mode=live"] }), queryClient);
|
||||||
|
render(
|
||||||
|
<AuthProvider>
|
||||||
|
<QueryClientProvider client={queryClient}>
|
||||||
|
<RouterContextProvider router={router}>
|
||||||
|
<LiveActivity origin={LIVE_ORIGIN} capacity={capacity} />
|
||||||
|
</RouterContextProvider>
|
||||||
|
</QueryClientProvider>
|
||||||
|
</AuthProvider>,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
test("an open streamed detail survives the row being evicted from the ring buffer", () => {
|
||||||
|
renderLiveWithCapacity(5);
|
||||||
|
act(() => sources[0]!.emit("open"));
|
||||||
act(() => sources[0]!.emit("query", frame(1000, "evicted.example")));
|
act(() => sources[0]!.emit("query", frame(1000, "evicted.example")));
|
||||||
fireEvent.click(screen.getByRole("button", { name: "evicted.example" }));
|
fireEvent.click(screen.getByRole("button", { name: "evicted.example" }));
|
||||||
expect(screen.getByRole("heading", { level: 1, name: "evicted.example" })).toBeTruthy();
|
expect(screen.getByRole("heading", { level: 1, name: "evicted.example" })).toBeTruthy();
|
||||||
|
|
||||||
// 500 more queries: the ring keeps the newest 500, so the selected row is
|
// One ringful more: the ring keeps the newest 5, so the selected row is gone
|
||||||
// gone from the table. The detail is a snapshot, not a lookup into the ring.
|
// from the table. The detail is a snapshot, not a lookup into the ring.
|
||||||
act(() => {
|
act(() => {
|
||||||
for (let index = 0; index < 500; index += 1) {
|
for (let index = 0; index < 5; index += 1) {
|
||||||
sources[0]!.emit("query", frame(2000 + index, `filler${index}.example`));
|
sources[0]!.emit("query", frame(2000 + index, `filler${index}.example`));
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
expect(screen.getAllByRole("row")).toHaveLength(6);
|
||||||
expect(screen.queryByRole("button", { name: "evicted.example" })).toBeNull();
|
expect(screen.queryByRole("button", { name: "evicted.example" })).toBeNull();
|
||||||
expect(screen.getByRole("heading", { level: 1, name: "evicted.example" })).toBeTruthy();
|
expect(screen.getByRole("heading", { level: 1, name: "evicted.example" })).toBeTruthy();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("the route renders the live ring at its production capacity", async () => {
|
||||||
|
await openLive();
|
||||||
|
act(() => sources[0]!.emit("query", frame(1000, "pinned.example")));
|
||||||
|
|
||||||
|
expect(screen.getByText(/last 500 kept/)).toBeTruthy();
|
||||||
|
fireEvent.click(screen.getByRole("button", { name: "Freeze" }));
|
||||||
|
expect(screen.getByText(/newest 500 kept/)).toBeTruthy();
|
||||||
|
});
|
||||||
|
|
||||||
test("an open streamed detail survives Freeze and Resume", async () => {
|
test("an open streamed detail survives Freeze and Resume", async () => {
|
||||||
await openLive();
|
await openLive();
|
||||||
act(() => sources[0]!.emit("query", frame(1000, "held.example")));
|
act(() => sources[0]!.emit("query", frame(1000, "held.example")));
|
||||||
|
|||||||
@@ -276,8 +276,19 @@ function LiveDetail({ row, origin, onClose }: { row: StreamedRow; origin: Activi
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
export default function LiveActivity({ origin }: { origin: ActivitySearch }) {
|
/**
|
||||||
const live = useLiveQueries();
|
* `capacity` is the ring size, a parameter only so a test can provoke an
|
||||||
|
* eviction with a handful of rows rather than 500 frames through React state.
|
||||||
|
* The route renders this without it, so the app is always the 500-row ring.
|
||||||
|
*/
|
||||||
|
export default function LiveActivity({
|
||||||
|
origin,
|
||||||
|
capacity = RING_CAPACITY,
|
||||||
|
}: {
|
||||||
|
origin: ActivitySearch;
|
||||||
|
capacity?: number;
|
||||||
|
}) {
|
||||||
|
const live = useLiveQueries({ capacity });
|
||||||
const clientNames = useClientNames();
|
const clientNames = useClientNames();
|
||||||
const [selected, setSelected] = useState<StreamedRow | null>(null);
|
const [selected, setSelected] = useState<StreamedRow | null>(null);
|
||||||
const trigger = useRef<HTMLButtonElement | null>(null);
|
const trigger = useRef<HTMLButtonElement | null>(null);
|
||||||
@@ -312,8 +323,7 @@ export default function LiveActivity({ origin }: { origin: ActivitySearch }) {
|
|||||||
|
|
||||||
{live.frozen && (
|
{live.frozen && (
|
||||||
<p {...stylex.props(styles.note)} role="status">
|
<p {...stylex.props(styles.note)} role="status">
|
||||||
Display frozen — new queries keep buffering ({live.liveCount} in buffer, newest {RING_CAPACITY}{" "}
|
Display frozen — new queries keep buffering ({live.liveCount} in buffer, newest {capacity} kept).
|
||||||
kept).
|
|
||||||
</p>
|
</p>
|
||||||
)}
|
)}
|
||||||
|
|
||||||
@@ -413,7 +423,7 @@ export default function LiveActivity({ origin }: { origin: ActivitySearch }) {
|
|||||||
</div>
|
</div>
|
||||||
<p {...stylex.props(styles.footnote)}>
|
<p {...stylex.props(styles.footnote)}>
|
||||||
Showing {live.rows.length} {live.rows.length === 1 ? "query" : "queries"} (newest first, last{" "}
|
Showing {live.rows.length} {live.rows.length === 1 ? "query" : "queries"} (newest first, last{" "}
|
||||||
{RING_CAPACITY} kept).
|
{capacity} kept).
|
||||||
</p>
|
</p>
|
||||||
</>
|
</>
|
||||||
)}
|
)}
|
||||||
|
|||||||
@@ -24,6 +24,12 @@ export interface LiveQueriesOptions {
|
|||||||
fetchSince?: (since: number) => Promise<QueriesPage>;
|
fetchSince?: (since: number) => Promise<QueriesPage>;
|
||||||
/** Cheap session-gated GET fired once on entering capped, to distinguish an expired session from a real cap. */
|
/** Cheap session-gated GET fired once on entering capped, to distinguish an expired session from a real cap. */
|
||||||
probeSession?: () => Promise<unknown>;
|
probeSession?: () => Promise<unknown>;
|
||||||
|
/**
|
||||||
|
* Ring size. Injectable so a test can provoke an eviction with a handful of
|
||||||
|
* rows instead of pushing 500 frames through React state; the app never
|
||||||
|
* passes it, and the operator never sees it.
|
||||||
|
*/
|
||||||
|
capacity?: number;
|
||||||
}
|
}
|
||||||
|
|
||||||
// A transient drop is invisible to EventSource beyond a bare `error` event;
|
// A transient drop is invisible to EventSource beyond a bare `error` event;
|
||||||
@@ -34,7 +40,10 @@ export interface LiveQueriesOptions {
|
|||||||
export const CAP_ERROR_THRESHOLD = 3;
|
export const CAP_ERROR_THRESHOLD = 3;
|
||||||
|
|
||||||
const defaultEventSource: EventSourceFactory = (url) => new EventSource(url);
|
const defaultEventSource: EventSourceFactory = (url) => new EventSource(url);
|
||||||
const defaultFetchSince = (since: number): Promise<QueriesPage> => api.getQueries({ since, limit: RING_CAPACITY });
|
const defaultFetchSince =
|
||||||
|
(capacity: number) =>
|
||||||
|
(since: number): Promise<QueriesPage> =>
|
||||||
|
api.getQueries({ since, limit: capacity });
|
||||||
const defaultProbeSession = (): Promise<unknown> => api.getPause();
|
const defaultProbeSession = (): Promise<unknown> => api.getPause();
|
||||||
|
|
||||||
function isUnauthorized(error: unknown): boolean {
|
function isUnauthorized(error: unknown): boolean {
|
||||||
@@ -79,7 +88,8 @@ export function useLiveQueries(options?: LiveQueriesOptions): LiveQueries {
|
|||||||
errorsRef.current = 0;
|
errorsRef.current = 0;
|
||||||
setStatus("connecting");
|
setStatus("connecting");
|
||||||
const opts = optionsRef.current;
|
const opts = optionsRef.current;
|
||||||
const fetchSince = opts?.fetchSince ?? defaultFetchSince;
|
const capacity = opts?.capacity ?? RING_CAPACITY;
|
||||||
|
const fetchSince = opts?.fetchSince ?? defaultFetchSince(capacity);
|
||||||
const probeSession = opts?.probeSession ?? defaultProbeSession;
|
const probeSession = opts?.probeSession ?? defaultProbeSession;
|
||||||
const es = (opts?.createEventSource ?? defaultEventSource)(opts?.url ?? api.liveQueriesUrl);
|
const es = (opts?.createEventSource ?? defaultEventSource)(opts?.url ?? api.liveQueriesUrl);
|
||||||
esRef.current = es;
|
esRef.current = es;
|
||||||
@@ -94,7 +104,7 @@ export function useLiveQueries(options?: LiveQueriesOptions): LiveQueries {
|
|||||||
fetchSince(since).then(
|
fetchSince(since).then(
|
||||||
(page) => {
|
(page) => {
|
||||||
if (esRef.current !== es) return;
|
if (esRef.current !== es) return;
|
||||||
const merged = mergeGap(bufferRef.current, page.queries, () => ++keyRef.current);
|
const merged = mergeGap(bufferRef.current, page.queries, () => ++keyRef.current, capacity);
|
||||||
bufferRef.current = merged.rows;
|
bufferRef.current = merged.rows;
|
||||||
setRows(merged.rows);
|
setRows(merged.rows);
|
||||||
setMissed(merged.missed);
|
setMissed(merged.missed);
|
||||||
@@ -122,11 +132,11 @@ export function useLiveQueries(options?: LiveQueriesOptions): LiveQueries {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
lastSeenTsRef.current = payload.request.time;
|
lastSeenTsRef.current = payload.request.time;
|
||||||
bufferRef.current = pushRow(bufferRef.current, {
|
bufferRef.current = pushRow(
|
||||||
kind: "streamed",
|
bufferRef.current,
|
||||||
event: payload,
|
{ kind: "streamed", event: payload, key: ++keyRef.current },
|
||||||
key: ++keyRef.current,
|
capacity,
|
||||||
});
|
);
|
||||||
setRows(bufferRef.current);
|
setRows(bufferRef.current);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
.{
|
.{
|
||||||
.name = .nxdns,
|
.name = .nxdns,
|
||||||
.version = "0.0.8",
|
.version = "0.0.9",
|
||||||
.minimum_zig_version = "0.16.0",
|
.minimum_zig_version = "0.16.0",
|
||||||
.paths = .{""},
|
.paths = .{""},
|
||||||
.fingerprint = 0x3307b311dded1d91,
|
.fingerprint = 0x3307b311dded1d91,
|
||||||
|
|||||||
+55
-2
@@ -1334,6 +1334,13 @@ test "shutdown writes the batch the writer holds and the rest of the queue" {
|
|||||||
&database,
|
&database,
|
||||||
@as(?*disk_monitor.Monitor, null),
|
@as(?*disk_monitor.Monitor, null),
|
||||||
});
|
});
|
||||||
|
// Declared after the `database.close` defer so LIFO stops the writer first:
|
||||||
|
// an early return anywhere below would otherwise close the handle under a
|
||||||
|
// live writer and leave `Threaded.deinit` joining a task nothing ends.
|
||||||
|
defer {
|
||||||
|
logger.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
var names: [250][32]u8 = undefined;
|
var names: [250][32]u8 = undefined;
|
||||||
for (&names, 0..) |*name, i| {
|
for (&names, 0..) |*name, i| {
|
||||||
@@ -1420,6 +1427,12 @@ test "the writer holds an entry for the length of the flush interval" {
|
|||||||
&database,
|
&database,
|
||||||
@as(?*disk_monitor.Monitor, null),
|
@as(?*disk_monitor.Monitor, null),
|
||||||
});
|
});
|
||||||
|
// See the note in "shutdown writes the batch the writer holds": this must
|
||||||
|
// run before the deferred `database.close`.
|
||||||
|
defer {
|
||||||
|
logger.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
logger.log(io, sampleEntry(1, "only.example"));
|
logger.log(io, sampleEntry(1, "only.example"));
|
||||||
|
|
||||||
@@ -1466,6 +1479,12 @@ test "a full batch flushes without waiting for the interval" {
|
|||||||
&database,
|
&database,
|
||||||
@as(?*disk_monitor.Monitor, null),
|
@as(?*disk_monitor.Monitor, null),
|
||||||
});
|
});
|
||||||
|
// See the note in "shutdown writes the batch the writer holds": this must
|
||||||
|
// run before the deferred `database.close`.
|
||||||
|
defer {
|
||||||
|
logger.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
for (0..150) |i| logger.log(io, sampleEntry(@intCast(i), "burst.example"));
|
for (0..150) |i| logger.log(io, sampleEntry(@intCast(i), "burst.example"));
|
||||||
|
|
||||||
@@ -1513,6 +1532,12 @@ test "a gated flush holds the batch until the disk recovers" {
|
|||||||
@as([]const Entry, &entries),
|
@as([]const Entry, &entries),
|
||||||
@as(?*disk_monitor.Monitor, &monitor),
|
@as(?*disk_monitor.Monitor, &monitor),
|
||||||
});
|
});
|
||||||
|
// This task is `flush`, not `runWriter`: no queue shutdown can release it,
|
||||||
|
// so only cancellation ends it on an early return. Declared after the
|
||||||
|
// `writer.deinit`/`database.close` defers so LIFO runs it first.
|
||||||
|
defer {
|
||||||
|
_ = future.cancel(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||||
var waited: usize = 0;
|
var waited: usize = 0;
|
||||||
@@ -1634,6 +1659,12 @@ test "the gating episode opens on the gate, turns losing on a drop, and clears o
|
|||||||
&database,
|
&database,
|
||||||
@as(?*disk_monitor.Monitor, &monitor),
|
@as(?*disk_monitor.Monitor, &monitor),
|
||||||
});
|
});
|
||||||
|
// See the note in "shutdown writes the batch the writer holds": this must
|
||||||
|
// run before the deferred `database.close`.
|
||||||
|
defer {
|
||||||
|
logger.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||||
var waited: usize = 0;
|
var waited: usize = 0;
|
||||||
@@ -1657,11 +1688,14 @@ test "the gating episode opens on the gate, turns losing on a drop, and clears o
|
|||||||
// The disk recovers: the held batch goes out and the episode ends.
|
// The disk recovers: the held batch goes out and the episode ends.
|
||||||
monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
|
monitor.state_raw.store(@intFromEnum(disk_monitor.State.ok), .monotonic);
|
||||||
waited = 0;
|
waited = 0;
|
||||||
while (logger.gateEpisode() != .open) : (waited += 1) {
|
// The write is the completion condition, not the episode: `flush` reopens
|
||||||
|
// the gate before it calls `writeBatch`, so a poll on the episode alone
|
||||||
|
// returns while the row is still in flight and `rows_written` is still 0.
|
||||||
|
while (logger.rows_written.load(.monotonic) == 0) : (waited += 1) {
|
||||||
try testing.expect(waited < 400);
|
try testing.expect(waited < 400);
|
||||||
try poll.sleep(io);
|
try poll.sleep(io);
|
||||||
}
|
}
|
||||||
try testing.expect(logger.rows_written.load(.monotonic) > 0);
|
try testing.expectEqual(GateEpisode.open, logger.gateEpisode());
|
||||||
// The count keeps the history the state does not.
|
// The count keeps the history the state does not.
|
||||||
try testing.expect(logger.queries_dropped.load(.monotonic) > 0);
|
try testing.expect(logger.queries_dropped.load(.monotonic) > 0);
|
||||||
|
|
||||||
@@ -1736,6 +1770,13 @@ test "a gate that closes mid-enqueue still gets the discard that follows it" {
|
|||||||
// The producer enters `enqueue` with the gate open — which is the reading a
|
// The producer enters `enqueue` with the gate open — which is the reading a
|
||||||
// sample taken before the loop would keep for the rest of the call.
|
// sample taken before the loop would keep for the rest of the call.
|
||||||
var producer = try io.concurrent(logOne, .{ &logger, io });
|
var producer = try io.concurrent(logOne, .{ &logger, io });
|
||||||
|
// The producer parks inside `enqueue` and only `release` frees it: an early
|
||||||
|
// return below would otherwise leave `Threaded.deinit` joining it forever.
|
||||||
|
defer {
|
||||||
|
discard_stall.armed = false;
|
||||||
|
discard_stall.release.set(io);
|
||||||
|
producer.await(io);
|
||||||
|
}
|
||||||
discard_stall.parked.waitUncancelable(io);
|
discard_stall.parked.waitUncancelable(io);
|
||||||
|
|
||||||
// The producer is parked with the row it will evict still on the queue, so
|
// The producer is parked with the row it will evict still on the queue, so
|
||||||
@@ -1799,6 +1840,12 @@ test "a canceled writer counts the batch it was holding" {
|
|||||||
&database,
|
&database,
|
||||||
@as(?*disk_monitor.Monitor, &monitor),
|
@as(?*disk_monitor.Monitor, &monitor),
|
||||||
});
|
});
|
||||||
|
// Cancellation is this case's subject, so the guard is the same operation
|
||||||
|
// the body performs; declared after the `database.close` defer so LIFO ends
|
||||||
|
// the writer before the handle goes away.
|
||||||
|
defer {
|
||||||
|
_ = future.cancel(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||||
var waited: usize = 0;
|
var waited: usize = 0;
|
||||||
@@ -1842,6 +1889,12 @@ test "a disk-gated writer drops what it holds at shutdown instead of hanging" {
|
|||||||
&database,
|
&database,
|
||||||
@as(?*disk_monitor.Monitor, &monitor),
|
@as(?*disk_monitor.Monitor, &monitor),
|
||||||
});
|
});
|
||||||
|
// See the note in "shutdown writes the batch the writer holds": this must
|
||||||
|
// run before the deferred `fx.deinit` and `database.close`.
|
||||||
|
defer {
|
||||||
|
logger.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
const poll: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(5), .clock = .awake };
|
||||||
var waited: usize = 0;
|
var waited: usize = 0;
|
||||||
|
|||||||
@@ -211,6 +211,13 @@ test "S8 case 1: the logger writes a real querylog.db end to end" {
|
|||||||
log_db.database(),
|
log_db.database(),
|
||||||
@as(?*disk_monitor.Monitor, null),
|
@as(?*disk_monitor.Monitor, null),
|
||||||
});
|
});
|
||||||
|
// Declared after the `log_db.deinit` defer so LIFO stops the writer first:
|
||||||
|
// an early return below would otherwise close the database under a live
|
||||||
|
// writer and leave the never-closed queue parking it forever.
|
||||||
|
defer {
|
||||||
|
query_log.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
var name_buf: [32]u8 = undefined;
|
var name_buf: [32]u8 = undefined;
|
||||||
for (0..250) |i| {
|
for (0..250) |i| {
|
||||||
@@ -252,6 +259,13 @@ test "S8 case 2: a single entry reaches the file once the flush interval passes"
|
|||||||
log_db.database(),
|
log_db.database(),
|
||||||
@as(?*disk_monitor.Monitor, null),
|
@as(?*disk_monitor.Monitor, null),
|
||||||
});
|
});
|
||||||
|
// Declared after the `log_db.deinit` defer so LIFO stops the writer first:
|
||||||
|
// an early return below would otherwise close the database under a live
|
||||||
|
// writer and leave the never-closed queue parking it forever.
|
||||||
|
defer {
|
||||||
|
query_log.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
query_log.log(io, entryAt(1, "only.example"));
|
query_log.log(io, entryAt(1, "only.example"));
|
||||||
|
|
||||||
@@ -300,6 +314,13 @@ test "S8 case 3: a full queue drops the oldest entries and the newest survive" {
|
|||||||
log_db.database(),
|
log_db.database(),
|
||||||
@as(?*disk_monitor.Monitor, &monitor),
|
@as(?*disk_monitor.Monitor, &monitor),
|
||||||
});
|
});
|
||||||
|
// Declared after the `log_db.deinit` defer so LIFO stops the writer first:
|
||||||
|
// an early return below would otherwise close the database under a live
|
||||||
|
// writer and leave the never-closed queue parking it forever.
|
||||||
|
defer {
|
||||||
|
query_log.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
try awaitCount(&query_log.batches_gated, 1, 200);
|
try awaitCount(&query_log.batches_gated, 1, 200);
|
||||||
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(log_db.database()));
|
try testing.expectEqual(@as(i64, 0), try queries_repo.countRows(log_db.database()));
|
||||||
@@ -341,6 +362,13 @@ test "S8 case 4: the privacy transforms reach the stored rows" {
|
|||||||
log_db.database(),
|
log_db.database(),
|
||||||
@as(?*disk_monitor.Monitor, null),
|
@as(?*disk_monitor.Monitor, null),
|
||||||
});
|
});
|
||||||
|
// Declared after the `log_db.deinit` defer so LIFO stops the writer first:
|
||||||
|
// an early return below would otherwise close the database under a live
|
||||||
|
// writer and leave the never-closed queue parking it forever.
|
||||||
|
defer {
|
||||||
|
query_log.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
var name_buf: [32]u8 = undefined;
|
var name_buf: [32]u8 = undefined;
|
||||||
for (0..20) |i| {
|
for (0..20) |i| {
|
||||||
@@ -433,6 +461,13 @@ test "S8 case 6: a critical disk gates the flushes and recovery releases them" {
|
|||||||
log_db.database(),
|
log_db.database(),
|
||||||
@as(?*disk_monitor.Monitor, &monitor),
|
@as(?*disk_monitor.Monitor, &monitor),
|
||||||
});
|
});
|
||||||
|
// Declared after the `log_db.deinit` defer so LIFO stops the writer first:
|
||||||
|
// an early return below would otherwise close the database under a live
|
||||||
|
// writer and leave the never-closed queue parking it forever.
|
||||||
|
defer {
|
||||||
|
query_log.shutdown(io);
|
||||||
|
future.await(io) catch {};
|
||||||
|
}
|
||||||
|
|
||||||
try awaitCount(&query_log.batches_gated, 1, 200);
|
try awaitCount(&query_log.batches_gated, 1, 200);
|
||||||
try testing.expectEqual(@as(u64, 0), query_log.rows_written.load(.monotonic));
|
try testing.expectEqual(@as(u64, 0), query_log.rows_written.load(.monotonic));
|
||||||
|
|||||||
Reference in New Issue
Block a user