fix(subscribe): seed the rotation resume marker when tailing an existing log

changelog.d/fixed-297.md

@@ -0,0 +1 @@
+- `pluk subscribe` without `--from-beginning` no longer replays the kept tail of the log as new events when the log is rotated before the first new line arrives; the subscriber now remembers the last line already in the file at start-up and resumes after it.

src/subscriber.ts

@@ -47,7 +47,12 @@ export class Subscriber extends EventEmitter {
   private verbose: boolean;
   private counters: SubscriberStats = { linesRead: 0, malformedLines: 0, eventsFiltered: 0, eventsEmitted: 0 };
   private eventCount = 0;
-  /** Last non-blank line consumed — the resume marker after a rotation. */
+  /**
+   * Last non-blank line consumed — the resume marker after a rotation. When
+   * tailing an existing log without fromBeginning it is seeded with the last
+   * complete line already in the file, so a rotation before the first new
+   * line does not replay the kept tail.
+   */
   private lastLine = '';
   /**
    * Set when a rotation re-read found no complete line yet: the next bytes
@@ -118,6 +123,7 @@ export class Subscriber extends EventEmitter {
       // Holds an incomplete multi-byte sequence across read boundaries.
       let decoder = new StringDecoder('utf8');
       this.head = await readHead(fh);
+      if (!this.fromBeginning && position > 0) await this.seedResumeMarker(fh, position);
       this.log(`tailing from position ${position}${this.filterSet ? ` (filter: ${[...this.filterSet].join(',')})` : ''}`);
 
       while (!this.aborted) {
@@ -187,6 +193,9 @@ export class Subscriber extends EventEmitter {
    * progress: it is re-read a bounded number of times, and if it is still
    * empty the resume is deferred until content appears, so the kept tail is
    * never replayed from byte 0 without the marker check.
+   *
+   * The marker is the last consumed line, or — if nothing was consumed yet —
+   * the line seeded by {@link seedResumeMarker} at start-up.
    */
   private async resumeAfterRotation(fh: FileHandle): Promise<{ position: number; partial: string; decoder: StringDecoder }> {
     this.head = await readHead(fh);
@@ -209,6 +218,34 @@ export class Subscriber extends EventEmitter {
     return { position, partial, decoder };
   }
 
+  /**
+   * Seed {@link lastLine} with the last complete (newline-terminated)
+   * non-blank line in the first `size` bytes, without consuming or emitting
+   * anything. Reads only the final chunk unless it holds fewer than two
+   * newlines (a mid-file chunk needs one to cut the leading partial line and
+   * one to terminate a complete line).
+   */
+  private async seedResumeMarker(fh: FileHandle, size: number): Promise<void> {
+    let start = Math.max(0, size - READ_CHUNK_BYTES);
+    const buf = Buffer.alloc(size - start);
+    const { bytesRead } = await fh.read(buf, 0, buf.length, start);
+    let text = buf.subarray(0, bytesRead).toString('utf8');
+    if (start > 0 && text.indexOf('\n') === text.lastIndexOf('\n')) {
+      text = (await readFromStart(fh)).text.subarray(0, size).toString('utf8');
+      start = 0;
+    }
+    const lines = text.split('\n');
+    lines.pop();
+    // The first line of a mid-file chunk may be cut; only trust it when the chunk starts at byte 0.
+    if (start > 0) lines.shift();
+    for (let i = lines.length - 1; i >= 0; i--) {
+      if (lines[i].trim()) {
+        this.lastLine = lines[i];
+        return;
+      }
+    }
+  }
+
   /** True when the log's leading bytes differ from those remembered. */
   private async headChanged(fh: FileHandle): Promise<boolean> {
     const current = await readHead(fh);

test/subscriber.test.js

@@ -347,11 +347,68 @@ test('rotation: when the last seen line was trimmed away, the whole kept tail is
   }
 });
 
-test('rotation: before any line was seen, the kept tail is delivered and a trailing partial line is held', async () => {
+test('rotation: tailing an existing log without fromBeginning does not replay the kept tail', async () => {
+  const dir = makeRunDir();
+  const log = join(dir, 'logs', 'rot4.jsonl');
+  let body = '';
+  for (let i = 0; i < 200; i++) body += seqLine(i);
+  writeFileSync(log, body);
+  const { sub, events, done } = await startTail(dir, 'rot4');
+  try {
+    await new Promise(r => setTimeout(r, 250));
+    assert.equal(rotateLogFileIfNeeded(log, { maxBytes: 10_000, keepLines: 50 }), true);
+    await new Promise(r => setTimeout(r, 500));
+    assert.equal(events.length, 0, 'pre-existing lines must not be replayed');
+
+    appendFileSync(log, seqLine(999, 'bypass_permissions'));
+    await waitFor(() => events.length >= 1);
+    await new Promise(r => setTimeout(r, 300));
+    assert.equal(events.length, 1);
+    assert.equal(events[0].type, 'bypass_permissions');
+  } finally {
+    sub.stop();
+    await done;
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('rotation: the resume marker is seeded even when the last complete line exceeds the tail chunk', async () => {
+  const dir = makeRunDir();
+  const log = join(dir, 'logs', 'rot4b.jsonl');
+  let body = '';
+  for (let i = 0; i < 20; i++) body += seqLine(i);
+  // One trailing line larger than the 16 KB tail chunk: the chunk holds at most one newline.
+  const big = JSON.stringify({
+    v: 1, ts: new Date().toISOString(), seq: 20, pid: 1,
+    session: 's', pane: 'p', source: 't', type: 'raw_output', data: { pad: 'x'.repeat(20_000) },
+  }) + '\n';
+  body += big;
+  writeFileSync(log, body);
+  const { sub, events, done } = await startTail(dir, 'rot4b');
+  try {
+    await new Promise(r => setTimeout(r, 250));
+    assert.equal(rotateLogFileIfNeeded(log, { maxBytes: 10_000, keepLines: 10 }), true);
+    await new Promise(r => setTimeout(r, 500));
+    assert.equal(events.length, 0, 'pre-existing lines must not be replayed');
+
+    appendFileSync(log, seqLine(999, 'bypass_permissions'));
+    await waitFor(() => events.length >= 1);
+    await new Promise(r => setTimeout(r, 300));
+    assert.equal(events.length, 1);
+    assert.equal(events[0].type, 'bypass_permissions');
+  } finally {
+    sub.stop();
+    await done;
+    rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('rotation: before any line was seen, the kept tail is skipped and a trailing partial line is held', async () => {
   const dir = makeRunDir();
   const log = join(dir, 'logs', 'rot4.jsonl');
   // Start at EOF of a large file, then rotate without the subscriber ever
-  // having consumed a line (lastLine is empty → no marker → replay all).
+  // having consumed a line. The resume marker was seeded from the last
+  // complete line at start-up, so nothing already in the file is replayed.
   let body = '';
   for (let i = 0; i < 100; i++) body += seqLine(i);
   writeFileSync(log, body);
@@ -363,12 +420,14 @@ test('rotation: before any line was seen, the kept tail is delivered and a trail
     // Append a partial line so the resume path must carry it over.
     const half = seqLine(500, 'error');
     appendFileSync(log, half.slice(0, 10));
-    await waitFor(() => events.length >= 5);
-    assert.deepEqual(events.map(e => e.seq), [95, 96, 97, 98, 99]);
+    await new Promise(r => setTimeout(r, 500));
+    assert.equal(events.length, 0, 'kept tail must not be replayed and the partial line must be held');
     appendFileSync(log, half.slice(10));
-    await waitFor(() => events.length >= 6);
-    assert.equal(events[5].seq, 500);
-    assert.equal(events[5].type, 'error');
+    await waitFor(() => events.length >= 1);
+    await new Promise(r => setTimeout(r, 200));
+    assert.equal(events.length, 1);
+    assert.equal(events[0].seq, 500);
+    assert.equal(events[0].type, 'error');
   } finally {
     sub.stop();
     await done;