fix(dashboard): deduplicate realtime log replay
This commit is contained in:
1 parent
e003b0d8b2
commit
89f7376b1b
4 files changed
+49
-9
No files matched your search
@@ -31,10 +31,19 @@ function boundedAppend(
|
|||||||
entry: ConsoleEntry,
|
entry: ConsoleEntry,
|
||||||
capacity: number,
|
capacity: number,
|
||||||
): readonly ConsoleEntry[] {
|
): readonly ConsoleEntry[] {
|
||||||
if (entries.some(({ id }) => id === entry.id)) return entries;
|
if (entries.some((candidate) => isSameEntry(candidate, entry))) return entries;
|
||||||
return [...entries, entry].slice(-capacity);
|
return [...entries, entry].slice(-capacity);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function isSameEntry(left: ConsoleEntry, right: ConsoleEntry): boolean {
|
||||||
|
return left.id === right.id || (
|
||||||
|
left.occurredAt === right.occurredAt &&
|
||||||
|
left.level === right.level &&
|
||||||
|
left.message === right.message &&
|
||||||
|
searchableData(left.data) === searchableData(right.data)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
export function createLiveConsoleState(capacity = 500): LiveConsoleState {
|
export function createLiveConsoleState(capacity = 500): LiveConsoleState {
|
||||||
if (!Number.isInteger(capacity) || capacity <= 0) {
|
if (!Number.isInteger(capacity) || capacity <= 0) {
|
||||||
throw new RangeError('capacity must be a positive integer');
|
throw new RangeError('capacity must be a positive integer');
|
||||||
@@ -57,8 +66,8 @@ export function reduceLiveConsoleState(
|
|||||||
switch (action.type) {
|
switch (action.type) {
|
||||||
case 'append': {
|
case 'append': {
|
||||||
if (
|
if (
|
||||||
state.entries.some(({ id }) => id === action.entry.id) ||
|
state.entries.some((entry) => isSameEntry(entry, action.entry)) ||
|
||||||
state.pendingEntries.some(({ id }) => id === action.entry.id)
|
state.pendingEntries.some((entry) => isSameEntry(entry, action.entry))
|
||||||
) return state;
|
) return state;
|
||||||
return state.isPaused
|
return state.isPaused
|
||||||
? { ...state, pendingEntries: boundedAppend(state.pendingEntries, action.entry, state.capacity) }
|
? { ...state, pendingEntries: boundedAppend(state.pendingEntries, action.entry, state.capacity) }
|
||||||
|
|||||||
@@ -22,16 +22,20 @@ export class LogBroadcaster {
|
|||||||
}
|
}
|
||||||
|
|
||||||
push(level: LogEntry['level'], message: string, data?: unknown): void {
|
push(level: LogEntry['level'], message: string, data?: unknown): void {
|
||||||
const entry: LogEntry = { timestamp: Date.now(), level, message, data };
|
const realtimeEvent = this.eventHub.closed
|
||||||
|
? undefined
|
||||||
|
: this.eventHub.publish('log.created', { level, message, data });
|
||||||
|
const entry: LogEntry = {
|
||||||
|
timestamp: realtimeEvent ? Date.parse(realtimeEvent.occurredAt) : Date.now(),
|
||||||
|
level,
|
||||||
|
message,
|
||||||
|
data,
|
||||||
|
};
|
||||||
this.entries.push(entry);
|
this.entries.push(entry);
|
||||||
if (this.entries.length > MAX_LOGS) {
|
if (this.entries.length > MAX_LOGS) {
|
||||||
this.entries = this.entries.slice(-MAX_LOGS);
|
this.entries = this.entries.slice(-MAX_LOGS);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!this.eventHub.closed) {
|
|
||||||
this.eventHub.publish('log.created', { level, message, data });
|
|
||||||
}
|
|
||||||
|
|
||||||
this.fanout(JSON.stringify({ type: 'log', ...entry }));
|
this.fanout(JSON.stringify({ type: 'log', ...entry }));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -38,4 +38,19 @@ describe('live console state', () => {
|
|||||||
expect(state.entries.map(({ id }) => id)).toEqual(['3', '4']);
|
expect(state.entries.map(({ id }) => id)).toEqual(['3', '4']);
|
||||||
expect(state.pendingEntries).toEqual([]);
|
expect(state.pendingEntries).toEqual([]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('deduplicates the same log delivered by the initial snapshot and SSE replay', () => {
|
||||||
|
let state = createLiveConsoleState();
|
||||||
|
state = reduceLiveConsoleState(state, {
|
||||||
|
type: 'append',
|
||||||
|
entry: entry('initial-1', 'info', 'Database initialized'),
|
||||||
|
});
|
||||||
|
state = reduceLiveConsoleState(state, {
|
||||||
|
type: 'append',
|
||||||
|
entry: entry('realtime-1', 'info', 'Database initialized'),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(state.entries).toHaveLength(1);
|
||||||
|
expect(state.entries[0]?.id).toBe('initial-1');
|
||||||
|
});
|
||||||
});
|
});
|
||||||
@@ -4,12 +4,21 @@ import { LogBroadcaster } from '../src/web/log-broadcaster.js';
|
|||||||
|
|
||||||
describe('LogBroadcaster realtime bridge', () => {
|
describe('LogBroadcaster realtime bridge', () => {
|
||||||
it('publishes each log through the typed event hub', () => {
|
it('publishes each log through the typed event hub', () => {
|
||||||
const hub = new RealtimeEventHub({ capacity: 10 });
|
const occurredAt = new Date('2026-08-08T15:00:00.000Z');
|
||||||
|
const hub = new RealtimeEventHub({
|
||||||
|
capacity: 10,
|
||||||
|
clock: { now: () => occurredAt },
|
||||||
|
});
|
||||||
const received: RealtimeEvent[] = [];
|
const received: RealtimeEvent[] = [];
|
||||||
|
let recent: unknown;
|
||||||
hub.subscribe((event) => received.push(event));
|
hub.subscribe((event) => received.push(event));
|
||||||
const broadcaster = new LogBroadcaster(hub);
|
const broadcaster = new LogBroadcaster(hub);
|
||||||
|
|
||||||
broadcaster.push('warn', 'Connection is stale', { attempts: 2 });
|
broadcaster.push('warn', 'Connection is stale', { attempts: 2 });
|
||||||
|
broadcaster.recent(
|
||||||
|
{ query: {} } as never,
|
||||||
|
{ json: (value: unknown) => { recent = value; } } as never,
|
||||||
|
);
|
||||||
|
|
||||||
expect(received).toEqual([
|
expect(received).toEqual([
|
||||||
expect.objectContaining({
|
expect.objectContaining({
|
||||||
@@ -21,5 +30,8 @@ describe('LogBroadcaster realtime bridge', () => {
|
|||||||
},
|
},
|
||||||
}),
|
}),
|
||||||
]);
|
]);
|
||||||
|
expect(recent).toEqual([
|
||||||
|
expect.objectContaining({ timestamp: occurredAt.getTime() }),
|
||||||
|
]);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
Reference in new issue
Block a user