feat(logs): coalesce repeated identical lines into one row (×N badge)
A line identical to the last persisted row (level+source+message+path) within a 5-min refreshing window updates that row — context._repeat counts the fold, _firstAt keeps the first occurrence, createdAt tracks the latest so the storm stays at the top of the newest-first viewer. A continuous storm stays ONE row however long it rages, so it can't evict unrelated history via the 50k row cap or grind the appliance disk. LogsViewer badges coalesced rows ×N (tooltip: count + first occurrence, sq/en). In-memory last-row cache only; a pruned-under-us row falls through to a fresh insert. Claude-Session: https://claude.ai/code/session_01Xcm6ikLgGoCxxHrxtjkk5V
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
import { beforeEach, describe, expect, it } from "vitest";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { appLogs, type Db } from "@parking/db";
|
||||
import { createTestDb } from "@parking/db/testing";
|
||||
import { LogService, pinoDbStream } from "./log-service.js";
|
||||
@@ -52,3 +52,65 @@ describe("pinoDbStream level encodings", () => {
|
||||
expect(teed).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
||||
// Storm coalescing: a line identical to the LAST persisted row (level+source+message+
|
||||
// path), arriving within 5 min of its previous occurrence, UPDATES that row (bumping
|
||||
// context._repeat) instead of inserting — one screaming device can't evict unrelated
|
||||
// history. The row's createdAt tracks the LATEST occurrence; the first is preserved in
|
||||
// context._firstAt.
|
||||
describe("storm coalescing", () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("folds a burst of identical error lines into ONE row with a repeat counter", () => {
|
||||
for (let i = 0; i < 200; i++) {
|
||||
stream.write(`{"level":"error","msg":"button-light setAux failed (ctl R3): send ENETUNREACH"}\n`);
|
||||
}
|
||||
const all = rows();
|
||||
expect(all).toHaveLength(1);
|
||||
expect(all[0].context).toMatchObject({ _repeat: 200 });
|
||||
expect(teed).toHaveLength(200); // stdout still gets every line
|
||||
});
|
||||
|
||||
it("keeps first-occurrence time in _firstAt while createdAt tracks the latest", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-07-08T10:00:00.000Z"));
|
||||
stream.write(`{"level":"warn","msg":"same"}\n`);
|
||||
vi.setSystemTime(new Date("2026-07-08T10:02:00.000Z"));
|
||||
stream.write(`{"level":"warn","msg":"same"}\n`);
|
||||
const [row] = rows();
|
||||
expect(row.createdAt).toBe("2026-07-08T10:02:00.000Z");
|
||||
expect(row.context).toMatchObject({ _repeat: 2, _firstAt: "2026-07-08T10:00:00.000Z" });
|
||||
});
|
||||
|
||||
it("a different message (or level) breaks the run — separate rows", () => {
|
||||
stream.write(`{"level":"error","msg":"boom A"}\n`);
|
||||
stream.write(`{"level":"error","msg":"boom A"}\n`);
|
||||
stream.write(`{"level":"error","msg":"boom B"}\n`);
|
||||
stream.write(`{"level":"warn","msg":"boom B"}\n`);
|
||||
expect(rows()).toHaveLength(3);
|
||||
});
|
||||
|
||||
it("an occurrence past the 5-minute window starts a fresh row", () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-07-08T10:00:00.000Z"));
|
||||
stream.write(`{"level":"error","msg":"slow leak"}\n`);
|
||||
vi.setSystemTime(new Date("2026-07-08T10:06:00.000Z"));
|
||||
stream.write(`{"level":"error","msg":"slow leak"}\n`);
|
||||
expect(rows()).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("a CONTINUOUS storm stays one row past the window (each hit refreshes it)", () => {
|
||||
vi.useFakeTimers();
|
||||
let t = new Date("2026-07-08T10:00:00.000Z").getTime();
|
||||
for (let i = 0; i < 10; i++) {
|
||||
vi.setSystemTime(new Date(t));
|
||||
stream.write(`{"level":"error","msg":"storm"}\n`);
|
||||
t += 240_000; // 4 min apart — each inside the window of the PREVIOUS hit
|
||||
}
|
||||
const all = rows();
|
||||
expect(all).toHaveLength(1);
|
||||
expect(all[0].context).toMatchObject({ _repeat: 10 });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -25,6 +25,14 @@ const MAX_MESSAGE = 4_000;
|
||||
const MAX_STACK = 16_000;
|
||||
const MAX_CONTEXT_JSON = 16_000;
|
||||
|
||||
/** Storm coalescing: a line identical to the LAST persisted one (level+source+message+
|
||||
* path) within this window of its previous occurrence UPDATES that row (bumping a
|
||||
* `_repeat` counter in its context) instead of inserting a new one. A continuous storm
|
||||
* keeps refreshing the window, so it stays ONE row however long it rages — repeated
|
||||
* errors can't evict unrelated history or grind the appliance disk (field incident
|
||||
* 2026-07-07: one unreachable controller ≈ hundreds of identical rows/minute). */
|
||||
const COALESCE_WINDOW_MS = 300_000;
|
||||
|
||||
export interface LogRetention {
|
||||
/** Delete logs older than this many days. */
|
||||
readonly maxAgeDays: number;
|
||||
@@ -62,6 +70,16 @@ export class LogService {
|
||||
readonly #retention: LogRetention;
|
||||
/** Reentrancy guard: never let persisting a log itself emit a persisted log. */
|
||||
#writing = false;
|
||||
/** The last persisted row, for storm coalescing (in-memory only; a restart just
|
||||
* starts a fresh row — best-effort, like everything in this sink). */
|
||||
#last: {
|
||||
id: string;
|
||||
key: string;
|
||||
count: number;
|
||||
firstAt: string;
|
||||
lastAtMs: number;
|
||||
baseContext: Record<string, unknown> | null;
|
||||
} | null = null;
|
||||
|
||||
constructor(db: Db, retention: LogRetention = DEFAULT_RETENTION) {
|
||||
this.#db = db;
|
||||
@@ -85,22 +103,53 @@ export class LogService {
|
||||
if (this.#writing) return;
|
||||
this.#writing = true;
|
||||
try {
|
||||
const createdAt = row.createdAt ?? new Date().toISOString();
|
||||
const message = clamp(row.message, MAX_MESSAGE) ?? "";
|
||||
const path = clamp(row.path, 512);
|
||||
const key = `${row.level}|${row.source}|${message}|${path ?? ""}`;
|
||||
const nowMs = Date.now();
|
||||
|
||||
// Storm coalescing: identical to the last persisted row, within the window →
|
||||
// bump that row instead of inserting. createdAt moves to the LATEST occurrence
|
||||
// (keeps the storm visible at the top of the newest-first viewer); the first
|
||||
// occurrence's time is preserved in context._firstAt.
|
||||
const last = this.#last;
|
||||
if (last && last.key === key && nowMs - last.lastAtMs <= COALESCE_WINDOW_MS) {
|
||||
const res = this.#db
|
||||
.update(appLogs)
|
||||
.set({
|
||||
context: { ...(last.baseContext ?? {}), _repeat: last.count + 1, _firstAt: last.firstAt },
|
||||
createdAt,
|
||||
})
|
||||
.where(eq(appLogs.id, last.id))
|
||||
.run();
|
||||
if ((res.changes ?? 0) > 0) {
|
||||
last.count += 1;
|
||||
last.lastAtMs = nowMs;
|
||||
return;
|
||||
}
|
||||
// The row was pruned out from under us — fall through to a fresh insert.
|
||||
}
|
||||
|
||||
const id = randomUUID();
|
||||
const baseContext = safeContext(row.context);
|
||||
this.#db
|
||||
.insert(appLogs)
|
||||
.values({
|
||||
id: randomUUID(),
|
||||
id,
|
||||
level: row.level,
|
||||
source: row.source,
|
||||
message: clamp(row.message, MAX_MESSAGE) ?? "",
|
||||
context: safeContext(row.context),
|
||||
message,
|
||||
context: baseContext,
|
||||
httpStatus: row.httpStatus ?? null,
|
||||
path: clamp(row.path, 512),
|
||||
path,
|
||||
stack: clamp(row.stack, MAX_STACK),
|
||||
userId: row.userId ?? null,
|
||||
userAgent: clamp(row.userAgent, 512),
|
||||
createdAt: row.createdAt ?? new Date().toISOString(),
|
||||
createdAt,
|
||||
})
|
||||
.run();
|
||||
this.#last = { id, key, count: 1, firstAt: createdAt, lastAtMs: nowMs, baseContext };
|
||||
} catch {
|
||||
// Swallow — diagnostics must never take down the path they observe. (Can't log
|
||||
// it; that's the recursion we're guarding against.)
|
||||
|
||||
@@ -25,6 +25,10 @@ function LogRow({ log }: { log: AppLogRecord }) {
|
||||
const { t } = useTranslation();
|
||||
const [open, setOpen] = useState(false);
|
||||
const hasDetail = (log.context && Object.keys(log.context).length > 0) || log.stack;
|
||||
// Storm-coalesced row: the server folds repeated identical lines into one row and
|
||||
// counts them in context._repeat (first occurrence kept in _firstAt).
|
||||
const repeat = typeof log.context?._repeat === "number" ? (log.context?._repeat as number) : null;
|
||||
const firstAt = typeof log.context?._firstAt === "string" ? (log.context?._firstAt as string) : null;
|
||||
|
||||
return (
|
||||
<div className={`border-b border-term-border/50 ${log.level === "error" || log.level === "fatal" ? "bg-term-red/5" : ""}`}>
|
||||
@@ -38,7 +42,20 @@ function LogRow({ log }: { log: AppLogRecord }) {
|
||||
<span className="text-term-muted tabular-nums">{formatRelativeDateTime(log.createdAt, t)}</span>
|
||||
<span className={`font-semibold uppercase ${LEVEL_COLOR[log.level]}`}>{log.level}</span>
|
||||
<span className="text-term-muted">{t(log.source === "frontend" ? "logs.frontend" : "logs.backend")}</span>
|
||||
<span className="truncate text-term-text">{log.message}</span>
|
||||
<span className="truncate text-term-text">
|
||||
{repeat != null && repeat > 1 && (
|
||||
<span
|
||||
className="mr-1.5 rounded-term border border-term-amber/50 px-1 text-[0.625rem] font-semibold text-term-amber"
|
||||
title={t("logs.repeated", {
|
||||
count: repeat,
|
||||
firstAt: firstAt ? formatRelativeDateTime(firstAt, t) : "—",
|
||||
})}
|
||||
>
|
||||
×{repeat}
|
||||
</span>
|
||||
)}
|
||||
{log.message}
|
||||
</span>
|
||||
<span className="text-term-muted tabular-nums">{log.httpStatus ?? ""}</span>
|
||||
</button>
|
||||
{open && hasDetail && (
|
||||
|
||||
@@ -947,6 +947,7 @@ export const en: Catalog = {
|
||||
status: "Status",
|
||||
path: "Path",
|
||||
empty: "No logs.",
|
||||
repeated: "Repeated {{count}} times (first at {{firstAt}})",
|
||||
},
|
||||
backup: {
|
||||
title: "Backup",
|
||||
|
||||
@@ -963,6 +963,7 @@ export const sq = {
|
||||
status: "Statusi",
|
||||
path: "Rruga",
|
||||
empty: "Asnjë regjistër.",
|
||||
repeated: "Përsëritur {{count}} herë (hera e parë {{firstAt}})",
|
||||
},
|
||||
backup: {
|
||||
title: "Kopje rezervë",
|
||||
|
||||
Reference in New Issue
Block a user