Skip to content

Commit 8a24757

Browse files
committed
fix(logging): adopt a crashed aside and retry a busy flush
An empty flush puts a leftover rotating aside back on .1 before a later rotate can replace that path. A live holder keeps the lines queued for another flush, and that wait still does not rotate.
1 parent 2fd6138 commit 8a24757

2 files changed

Lines changed: 131 additions & 36 deletions

File tree

‎src/logging/sink.test.ts‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -255,6 +255,56 @@ describe("installFileLogSink", () => {
255255
}
256256
});
257257

258+
test("an aside left after a crash is restored even when the buffer is empty", () => {
259+
const { dir, file } = tempLogFile();
260+
try {
261+
installFileLogSink(file, { maxBytes: 10_000 });
262+
const aside = `${file}.${process.pid}.1.rotating.aside`;
263+
const marker = "aside-generation\n";
264+
fs.writeFileSync(aside, marker);
265+
flushFileLogSinkSync();
266+
const active = fs.existsSync(file) ? fs.readFileSync(file, "utf8") : "";
267+
const rotated = fs.existsSync(`${file}.1`)
268+
? fs.readFileSync(`${file}.1`, "utf8")
269+
: "";
270+
expect(`${active}${rotated}`).toContain("aside-generation");
271+
expect(fs.existsSync(aside)).toBe(false);
272+
} finally {
273+
fs.rmSync(dir, { recursive: true, force: true });
274+
}
275+
});
276+
277+
test("a busy lock retries the flush without rotating first", async () => {
278+
const { dir, file } = tempLogFile();
279+
const holder = liveLockHolder();
280+
try {
281+
installFileLogSink(file, { maxBytes: 300 });
282+
const first = getLogger(["logging", "retry-a"]);
283+
first.info("first-marker");
284+
first.info("first-marker-b");
285+
first.info("short-active");
286+
flushFileLogSinkSync();
287+
const rotated = `${file}.1`;
288+
const marker = fs.readFileSync(rotated, "utf8");
289+
const activeBefore = fs.readFileSync(file, "utf8");
290+
fs.writeFileSync(`${file}.lock`, `${holder.pid}:held`);
291+
getLogger(["logging", "retry-b"]).info("retry-line");
292+
await Promise.resolve();
293+
expect(fs.readFileSync(rotated, "utf8")).toBe(marker);
294+
expect(fs.readFileSync(file, "utf8")).toBe(activeBefore);
295+
fs.rmSync(`${file}.lock`);
296+
await Promise.resolve();
297+
const active = fs.readFileSync(file, "utf8");
298+
const kept = fs.existsSync(rotated)
299+
? fs.readFileSync(rotated, "utf8")
300+
: "";
301+
expect(`${active}${kept}`).toContain("retry-line");
302+
} finally {
303+
holder.kill();
304+
fs.rmSync(dir, { recursive: true, force: true });
305+
}
306+
});
307+
258308
test("a second reclaimer does not delete the lock a winner just created", () => {
259309
const { dir, file } = tempLogFile();
260310
const holder = liveLockHolder();

‎src/logging/sink.ts‎

Lines changed: 81 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -105,27 +105,53 @@ function leftoverTarget(current: FileSinkState): string | null {
105105
}
106106

107107
/**
108-
* A crash in the old rotate window leaves the only copy as
109-
* `corbits.log.<pid>.<n>.rotating`, with the active name and `.1` gone.
110-
* Put that file back under the lock before this flush decides to rotate.
108+
* A crash leaves the only copies as `*.rotating` (the active bytes) and
109+
* `*.rotating.aside` (the previous generation). Put both back before this
110+
* flush rotates, even when there is nothing queued. A finished publish that
111+
* died before deleting the aside can drop that aside once `.1` is present,
112+
* so a restarted counter cannot rename over it.
111113
*/
112114
function adoptLeftoverRotation(current: FileSinkState): void {
115+
const dir = dirname(current.path);
113116
let names: string[];
114117
try {
115-
names = fs.readdirSync(dirname(current.path));
118+
names = fs.readdirSync(dir);
116119
} catch {
117120
return;
118121
}
119122
const base = basename(current.path);
120-
const pattern = new RegExp(
123+
const rotatingRe = new RegExp(
121124
`^${escapeRegExp(base)}\\.\\d+\\.\\d+\\.rotating$`,
122125
);
123-
const leftovers = names.filter((name) => pattern.test(name)).sort();
124-
for (const name of leftovers) {
126+
const asideRe = new RegExp(
127+
`^${escapeRegExp(base)}\\.\\d+\\.\\d+\\.rotating\\.aside$`,
128+
);
129+
const rotatings = names.filter((name) => rotatingRe.test(name)).sort();
130+
for (const name of names.filter((entry) => asideRe.test(entry)).sort()) {
131+
const from = join(dir, name);
132+
const rotated = `${current.path}.1`;
133+
if (!fs.existsSync(rotated)) {
134+
try {
135+
fs.renameSync(from, rotated);
136+
} catch {
137+
// Leave the aside. Never delete the only copy.
138+
}
139+
continue;
140+
}
141+
const sibling = name.slice(0, -".aside".length);
142+
if (!rotatings.includes(sibling) && !fs.existsSync(join(dir, sibling))) {
143+
try {
144+
fs.rmSync(from);
145+
} catch {
146+
// Leave it so a later rotate cannot replace this path.
147+
}
148+
}
149+
}
150+
for (const name of rotatings) {
125151
const target = leftoverTarget(current);
126152
if (target === null) return;
127153
try {
128-
fs.renameSync(join(dirname(current.path), name), target);
154+
fs.renameSync(join(dir, name), target);
129155
} catch {
130156
// Leave the leftover. Never delete the only copy.
131157
}
@@ -147,10 +173,10 @@ function restoreRotated(temp: string, active: string): void {
147173
}
148174

149175
function rotateActive(current: FileSinkState): boolean {
150-
// Windows throws EPERM when the destination exists, so never rename onto
151-
// `.1`. Move `.1` aside, publish the temp onto that name, then delete the
152-
// aside. A kill after the aside rename still leaves both copies on disk.
153-
// A thrown error puts the temp back and does not truncate the active file.
176+
// Publish the active bytes under `*.rotating` first, which the next flush
177+
// adopts. Only then move `.1` to `*.rotating.aside` (also adopted). Never
178+
// rename onto an existing aside: that would replace a generation a crash
179+
// left behind. A thrown error puts the temp back and does not truncate.
154180
const temp = `${current.path}.${String(process.pid)}.${String((rotateCounter += 1))}.rotating`;
155181
const rotated = `${current.path}.1`;
156182
const aside = `${temp}.aside`;
@@ -161,6 +187,10 @@ function rotateActive(current: FileSinkState): boolean {
161187
}
162188
let movedAside = false;
163189
try {
190+
if (fs.existsSync(aside)) {
191+
restoreRotated(temp, current.path);
192+
return false;
193+
}
164194
if (fs.existsSync(rotated)) {
165195
const info = fs.lstatSync(rotated);
166196
if (!info.isFile()) {
@@ -186,7 +216,7 @@ function rotateActive(current: FileSinkState): boolean {
186216
try {
187217
fs.rmSync(aside);
188218
} catch {
189-
// The published `.1` is the copy that matters. Leave the aside.
219+
// `.1` already holds the new generation. Leave the aside.
190220
}
191221
}
192222
current.size = 0;
@@ -336,25 +366,30 @@ function reclaimDeadLock(lockPath: string, deadClaim: string): void {
336366
}
337367
}
338368

339-
function acquireSinkLock(
340-
lockPath: string,
341-
): { fd: number; claim: string } | null {
369+
type SinkLock =
370+
| { readonly acquired: true; readonly fd: number; readonly claim: string }
371+
| { readonly acquired: false; readonly retry: boolean };
372+
373+
function acquireSinkLock(lockPath: string): SinkLock {
342374
const deadline = Date.now() + LOCK_WAIT_MS;
343375
const claim = `${String(process.pid)}:${String((lockClaimCounter += 1))}`;
344376
for (;;) {
345377
try {
346378
const fd = fs.openSync(lockPath, "wx", 0o600);
347379
fs.writeSync(fd, claim);
348-
return { fd, claim };
380+
return { acquired: true, fd, claim };
349381
} catch (err) {
350-
if (!isErrno(err, "EEXIST")) return null;
382+
if (!isErrno(err, "EEXIST")) return { acquired: false, retry: false };
351383
const read = readLockClaim(lockPath);
352-
if (read.status === "missing") continue;
353-
if (read.status === "ok" && isDeadClaim(lockPath, read.claim)) {
354-
reclaimDeadLock(lockPath, read.claim);
384+
if (
385+
read.status === "missing" ||
386+
(read.status === "ok" && isDeadClaim(lockPath, read.claim))
387+
) {
388+
if (read.status === "ok") reclaimDeadLock(lockPath, read.claim);
389+
if (Date.now() >= deadline) return { acquired: false, retry: true };
355390
continue;
356391
}
357-
if (Date.now() >= deadline) return null;
392+
if (Date.now() >= deadline) return { acquired: false, retry: true };
358393
sleepSync(LOCK_POLL_MS);
359394
}
360395
}
@@ -379,23 +414,36 @@ function releaseSinkLock(
379414
}
380415
}
381416

382-
function withSinkLock(lockPath: string, body: () => void): boolean {
417+
function withSinkLock(
418+
lockPath: string,
419+
body: () => void,
420+
): "ok" | "busy" | "error" {
383421
if (sinkLockDepth > 0) {
384422
body();
385-
return true;
423+
return "ok";
386424
}
387425
const held = acquireSinkLock(lockPath);
388-
if (held === null) return false;
426+
if (!held.acquired) return held.retry ? "busy" : "error";
389427
sinkLockDepth += 1;
390428
try {
391429
body();
392-
return true;
430+
return "ok";
393431
} finally {
394432
sinkLockDepth -= 1;
395433
releaseSinkLock(lockPath, held);
396434
}
397435
}
398436

437+
function scheduleDrain(current: FileSinkState): void {
438+
if (current.drainScheduled || current.buffer.length === 0) return;
439+
current.drainScheduled = true;
440+
const scheduled = current;
441+
queueMicrotask(() => {
442+
if (state !== scheduled) return;
443+
flushFileLogSinkSync();
444+
});
445+
}
446+
399447
/**
400448
* Writes the buffered tail. Crash and exit paths must call this: the hot
401449
* path only queues, and `process.exit` does not run the microtask drain.
@@ -407,28 +455,25 @@ export function flushFileLogSinkSync(): void {
407455
const current = state;
408456
if (current === null) return;
409457
current.drainScheduled = false;
410-
if (current.buffer.length === 0) return;
411458
const lines = current.buffer;
412459
current.buffer = [];
413-
const locked = withSinkLock(`${current.path}.lock`, () => {
460+
const outcome = withSinkLock(`${current.path}.lock`, () => {
414461
adoptLeftoverRotation(current);
462+
if (lines.length === 0) return;
415463
adoptActive(current);
416464
writeLines(current, lines);
417465
});
418-
if (!locked) current.buffer = [...lines, ...current.buffer];
466+
if (outcome !== "ok") {
467+
current.buffer = [...lines, ...current.buffer];
468+
if (outcome === "busy") scheduleDrain(current);
469+
}
419470
}
420471

421472
function enqueue(record: LogRecord): void {
422473
const current = state;
423474
if (current === null) return;
424475
current.buffer.push(formatRecord(record));
425-
if (current.drainScheduled) return;
426-
current.drainScheduled = true;
427-
const scheduled = current;
428-
queueMicrotask(() => {
429-
if (state !== scheduled) return;
430-
flushFileLogSinkSync();
431-
});
476+
scheduleDrain(current);
432477
}
433478

434479
interface FileLogSinkOptions {

0 commit comments

Comments
 (0)