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

doc/operations.md

@@ -148,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 |
@@ -185,11 +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 or
+
-startup-replay burst problem, and a bigger buffer only buys time. The HTTP
+`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 distinguish configuration from demand.
+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,11 +152,20 @@ 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.

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{
@@ -536,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

@@ -1,7 +1,9 @@
 package http
 
 import (
+	"context"
 	"encoding/json"
+	"net/http"
 	"net/http/httptest"
 	"testing"
 	"time"
@@ -73,3 +75,93 @@ func TestStatusReportsQueueAndConnectionBounds(t *testing.T) {
 
 	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",
@@ -296,6 +303,25 @@ func (fs *FileSource) ensureWatcher(path string) {
 	}()
 }
 
+// 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.

internal/source/file/file_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"syscall"
 	"testing"
 	"time"
 
@@ -44,3 +45,40 @@ func TestRemoveWatcherPreservesReplacement(t *testing.T) {
 		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
@@ -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()