Skip to content

Commit e38cf59

Browse files
badcubanbadcuban
authored
perf(observability): bound log flushing and retries (#185)
1 parent 01edbf5 commit e38cf59

4 files changed

Lines changed: 989 additions & 43 deletions

File tree

‎apps/server/src/provider/Layers/EventNdjsonLogger.test.ts‎

Lines changed: 266 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,14 @@ import path from "node:path";
66
import { ThreadId } from "@threadlines/contracts";
77
import { assert, describe, it } from "@effect/vitest";
88
import * as Effect from "effect/Effect";
9+
import * as TestClock from "effect/testing/TestClock";
10+
import { vi } from "vitest";
911

10-
import { cleanupProviderEventLogDirectory, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
12+
import {
13+
cleanupProviderEventLogDirectory,
14+
type EventNdjsonLogger,
15+
makeEventNdjsonLogger,
16+
} from "./EventNdjsonLogger.ts";
1117

1218
function parseLogLine(line: string) {
1319
const match = /^\[([^\]]+)\] ([A-Z]+): (.+)$/.exec(line);
@@ -147,6 +153,264 @@ describe("EventNdjsonLogger", () => {
147153
}),
148154
);
149155

156+
it.effect("flushes after the batch window and re-arms for later writes", () =>
157+
Effect.gen(function* () {
158+
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
159+
const basePath = path.join(tempDir, "provider-canonical.ndjson");
160+
const globalPath = path.join(tempDir, "_global.log");
161+
let logger: EventNdjsonLogger | undefined;
162+
163+
try {
164+
logger = yield* makeEventNdjsonLogger(basePath, {
165+
stream: "canonical",
166+
batchWindowMs: 1_000,
167+
});
168+
assert.notEqual(logger, undefined);
169+
if (!logger) {
170+
return;
171+
}
172+
173+
yield* logger.write({ id: "evt-first" }, null);
174+
assert.equal(fs.existsSync(globalPath), false);
175+
yield* TestClock.adjust("999 millis");
176+
yield* Effect.yieldNow;
177+
assert.equal(fs.existsSync(globalPath), false);
178+
yield* TestClock.adjust("1 millis");
179+
yield* Effect.yieldNow;
180+
assert.equal(fs.existsSync(globalPath), true);
181+
182+
yield* logger.write({ id: "evt-second" }, null);
183+
assert.equal(fs.readFileSync(globalPath, "utf8").includes("evt-second"), false);
184+
yield* TestClock.adjust("1 second");
185+
yield* Effect.yieldNow;
186+
187+
const lines = fs
188+
.readFileSync(globalPath, "utf8")
189+
.trim()
190+
.split("\n")
191+
.map((line) => parseLogLine(line));
192+
assert.deepEqual(
193+
lines.map((line) => line.payload),
194+
['{"id":"evt-first"}', '{"id":"evt-second"}'],
195+
);
196+
} finally {
197+
if (logger) {
198+
yield* logger.close();
199+
}
200+
fs.rmSync(tempDir, { recursive: true, force: true });
201+
}
202+
}),
203+
);
204+
205+
it.effect("flushes immediately when buffered bytes reach the configured cap", () =>
206+
Effect.gen(function* () {
207+
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
208+
const basePath = path.join(tempDir, "provider-native.ndjson");
209+
const globalPath = path.join(tempDir, "_global.log");
210+
let logger: EventNdjsonLogger | undefined;
211+
212+
try {
213+
logger = yield* makeEventNdjsonLogger(basePath, {
214+
stream: "native",
215+
batchWindowMs: 10_000,
216+
maxBufferedBytes: 350,
217+
});
218+
assert.notEqual(logger, undefined);
219+
if (!logger) {
220+
return;
221+
}
222+
223+
yield* logger.write({ id: "evt-byte-threshold", payload: "\u{1F642}".repeat(100) }, null);
224+
assert.equal(fs.existsSync(globalPath), true);
225+
assert.equal(fs.readFileSync(globalPath, "utf8").includes("evt-byte-threshold"), true);
226+
} finally {
227+
if (logger) {
228+
yield* logger.close();
229+
}
230+
fs.rmSync(tempDir, { recursive: true, force: true });
231+
}
232+
}),
233+
);
234+
235+
it.effect("drops oversized records before the rotating sink can append them", () =>
236+
Effect.gen(function* () {
237+
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
238+
const basePath = path.join(tempDir, "provider-canonical.ndjson");
239+
const globalPath = path.join(tempDir, "_global.log");
240+
const blockingBackup = `${globalPath}.2`;
241+
let logger: EventNdjsonLogger | undefined;
242+
243+
try {
244+
yield* TestClock.setTime(0);
245+
logger = yield* makeEventNdjsonLogger(basePath, {
246+
stream: "canonical",
247+
maxBytes: 180,
248+
maxFiles: 2,
249+
batchWindowMs: 0,
250+
});
251+
assert.notEqual(logger, undefined);
252+
if (!logger) {
253+
return;
254+
}
255+
256+
fs.mkdirSync(path.join(blockingBackup, "child"), { recursive: true });
257+
yield* logger.write({ id: "evt-oversized", payload: "x".repeat(300) }, null);
258+
fs.rmSync(blockingBackup, { recursive: true, force: true });
259+
yield* TestClock.adjust("200 millis");
260+
yield* Effect.yieldNow;
261+
yield* logger.write({ id: "evt-small" }, null);
262+
263+
const payloadIds = fs
264+
.readdirSync(tempDir)
265+
.filter((entry) => entry === "_global.log" || entry.startsWith("_global.log."))
266+
.flatMap((entry) =>
267+
fs
268+
.readFileSync(path.join(tempDir, entry), "utf8")
269+
.trim()
270+
.split("\n")
271+
.map((line) => JSON.parse(parseLogLine(line).payload).id as string),
272+
);
273+
assert.deepEqual(payloadIds, ["evt-small"]);
274+
} finally {
275+
if (logger) {
276+
yield* logger.close();
277+
}
278+
fs.rmSync(tempDir, { recursive: true, force: true });
279+
}
280+
}),
281+
);
282+
283+
it.effect("waits for the retry timer and does not duplicate completed chunks", () =>
284+
Effect.gen(function* () {
285+
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
286+
const basePath = path.join(tempDir, "provider-canonical.ndjson");
287+
const realAppendFileSync = fs.appendFileSync.bind(fs);
288+
let appendAttempts = 0;
289+
let failWritesAfterFirst = true;
290+
let logger: EventNdjsonLogger | undefined;
291+
const appendSpy = vi.spyOn(fs, "appendFileSync").mockImplementation((...args) => {
292+
appendAttempts += 1;
293+
if (failWritesAfterFirst && appendAttempts > 1) {
294+
throw new Error("simulated provider log write failure");
295+
}
296+
Reflect.apply(realAppendFileSync, fs, args);
297+
});
298+
299+
try {
300+
yield* TestClock.setTime(0);
301+
logger = yield* makeEventNdjsonLogger(basePath, {
302+
stream: "canonical",
303+
maxBytes: 180,
304+
maxFiles: 10,
305+
batchWindowMs: 1_000,
306+
maxBufferedBytes: 2_000,
307+
});
308+
assert.notEqual(logger, undefined);
309+
if (!logger) {
310+
return;
311+
}
312+
313+
yield* logger.write({ id: "evt-1", payload: "x".repeat(60) }, null);
314+
yield* logger.write({ id: "evt-2", payload: "x".repeat(60) }, null);
315+
yield* logger.write({ id: "evt-3", payload: "x".repeat(60) }, null);
316+
yield* TestClock.adjust("1 second");
317+
yield* Effect.yieldNow;
318+
assert.equal(appendAttempts, 2);
319+
320+
yield* logger.write({ id: "evt-4", payload: "x".repeat(60) }, null);
321+
assert.equal(appendAttempts, 2);
322+
yield* TestClock.adjust("999 millis");
323+
yield* Effect.yieldNow;
324+
assert.equal(appendAttempts, 2);
325+
326+
failWritesAfterFirst = false;
327+
yield* TestClock.adjust("1 millis");
328+
yield* Effect.yieldNow;
329+
yield* TestClock.adjust("1 second");
330+
yield* Effect.yieldNow;
331+
const payloadIds = fs
332+
.readdirSync(tempDir)
333+
.filter((entry) => entry === "_global.log" || entry.startsWith("_global.log."))
334+
.flatMap((entry) =>
335+
fs
336+
.readFileSync(path.join(tempDir, entry), "utf8")
337+
.trim()
338+
.split("\n")
339+
.map((line) => JSON.parse(parseLogLine(line).payload).id as string),
340+
)
341+
.toSorted();
342+
assert.deepEqual(payloadIds, ["evt-1", "evt-2", "evt-3", "evt-4"]);
343+
} finally {
344+
if (logger) {
345+
yield* logger.close();
346+
}
347+
appendSpy.mockRestore();
348+
fs.rmSync(tempDir, { recursive: true, force: true });
349+
}
350+
}),
351+
);
352+
353+
it.effect("bounds buffered records globally across thread writers", () =>
354+
Effect.gen(function* () {
355+
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
356+
const basePath = path.join(tempDir, "provider-canonical.ndjson");
357+
const realAppendFileSync = fs.appendFileSync.bind(fs);
358+
let failWrites = true;
359+
let logger: EventNdjsonLogger | undefined;
360+
const appendSpy = vi.spyOn(fs, "appendFileSync").mockImplementation((...args) => {
361+
if (failWrites) {
362+
throw new Error("simulated provider log write failure");
363+
}
364+
Reflect.apply(realAppendFileSync, fs, args);
365+
});
366+
367+
try {
368+
yield* TestClock.setTime(0);
369+
logger = yield* makeEventNdjsonLogger(basePath, {
370+
stream: "canonical",
371+
batchWindowMs: 1_000,
372+
maxBufferedBytes: 500,
373+
});
374+
assert.notEqual(logger, undefined);
375+
if (!logger) {
376+
return;
377+
}
378+
379+
for (let index = 0; index < 20; index += 1) {
380+
yield* logger.write(
381+
{ id: `evt-${index}`, payload: "x".repeat(40) },
382+
ThreadId.make(`thread-${index}`),
383+
);
384+
}
385+
yield* TestClock.adjust("1 second");
386+
yield* Effect.yieldNow;
387+
388+
failWrites = false;
389+
yield* TestClock.adjust("1 second");
390+
yield* Effect.yieldNow;
391+
const logFiles = fs.readdirSync(tempDir).filter((entry) => entry.endsWith(".log"));
392+
const persistedBytes = logFiles.reduce(
393+
(total, entry) => total + fs.statSync(path.join(tempDir, entry)).size,
394+
0,
395+
);
396+
const persistedRecords = logFiles.reduce(
397+
(total, entry) =>
398+
total + fs.readFileSync(path.join(tempDir, entry), "utf8").trim().split("\n").length,
399+
0,
400+
);
401+
assert.equal(persistedBytes <= 500, true);
402+
assert.equal(persistedRecords > 0, true);
403+
assert.equal(persistedRecords < 20, true);
404+
} finally {
405+
if (logger) {
406+
yield* logger.close();
407+
}
408+
appendSpy.mockRestore();
409+
fs.rmSync(tempDir, { recursive: true, force: true });
410+
}
411+
}),
412+
);
413+
150414
it.effect(
151415
"falls back to a global segment when orchestration thread id is missing or invalid",
152416
() =>
@@ -236,7 +500,7 @@ describe("EventNdjsonLogger", () => {
236500
try {
237501
const logger = yield* makeEventNdjsonLogger(basePath, {
238502
stream: "native",
239-
maxBytes: 120,
503+
maxBytes: 200,
240504
maxFiles: 2,
241505
});
242506
assert.notEqual(logger, undefined);

0 commit comments

Comments
 (0)