v0.18.2 resume rotated files instead of replay, keep quiet SSE alive, refuse HEAD on strem path

Makefile

@@ -33,7 +33,7 @@ install: build
 
 # Uninstall the binary
 uninstall:
-	rm -f $(BINDIR)/$(BINARY_PATH)
+	rm -f $(BINDIR)/$(BINARY_NAME)
 
 # Clean build artifacts
 clean:

README.md

@@ -6,7 +6,7 @@
     <td>
       <h1>LogWisp</h1>
       <p>
-        <a href="https://golang.org"><img src="https://img.shields.io/badge/Go-1.26-00ADD8?style=flat&logo=go" alt="Go"></a>
+        <a href="https://golang.org"><img src="https://img.shields.io/badge/Go-1.27.1-00ADD8?style=flat&logo=go" alt="Go"></a>
         <a href="https://opensource.org/licenses/BSD-3-Clause"><img src="https://img.shields.io/badge/License-BSD_3--Clause-blue.svg" alt="License"></a>
         <a href="doc/"><img src="https://img.shields.io/badge/Docs-Available-green.svg" alt="Documentation"></a>
       </p>
@@ -137,7 +137,7 @@ synthetic generator writing JSON to stdout.
 
 - **Operating systems**: Linux (kernel 6.10+), FreeBSD (14.0+)
 - **Architecture**: amd64
-- **Go**: 1.26+ to build from source
+- **Go**: 1.27.1+ to build from source
 
 Network sources and sinks bind and dial over IPv4 only.
 

doc/installation.md

@@ -22,7 +22,7 @@ The Makefile works with both GNU make and BSD make. Targets:
 | `make` / `make build` | Build `bin/logwisp` with version metadata |
 | `make dev` | Build with the race detector enabled |
 | `make install` | Install the binary to `$(PREFIX)/bin` (default `/usr/local`) |
-| `make uninstall` | Intended to remove the installed binary — currently broken: it expands to `$(BINDIR)/bin/logwisp` instead of `$(BINDIR)/logwisp`, so it removes nothing. Delete the binary by hand |
+| `make uninstall` | Remove `$(BINDIR)/logwisp` |
 | `make clean` | Remove the built binary |
 | `make version` | Print the version, commit, and build time that would be embedded |
 

doc/operations.md

@@ -124,6 +124,9 @@ curl -s http://127.0.0.1:8080/status | jq .
     "tls": false,
     "active_clients": 3,
     "buffer_size": 1000,
+    "client_buffer_size": 256,
+    "max_connections": 32,
+    "write_timeout_ms": 5000,
     "uptime_seconds": 8130
   },
   "endpoints": { "stream": "/stream", "status": "/status" },
@@ -145,7 +148,7 @@ This endpoint is scoped to one sink, not to the whole process, and it is
 | `dropped_entries` | source | Downstream cannot keep up with the source |
 | `total_dropped` | flow | Rate limit or filters are discarding entries (often intended) |
 | `total_dropped_by_sink` | pipeline | A sink's input queue is full |
-| `dropped_writes` | tcp/http sink | A specific client is too slow |
+| `dropped_writes` | tcp/http sink | A client's queue overflowed: either it is too slow, or one burst exceeded `client_buffer_size` |
 | `rejected_conns` / `rejected_clients` | tcp/http sink, tcp_chain source | `max_connections` is being hit |
 | `tls_handshake_errors` | tcp sink, tcp_chain source | Certificate or version mismatch, or scanning |
 | `parse_errors` | chain source | Protocol or version skew upstream |
@@ -182,9 +185,18 @@ the filter stage logs several lines per entry evaluated.
 ### Buffers
 
 Raise `buffer_size` when `total_dropped_by_sink` is climbing but the sink itself
-is healthy — that is a burst-absorption problem. Raise `client_buffer_size` when
+is healthy — that is a burst-absorption problem.
-`dropped_writes` is climbing for network sinks; that is a slow-consumer problem,
+
-and a bigger buffer only buys time.
+`dropped_writes` on a network sink has two causes that a counter alone does not
+separate. A consumer slower than the sustained rate cannot be bought off with
+buffer, and drops are the intended outcome. A burst the consumer would have
+drained, arriving faster than it reads, is configuration: the sink queues a
+whole burst while the client writes one frame at a time, so the part of a burst
+above `client_buffer_size` is lost even to a loopback reader.
+Where a `rate_limit` bounds the pipeline, its `burst` is that number — keep
+`client_buffer_size` at or above it and the second cause disappears. The HTTP
+status endpoint reports both queue bounds alongside the counters so an operator
+can tell which one is in play.
 
 ```toml
 [pipelines.plugin_sinks.config]

doc/sinks.md

@@ -140,7 +140,9 @@ allow = ["viewer-01"]
 
 **Behaviour**
 
-- Only `GET` is routed to either path; anything else gets `405`.
+- Only `GET` is routed to either path; anything else gets `405`, `HEAD` on
+  `stream_path` included — a stream is a body, and a client registered to have
+  its body discarded never reads and never leaves.
 - With an `auth` block, one middleware gates **both** endpoints: an
   unauthorized client gets `403` with no body detail, and the rejection is
   logged at WARN and counted in `auth_rejected`. The authorized identity is
@@ -150,19 +152,29 @@ allow = ["viewer-01"]
 - Payloads are framed per the SSE spec, one `data:` line per newline in the
   payload, so multi-line entries stream correctly.
 - The server sets no `WriteTimeout` (that would kill long-lived streams);
-  per-write deadlines come from `write_timeout_ms` via `http.ResponseController`.
+  per-write deadlines come from `write_timeout_ms` via `http.ResponseController`
+  and cover the connected frame, every payload, and the idle comment.
+- A quiet stream emits an SSE comment every 15 s. It refreshes the client's
+  session and is how a peer that stopped reading is noticed.
 - A client whose send queue is full has that event dropped
-  (`dropped_writes`); it is not disconnected.
+  (`dropped_writes`); it is not disconnected. A `dropped_writes` that rises while
+  no client is behind is a burst larger than `client_buffer_size`, not
+  backpressure: size the queue at or above whatever burst the pipeline's
+  `rate_limit` releases at once.
+- A client is registered only once its connected frame has flushed, so the
+  broker never queues into a buffer whose reader has not started.
 - Clients whose session has been idle-expired by the session manager are
-  evicted by the broker.
+  evicted by the broker. With the idle comment above, that reaches only a peer
+  that has stopped accepting bytes on a sink configured `write_timeout_ms = 0`.
 - On shutdown, connected clients receive
   `event: disconnect / data: {"reason":"server_shutdown"}`.
 - HTTP/2 is negotiated via ALPN when TLS is enabled; plaintext is HTTP/1.1.
 
 **Status endpoint** returns service and version identity, host, port, TLS flag,
-the compiled auth policy, active client count, buffer size, uptime, endpoint
+the compiled auth policy, active client count, sink and per-client buffer sizes,
-paths, and the `total_processed` / `dropped_writes` / `rejected_clients` /
+connection limit, write timeout, uptime, endpoint paths, and the
-`auth_rejected` counters.
+`total_processed` / `dropped_writes` / `rejected_clients` / `auth_rejected`
+counters.
 
 > Without an `auth` block both endpoints are unauthenticated, and the stream
 > response carries `Access-Control-Allow-Origin: *`, so any web origin can read

doc/sources.md

@@ -60,6 +60,10 @@ from              = "end"
   beyond end-of-file, or an inode change. An inode change where the new file is
   already larger than the recorded position is treated as an atomic save, not a
   rotation, and the position is preserved.
+- A rotation that renames in place — what a size-capped writer does — puts the
+  same inode back under a name `pattern` also matches. Its watcher resumes at
+  the position the original reached, so `from = "start"` reads the tail an
+  unfinished read left behind rather than the whole archive a second time.
 - A line is parsed as JSON only when it is an object whose top-level keys are
   all drawn from `time`, `level`, `msg` and `fields` — the four an entry can
   carry. `time` is read as RFC3339Nano. Any other key, and any non-object line,

internal/core/const.go

@@ -13,6 +13,10 @@ const (
 
 	SessionCleanupInterval = 5 * time.Minute
 
+	// Idle keepalive for a served stream. Well under SessionDefaultMaxIdleTime,
+	// so a quiet stream refreshes its session long before the sweep expires it.
+	StreamKeepaliveInterval = 15 * time.Second
+
 	ServiceStatsUpdateInterval = 1 * time.Second
 
 	ShutdownTimeout = 10 * time.Second

internal/sink/http/http.go

@@ -68,6 +68,7 @@ type HTTPSink struct {
 	clientsMu    sync.Mutex
 	nextClientID atomic.Uint64
 	writeTimeout time.Duration
+	keepalive    time.Duration
 
 	// TLS
 	tlsConfig *tls.Config
@@ -150,6 +151,7 @@ func NewHTTPSinkPlugin(
 		logger:       logger,
 		clients:      make(map[uint64]*sseClient),
 		writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond,
+		keepalive:    core.StreamKeepaliveInterval,
 		tlsConfig:    tlsCfg,
 		auth:         authPolicy,
 	}
@@ -205,6 +207,10 @@ func (h *HTTPSink) Start(ctx context.Context) error {
 	// Method-scoped patterns: mux answers 405 with Allow header on non-GET
 	mux.HandleFunc(http.MethodGet+" "+h.config.StreamPath, h.handleStream)
 	mux.HandleFunc(http.MethodGet+" "+h.config.StatusPath, h.handleStatus)
+	// A GET pattern also serves HEAD, and a HEAD stream is a registered client
+	// whose body writes are discarded: it never reads, so nothing but the peer
+	// closing the connection ends it. The status path answers one either way.
+	mux.HandleFunc(http.MethodHead+" "+h.config.StreamPath, streamHeadNotAllowed)
 
 	// One wrapper covers stream and status, and keeps the handlers themselves
 	// unaware of authorization
@@ -287,9 +293,9 @@ func (h *HTTPSink) shutdown() {
 	})
 }
 
-// removeClient unregisters a client; the first caller closes the send
+// removeClient unregisters a client; the first caller closes the send channel.
-// channel and removes the session. Broker (stale-session eviction) and
+// Broker (stale-session eviction) and stream handler (disconnect) may race here
-// stream handler (disconnect) may race here safely.
+// safely. The session is the handler's, released when it returns.
 func (h *HTTPSink) removeClient(id uint64) {
 	h.clientsMu.Lock()
 	c, ok := h.clients[id]
@@ -299,7 +305,6 @@ func (h *HTTPSink) removeClient(id uint64) {
 	h.clientsMu.Unlock()
 	if ok {
 		close(c.send)
-		h.proxy.RemoveSession(c.sessionID)
 	}
 }
 
@@ -374,10 +379,6 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 	}
 	id := h.nextClientID.Add(1)
 
-	h.clientsMu.Lock()
-	h.clients[id] = c
-	h.clientsMu.Unlock()
-
 	count := h.activeClients.Add(1)
 	h.logger.Debug("msg", "HTTP client connected",
 		"component", "http_sink",
@@ -389,6 +390,7 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 
 	defer func() {
 		h.removeClient(id)
+		h.proxy.RemoveSession(sess.ID)
 		newCount := h.activeClients.Add(-1)
 		h.logger.Debug("msg", "HTTP client disconnected",
 			"component", "http_sink",
@@ -413,11 +415,24 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 		"status_path": h.config.StatusPath,
 		"buffer_size": h.config.ClientBufferSize,
 	})
+	h.armWrite(rc)
 	fmt.Fprintf(w, "event: connected\ndata: %s\n\n", info)
 	if err := rc.Flush(); err != nil {
 		return
 	}
 
+	// Registered only now: a client the broker can queue into before its reader
+	// reaches the loop below loses a burst to a buffer nobody is draining.
+	h.clientsMu.Lock()
+	h.clients[id] = c
+	h.clientsMu.Unlock()
+
+	// A stream with nothing to carry still has to prove the peer is there. The
+	// comment refreshes the session the broker evicts on, and fails on a peer
+	// that stopped reading.
+	idle := time.NewTicker(h.keepalive)
+	defer idle.Stop()
+
 	clientGone := r.Context().Done()
 	for {
 		select {
@@ -425,9 +440,7 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 			if !ok {
 				return // broker evicted (stale session)
 			}
-			if h.writeTimeout > 0 {
+			h.armWrite(rc)
-				_ = rc.SetWriteDeadline(time.Now().Add(h.writeTimeout))
-			}
 			if err := writeSSE(w, payload); err != nil {
 				return
 			}
@@ -435,6 +448,15 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			h.proxy.UpdateActivity(sess.ID)
+		case <-idle.C:
+			h.armWrite(rc)
+			if _, err := fmt.Fprint(w, ":\n\n"); err != nil {
+				return
+			}
+			if err := rc.Flush(); err != nil {
+				return
+			}
+			h.proxy.UpdateActivity(sess.ID)
 		case <-clientGone:
 			return
 		case <-h.done:
@@ -445,6 +467,14 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// armWrite bounds the next response write. Without it an SSE write is unbounded
+// and a peer that stops reading wedges its handler for as long as it stays open.
+func (h *HTTPSink) armWrite(rc *http.ResponseController) {
+	if h.writeTimeout > 0 {
+		_ = rc.SetWriteDeadline(time.Now().Add(h.writeTimeout))
+	}
+}
+
 // handleStatus provides a JSON status report
 func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) {
 	status := map[string]any{
@@ -459,6 +489,9 @@ func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) {
 			"auth":               h.auth.Describe(),
 			"active_clients":     h.activeClients.Load(),
 			"buffer_size":        h.config.BufferSize,
+			"client_buffer_size": h.config.ClientBufferSize,
+			"max_connections":    h.config.MaxConnections,
+			"write_timeout_ms":   h.config.WriteTimeoutMS,
 			"uptime_seconds":     int(time.Since(h.startTime).Seconds()),
 		},
 		"endpoints": map[string]string{
@@ -484,6 +517,9 @@ func (h *HTTPSink) GetStats() sink.SinkStats {
 		"host":               h.config.Host,
 		"port":               h.config.Port,
 		"buffer_size":        h.config.BufferSize,
+		"client_buffer_size": h.config.ClientBufferSize,
+		"max_connections":    h.config.MaxConnections,
+		"write_timeout_ms":   h.config.WriteTimeoutMS,
 		"tls":                h.tlsConfig != nil,
 		"dropped_writes":     h.droppedWrites.Load(),
 		"rejected_clients":   h.rejectedClients.Load(),
@@ -530,6 +566,12 @@ func (h *HTTPSink) authMiddleware(next http.Handler) http.Handler {
 	})
 }
 
+// streamHeadNotAllowed refuses a body-less read of a stream that is only a body
+func streamHeadNotAllowed(w http.ResponseWriter, _ *http.Request) {
+	w.Header().Set("Allow", http.MethodGet)
+	http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+}
+
 // writeSSE frames a payload per the W3C SSE spec (multi-line safe)
 func writeSSE(w http.ResponseWriter, payload []byte) error {
 	for _, line := range splitLines(payload) {

internal/sink/http/http_test.go

@@ -0,0 +1,167 @@
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"logwisp/internal/session"
+	"logwisp/internal/sink"
+
+	"github.com/lixenwraith/log"
+)
+
+func TestStatusReportsQueueAndConnectionBounds(t *testing.T) {
+	manager := session.NewManager(time.Hour)
+	defer manager.Stop()
+	created, err := NewHTTPSinkPlugin(
+		"stream",
+		map[string]any{
+			"host":               "127.0.0.1",
+			"port":               int64(8081),
+			"buffer_size":        int64(4096),
+			"client_buffer_size": int64(512),
+			"max_connections":    int64(32),
+			"write_timeout_ms":   int64(5000),
+		},
+		log.NewLogger(),
+		session.NewProxy(manager, "stream"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	httpSink, ok := created.(*HTTPSink)
+	if !ok {
+		t.Fatalf("sink type = %T", created)
+	}
+
+	recorder := httptest.NewRecorder()
+	httpSink.handleStatus(recorder, httptest.NewRequest("GET", "/status", nil))
+	if recorder.Code != 200 {
+		t.Fatalf("status code = %d", recorder.Code)
+	}
+	var response struct {
+		Server map[string]any `json:"server"`
+	}
+	if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil {
+		t.Fatal(err)
+	}
+	for key, want := range map[string]float64{
+		"buffer_size":        4096,
+		"client_buffer_size": 512,
+		"max_connections":    32,
+		"write_timeout_ms":   5000,
+	} {
+		if got := response.Server[key]; got != want {
+			t.Errorf("server.%s = %v, want %v", key, got, want)
+		}
+	}
+
+	stats := httpSink.GetStats()
+	details := stats.Details
+	for key, want := range map[string]int64{
+		"buffer_size":        4096,
+		"client_buffer_size": 512,
+		"max_connections":    32,
+		"write_timeout_ms":   5000,
+	} {
+		if got := details[key]; got != want {
+			t.Errorf("details[%q] = %v, want %v", key, got, want)
+		}
+	}
+
+	var _ sink.Sink = httpSink
+}
+
+// A stream carrying nothing still refreshes its session. Log traffic is what
+// bumps activity otherwise, so a quiet source would idle-expire a healthy client
+// and the broker would evict it on the next entry.
+func TestQuietStreamRefreshesItsSession(t *testing.T) {
+	manager := session.NewManager(time.Hour)
+	defer manager.Stop()
+	created, err := NewHTTPSinkPlugin(
+		"stream",
+		map[string]any{"host": "127.0.0.1", "port": int64(18191), "write_timeout_ms": int64(5000)},
+		log.NewLogger(),
+		session.NewProxy(manager, "stream"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	httpSink := created.(*HTTPSink)
+	httpSink.keepalive = 100 * time.Millisecond
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if err := httpSink.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer httpSink.Stop()
+
+	resp, err := http.Get("http://127.0.0.1:18191/stream")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	activity := func() time.Time {
+		for _, s := range manager.GetActiveSessions() {
+			return s.LastActivity
+		}
+		t.Fatal("no session for the connected client")
+		return time.Time{}
+	}
+
+	deadline := time.Now().Add(2 * time.Second)
+	for activity().IsZero() && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	before := activity()
+
+	// No events are sent for several keepalive periods.
+	time.Sleep(350 * time.Millisecond)
+	if after := activity(); !after.After(before) {
+		t.Fatalf("last activity %v did not advance on a silent stream", after)
+	}
+}
+
+// HEAD on the stream path is refused rather than served from the GET pattern:
+// its body writes are discarded, so the client it would register never reads.
+func TestHeadOnStreamPathIsRefused(t *testing.T) {
+	manager := session.NewManager(time.Hour)
+	defer manager.Stop()
+	created, err := NewHTTPSinkPlugin(
+		"stream",
+		map[string]any{"host": "127.0.0.1", "port": int64(18192)},
+		log.NewLogger(),
+		session.NewProxy(manager, "stream"),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	httpSink := created.(*HTTPSink)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if err := httpSink.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer httpSink.Stop()
+
+	resp, err := http.Head("http://127.0.0.1:18192/stream")
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp.Body.Close()
+	if resp.StatusCode != http.StatusMethodNotAllowed {
+		t.Fatalf("HEAD /stream = %d, want %d", resp.StatusCode, http.StatusMethodNotAllowed)
+	}
+	if got := resp.Header.Get("Allow"); got != http.MethodGet {
+		t.Errorf("Allow = %q, want %q", got, http.MethodGet)
+	}
+	if n := manager.GetSessionCount(); n != 0 {
+		t.Errorf("sessions after HEAD = %d, want 0", n)
+	}
+}

internal/source/file/file.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"logwisp/internal/config"
@@ -270,6 +271,12 @@ func (fs *FileSource) ensureWatcher(path string) {
 	}
 
 	w := newFileWatcher(path, fs.config.Raw, fs.config.From == "start", fs.publish, fs.logger)
+	// A rotation renames the file out from under its watcher, so the same inode
+	// reappears here under the archive name. Resume where it was left: from the
+	// start would re-emit every record the file has already delivered.
+	if position, ok := fs.readPosition(path); ok {
+		w.position = position
+	}
 	fs.watchers[path] = w
 
 	fs.logger.Debug("msg", "Created file watcher",
@@ -292,12 +299,40 @@ func (fs *FileSource) ensureWatcher(path string) {
 			}
 		}
 
-		fs.mu.Lock()
+		fs.removeWatcher(path, w)
-		delete(fs.watchers, path)
-		fs.mu.Unlock()
 	}()
 }
 
+// readPosition reports how far a running watcher has read the file now at path.
+// Callers hold fs.mu.
+func (fs *FileSource) readPosition(path string) (int64, bool) {
+	info, err := os.Stat(path)
+	if err != nil {
+		return 0, false
+	}
+	stat, ok := info.Sys().(*syscall.Stat_t)
+	if !ok {
+		return 0, false
+	}
+	for _, w := range fs.watchers {
+		if position, ok := w.readTo(stat.Ino); ok {
+			return position, true
+		}
+	}
+	return 0, false
+}
+
+// removeWatcher removes only the watcher that finished. A deleted file can be
+// recreated before its old watcher observes stop; in that case ensureWatcher
+// has already installed a replacement under the same path, which must survive.
+func (fs *FileSource) removeWatcher(path string, watcher *fileWatcher) {
+	fs.mu.Lock()
+	if fs.watchers[path] == watcher {
+		delete(fs.watchers, path)
+	}
+	fs.mu.Unlock()
+}
+
 // cleanupWatchers stops and removes watchers for files that no longer exist.
 func (fs *FileSource) cleanupWatchers() {
 	fs.mu.Lock()
@@ -369,4 +404,3 @@ func globToRegex(glob string) string {
 	regex = strings.ReplaceAll(regex, `\?`, `.`)
 	return "^" + regex + "$"
 }
-

internal/source/file/file_test.go

@@ -0,0 +1,84 @@
+package file
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"syscall"
+	"testing"
+	"time"
+
+	"logwisp/internal/core"
+)
+
+func TestStoppedWatcherReturnsNormally(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "session.jsonl")
+	if err := os.WriteFile(path, nil, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	watcher := newFileWatcher(path, true, true, func(_ core.LogEntry) {}, nil)
+	watcher.stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	if err := watcher.watch(ctx); err != nil {
+		t.Fatalf("stopped watcher returned error: %v", err)
+	}
+}
+
+func TestRemoveWatcherPreservesReplacement(t *testing.T) {
+	oldWatcher := &fileWatcher{}
+	replacement := &fileWatcher{}
+	source := &FileSource{
+		watchers: map[string]*fileWatcher{
+			"session.jsonl": replacement,
+		},
+	}
+
+	source.removeWatcher("session.jsonl", oldWatcher)
+	if got := source.watchers["session.jsonl"]; got != replacement {
+		t.Fatalf("replacement watcher = %p, want %p", got, replacement)
+	}
+
+	source.removeWatcher("session.jsonl", replacement)
+	if _, exists := source.watchers["session.jsonl"]; exists {
+		t.Fatal("finished watcher was not removed")
+	}
+}
+
+// A rotated file reappears under its archive name with the same inode. Its
+// replacement watcher resumes where the original stopped, so a `from = "start"`
+// source does not re-emit every record the file already delivered.
+func TestRotatedFileResumesInsteadOfReplaying(t *testing.T) {
+	dir := t.TempDir()
+	active := filepath.Join(dir, "session.jsonl")
+	if err := os.WriteFile(active, []byte("one\ntwo\n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	info, err := os.Stat(active)
+	if err != nil {
+		t.Fatal(err)
+	}
+	inode := info.Sys().(*syscall.Stat_t).Ino
+
+	archive := filepath.Join(dir, "session_260916_120000.jsonl")
+	if err := os.Rename(active, archive); err != nil {
+		t.Fatal(err)
+	}
+
+	for name, watcher := range map[string]*fileWatcher{
+		"still tailing the renamed inode": {inode: inode, position: 8},
+		"already moved on from it":        {inode: 99, prevInode: inode, prevPosition: 8},
+	} {
+		source := &FileSource{watchers: map[string]*fileWatcher{active: watcher}}
+		position, ok := source.readPosition(archive)
+		if !ok || position != 8 {
+			t.Errorf("%s: position = %d, ok = %v, want 8, true", name, position, ok)
+		}
+	}
+
+	unrelated := &FileSource{watchers: map[string]*fileWatcher{active: {inode: 99}}}
+	if _, ok := unrelated.readPosition(archive); ok {
+		t.Error("a file no watcher has read was treated as rotated")
+	}
+}

internal/source/file/file_watcher.go

@@ -42,6 +42,8 @@ type fileWatcher struct {
 	mu           sync.Mutex
 	stopped      bool
 	rotationSeq  int64
+	prevInode    uint64
+	prevPosition int64
 	entriesRead  atomic.Uint64
 	lastReadTime atomic.Value // time.Time
 	logger       *log.Logger
@@ -80,7 +82,7 @@ func (w *fileWatcher) watch(ctx context.Context) error {
 			return ctx.Err()
 		case <-ticker.C:
 			if w.isStopped() {
-				return fmt.Errorf("watcher stopped")
+				return nil
 			}
 			if err := w.checkFile(); err != nil {
 				// Log error but continue watching
@@ -220,6 +222,9 @@ func (w *fileWatcher) checkFile() error {
 		w.mu.Lock()
 		w.rotationSeq++
 		seq := w.rotationSeq
+		// Retained for the source: the renamed file is about to be discovered
+		// under its archive name, and only this says how much of it was read.
+		w.prevInode, w.prevPosition = oldInode, oldPos
 		w.inode = currentInode
 		w.position = 0 // Reset position on rotation
 		w.mu.Unlock()
@@ -347,6 +352,23 @@ func (w *fileWatcher) initPosition() error {
 	return nil
 }
 
+// readTo reports how far this watcher read the given inode: the file it tails
+// now, or the one a rotation renamed out from under it.
+func (w *fileWatcher) readTo(inode uint64) (int64, bool) {
+	if inode == 0 {
+		return 0, false
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	switch inode {
+	case w.inode:
+		return w.position, true
+	case w.prevInode:
+		return w.prevPosition, true
+	}
+	return 0, false
+}
+
 // isStopped checks if the watcher has been instructed to stop
 func (w *fileWatcher) isStopped() bool {
 	w.mu.Lock()