From c3723dc0e6b76d9f2a310103c55090e8fae3997a Mon Sep 17 00:00:00 2001 From: Tobias Gesellchen Date: Tue, 19 May 2026 20:06:22 +0200 Subject: [PATCH] feat(service): add Logs tab streaming the live stderr trace MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cloud-deploy operators on Discussion #295 needed to leave the admin UI for docker logs / journalctl to see what the service was doing. Mirror log.Default() output into an in-memory ring buffer and expose it under /setup/logs so the admin UI can show a live trace alongside the existing tabs. The buffer is a second sink under log.SetOutput(io.MultiWriter( os.Stderr, buf)) — stderr keeps receiving every line verbatim, so docker logs / journalctl are unaffected. Default capacity 2000 lines (~400 KB), tunable via SOUNDTOUCH_LOG_BUFFER_LINES. - pkg/service/logbuf: io.Writer ring with \n splitting, partial-line buffering, monotonic Seq, Since(since, limit) reporting dropped count when the caller falls behind. - New /setup/logs (GET) returns {entries, nextSince, dropped, capacity}. Polls at 1.5s while the tab is active; paused on document.hidden. - "8. Logs" tab with substring filter, tail-follow toggle (auto-disables when the user scrolls up), monospace dark view. Co-Authored-By: Claude Opus 4.7 (1M context) --- cmd/soundtouch-service/main.go | 40 +++ .../testdata/router_routes.txt | 1 + pkg/service/handlers/handlers_logs.go | 86 ++++++ pkg/service/handlers/handlers_logs_test.go | 179 ++++++++++++ pkg/service/handlers/server.go | 22 ++ pkg/service/handlers/web/index.html | 23 ++ pkg/service/handlers/web/js/script.js | 139 +++++++++ pkg/service/logbuf/logbuf.go | 189 +++++++++++++ pkg/service/logbuf/logbuf_test.go | 264 ++++++++++++++++++ 9 files changed, 943 insertions(+) create mode 100644 pkg/service/handlers/handlers_logs.go create mode 100644 pkg/service/handlers/handlers_logs_test.go create mode 100644 pkg/service/logbuf/logbuf.go create mode 100644 pkg/service/logbuf/logbuf_test.go diff --git a/cmd/soundtouch-service/main.go b/cmd/soundtouch-service/main.go index 034fca1..045fbcd 100644 --- a/cmd/soundtouch-service/main.go +++ b/cmd/soundtouch-service/main.go @@ -7,6 +7,7 @@ import ( "crypto/tls" "encoding/json" "fmt" + "io" "log" "net" "net/http" @@ -15,6 +16,7 @@ import ( "path/filepath" "runtime" "runtime/debug" + "strconv" "strings" "time" @@ -23,6 +25,7 @@ import ( "github.com/gesellix/bose-soundtouch/pkg/service/certmanager" "github.com/gesellix/bose-soundtouch/pkg/service/datastore" "github.com/gesellix/bose-soundtouch/pkg/service/handlers" + "github.com/gesellix/bose-soundtouch/pkg/service/logbuf" "github.com/gesellix/bose-soundtouch/pkg/service/proxy" "github.com/gesellix/bose-soundtouch/pkg/service/setup" "github.com/gesellix/bose-soundtouch/pkg/service/spotify" @@ -173,9 +176,44 @@ func initMusicServices(config serviceConfig, server *handlers.Server) { } } +// logBufferCapacityFromEnv reads SOUNDTOUCH_LOG_BUFFER_LINES and +// returns a positive capacity. Invalid or unset values fall back +// to the default; a value of 0 or negative is treated as "disable" +// and returns 0 so the caller can skip wiring the buffer. +func logBufferCapacityFromEnv(defaultCap int) int { + raw := os.Getenv("SOUNDTOUCH_LOG_BUFFER_LINES") + if raw == "" { + return defaultCap + } + + v, err := strconv.Atoi(raw) + if err != nil { + log.Printf("[Logs] Invalid SOUNDTOUCH_LOG_BUFFER_LINES=%q, using default %d", raw, defaultCap) + return defaultCap + } + + if v < 0 { + return 0 + } + + return v +} + func main() { updateBuildInfo() + // Mirror log output to an in-memory ring buffer so the admin + // UI can show a live trace. Stderr keeps receiving every line + // verbatim — the buffer is a second sink, not a replacement. + // Installing this before the cli.Action runs means every + // log.Printf from initialisation onwards is captured. + var logBuf *logbuf.Buffer + + if bufCap := logBufferCapacityFromEnv(2000); bufCap > 0 { + logBuf = logbuf.New(bufCap) + log.SetOutput(io.MultiWriter(os.Stderr, logBuf)) + } + app := &cli.App{ Name: "soundtouch-service", Usage: "Local service for Bose SoundTouch cloud emulation and management", @@ -395,6 +433,7 @@ func main() { sm.MgmtPassword = config.mgmtPassword server := handlers.NewServer(ds, sm, config.serverURL, config.redact, config.logBody, config.record) sm.GetDNSRunning = server.GetDNSRunning + server.SetLogBuffer(logBuf) server.SetHTTPServerURL(config.httpsServerURL) server.SetVersionInfo(version, commit, date, repoURL) server.SetDiscoverySettings(config.discoveryInterval, config.discoveryEnabled) @@ -1156,6 +1195,7 @@ func setupRouter(server *handlers.Server, stockholmHandler *stockholm.Handler) * r.Get("/health", server.HandleHealthChecks) r.Post("/health/fix", server.HandleHealthFix) + r.Get("/logs", server.HandleGetLogs) // Serve Stockholm setup wizard pages for paths not matched by the management API. // The Stockholm frontend has a setup/ directory that must be accessible at /setup/*. diff --git a/cmd/soundtouch-service/testdata/router_routes.txt b/cmd/soundtouch-service/testdata/router_routes.txt index 4e8d94a..1cb90bf 100644 --- a/cmd/soundtouch-service/testdata/router_routes.txt +++ b/cmd/soundtouch-service/testdata/router_routes.txt @@ -65,6 +65,7 @@ GET /setup/interaction-stats handlers.( GET /setup/interactions handlers.(*Server).HandleListInteractions-fm GET /setup/interactions/sessions/{session}/download handlers.(*Server).HandleDownloadSession-fm GET /setup/logging-settings handlers.(*Server).HandleGetLoggingSettings-fm +GET /setup/logs handlers.(*Server).HandleGetLogs-fm GET /setup/settings handlers.(*Server).HandleGetSettings-fm GET /setup/summary/{deviceId} handlers.(*Server).HandleGetMigrationSummary-fm GET /setup/version handlers.(*Server).HandleGetVersionInfo-fm diff --git a/pkg/service/handlers/handlers_logs.go b/pkg/service/handlers/handlers_logs.go new file mode 100644 index 0000000..939d3dd --- /dev/null +++ b/pkg/service/handlers/handlers_logs.go @@ -0,0 +1,86 @@ +package handlers + +import ( + "encoding/json" + "net/http" + "strconv" + + "github.com/gesellix/bose-soundtouch/pkg/service/logbuf" +) + +// logsResponse is the wire shape for GET /setup/logs. Entries is +// ordered by Seq ascending. NextSince is the highest Seq returned, +// suitable for the client's next poll's `since` parameter. Dropped +// is the count of entries the client missed because the ring +// evicted them before the client polled — a non-zero value +// signals the client to surface a gap to the operator. +type logsResponse struct { + Entries []logbuf.Entry `json:"entries"` + NextSince uint64 `json:"nextSince"` + Dropped uint64 `json:"dropped"` + Capacity int `json:"capacity"` +} + +// HandleGetLogs returns recent log entries from the in-process +// ring buffer. Query parameters: +// +// since — return entries with Seq strictly greater than this +// value. Omitted or "0" → full snapshot. +// limit — cap the number of entries returned. Omitted → no cap +// beyond the buffer's capacity. +// +// When no log buffer is attached (e.g. tests, or the env opted +// out via SOUNDTOUCH_LOG_BUFFER_LINES=0), the response is an +// empty snapshot rather than an error so the UI degrades +// gracefully. +func (s *Server) HandleGetLogs(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + + since, err := parseUint64Query(q.Get("since"), 0) + if err != nil { + writeJSONError(w, http.StatusBadRequest, "invalid since: "+err.Error()) + return + } + + limit, err := parseIntQuery(q.Get("limit"), 0) + if err != nil { + writeJSONError(w, http.StatusBadRequest, "invalid limit: "+err.Error()) + return + } + + resp := logsResponse{ + Entries: []logbuf.Entry{}, + NextSince: since, + } + + if buf := s.LogBuffer(); buf != nil { + entries, nextSince, dropped := buf.Since(since, limit) + resp.Entries = entries + resp.NextSince = nextSince + resp.Dropped = dropped + resp.Capacity = buf.Capacity() + } + + w.Header().Set("Content-Type", "application/json") + + if err := json.NewEncoder(w).Encode(resp); err != nil { + http.Error(w, "Failed to encode response", http.StatusInternalServerError) + return + } +} + +func parseUint64Query(raw string, defaultVal uint64) (uint64, error) { + if raw == "" { + return defaultVal, nil + } + + return strconv.ParseUint(raw, 10, 64) +} + +func parseIntQuery(raw string, defaultVal int) (int, error) { + if raw == "" { + return defaultVal, nil + } + + return strconv.Atoi(raw) +} diff --git a/pkg/service/handlers/handlers_logs_test.go b/pkg/service/handlers/handlers_logs_test.go new file mode 100644 index 0000000..22dea8e --- /dev/null +++ b/pkg/service/handlers/handlers_logs_test.go @@ -0,0 +1,179 @@ +package handlers + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gesellix/bose-soundtouch/pkg/service/logbuf" + "github.com/go-chi/chi/v5" +) + +func newLogsTestServer(t *testing.T, buf *logbuf.Buffer) *httptest.Server { + t.Helper() + + _, server := setupRouter("http://localhost:8001", nil) + server.SetLogBuffer(buf) + + r := chi.NewRouter() + r.Get("/setup/logs", server.HandleGetLogs) + + ts := httptest.NewServer(r) + t.Cleanup(ts.Close) + + return ts +} + +func TestHandleGetLogs_FullSnapshot(t *testing.T) { + buf := logbuf.New(16) + for _, line := range []string{"first\n", "second\n", "third\n"} { + _, _ = buf.Write([]byte(line)) + } + + ts := newLogsTestServer(t, buf) + + res, err := http.Get(ts.URL + "/setup/logs") + if err != nil { + t.Fatalf("GET: %v", err) + } + defer res.Body.Close() + + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d", res.StatusCode) + } + + var resp logsResponse + if err := json.NewDecoder(res.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + + if len(resp.Entries) != 3 { + t.Fatalf("expected 3 entries, got %d", len(resp.Entries)) + } + + if resp.Entries[0].Message != "first" || resp.Entries[2].Message != "third" { + t.Errorf("unexpected ordering: %+v", resp.Entries) + } + + if resp.NextSince != 3 { + t.Errorf("nextSince: want 3, got %d", resp.NextSince) + } + + if resp.Capacity != 16 { + t.Errorf("capacity: want 16, got %d", resp.Capacity) + } +} + +func TestHandleGetLogs_SinceRoundTrip(t *testing.T) { + buf := logbuf.New(16) + for i := 0; i < 5; i++ { + _, _ = buf.Write([]byte("line\n")) + } + + ts := newLogsTestServer(t, buf) + + res, err := http.Get(ts.URL + "/setup/logs?since=2") + if err != nil { + t.Fatalf("GET: %v", err) + } + defer res.Body.Close() + + var resp logsResponse + if err := json.NewDecoder(res.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + + if len(resp.Entries) != 3 { + t.Errorf("expected 3 entries with since=2, got %d", len(resp.Entries)) + } + + if resp.NextSince != 5 { + t.Errorf("nextSince: want 5, got %d", resp.NextSince) + } +} + +func TestHandleGetLogs_Limit(t *testing.T) { + buf := logbuf.New(16) + for i := 0; i < 8; i++ { + _, _ = buf.Write([]byte("line\n")) + } + + ts := newLogsTestServer(t, buf) + + res, err := http.Get(ts.URL + "/setup/logs?limit=3") + if err != nil { + t.Fatalf("GET: %v", err) + } + defer res.Body.Close() + + var resp logsResponse + if err := json.NewDecoder(res.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + + if len(resp.Entries) != 3 { + t.Errorf("limit=3 should cap result, got %d entries", len(resp.Entries)) + } +} + +func TestHandleGetLogs_MalformedSinceReturns400(t *testing.T) { + ts := newLogsTestServer(t, logbuf.New(4)) + + res, err := http.Get(ts.URL + "/setup/logs?since=notanumber") + if err != nil { + t.Fatalf("GET: %v", err) + } + defer res.Body.Close() + + if res.StatusCode != http.StatusBadRequest { + t.Errorf("expected 400, got %d", res.StatusCode) + } +} + +func TestHandleGetLogs_NoBufferAttached(t *testing.T) { + ts := newLogsTestServer(t, nil) + + res, err := http.Get(ts.URL + "/setup/logs") + if err != nil { + t.Fatalf("GET: %v", err) + } + defer res.Body.Close() + + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200 even without buffer, got %d", res.StatusCode) + } + + var resp logsResponse + if err := json.NewDecoder(res.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + + if len(resp.Entries) != 0 { + t.Errorf("expected empty entries when no buffer, got %d", len(resp.Entries)) + } +} + +func TestHandleGetLogs_DroppedReportedWhenLagging(t *testing.T) { + buf := logbuf.New(3) + for i := 0; i < 10; i++ { + _, _ = buf.Write([]byte("line\n")) + } + + ts := newLogsTestServer(t, buf) + + res, err := http.Get(ts.URL + "/setup/logs?since=2") + if err != nil { + t.Fatalf("GET: %v", err) + } + defer res.Body.Close() + + var resp logsResponse + if err := json.NewDecoder(res.Body).Decode(&resp); err != nil { + t.Fatalf("decode: %v", err) + } + + if resp.Dropped == 0 { + t.Errorf("expected dropped > 0 when buffer evicted past `since`, got 0") + } +} diff --git a/pkg/service/handlers/server.go b/pkg/service/handlers/server.go index e28dd87..1f3f65f 100644 --- a/pkg/service/handlers/server.go +++ b/pkg/service/handlers/server.go @@ -21,6 +21,7 @@ import ( "github.com/gesellix/bose-soundtouch/pkg/service/constants" "github.com/gesellix/bose-soundtouch/pkg/service/datastore" "github.com/gesellix/bose-soundtouch/pkg/service/health" + "github.com/gesellix/bose-soundtouch/pkg/service/logbuf" "github.com/gesellix/bose-soundtouch/pkg/service/marge" "github.com/gesellix/bose-soundtouch/pkg/service/proxy" "github.com/gesellix/bose-soundtouch/pkg/service/setup" @@ -64,6 +65,7 @@ type Server struct { amazonService *amazon.Service peerObserver *peerObserver healthRegistry *health.Registry + logBuf *logbuf.Buffer } // RequestSnapshot represents an immutable snapshot of an HTTP request. @@ -147,6 +149,26 @@ func (s *Server) SetVersionInfo(version, commit, date, repoURL string) { s.RepoURL = repoURL } +// SetLogBuffer attaches a logbuf.Buffer to the server. When set, +// HandleGetLogs returns its contents; when nil, the endpoint +// reports an empty snapshot. Optional so that tests and +// alternative composers (the standalone web binary, etc.) don't +// have to construct a buffer they don't need. +func (s *Server) SetLogBuffer(buf *logbuf.Buffer) { + s.mu.Lock() + defer s.mu.Unlock() + + s.logBuf = buf +} + +// LogBuffer returns the attached log buffer, or nil if none. +func (s *Server) LogBuffer() *logbuf.Buffer { + s.mu.RLock() + defer s.mu.RUnlock() + + return s.logBuf +} + // SetDiscoverySettings sets the discovery settings for the server. func (s *Server) SetDiscoverySettings(interval time.Duration, enabled bool) { s.mu.Lock() diff --git a/pkg/service/handlers/web/index.html b/pkg/service/handlers/web/index.html index 02733c2..a967af7 100644 --- a/pkg/service/handlers/web/index.html +++ b/pkg/service/handlers/web/index.html @@ -39,6 +39,9 @@ + @@ -1508,6 +1511,26 @@
Loading…
+ + +
+
+

Service Logs

+
+ + +
+
+

+ Live mirror of the service's stderr log. Stderr still + receives every line — this is a read-only view for + convenience. +

+
Idle.
+

+    
diff --git a/pkg/service/handlers/web/js/script.js b/pkg/service/handlers/web/js/script.js index b55b337..17df669 100644 --- a/pkg/service/handlers/web/js/script.js +++ b/pkg/service/handlers/web/js/script.js @@ -511,6 +511,12 @@ function openTab(evt, tabId) { fetchHealth(); } + if (tabId === "tab-logs") { + startLogsPolling(); + } else { + stopLogsPolling(); + } + if (evt) { evt.currentTarget.className += " active"; let hash = tabId; @@ -4160,3 +4166,136 @@ async function runQuickFix(checkId, fixId, target, confirmMsg, button) { button.disabled = false; } } + +// --------------------------------------------------------------------------- +// Logs tab +// --------------------------------------------------------------------------- + +const logsState = { + timerId: null, + nextSince: 0, + entries: [], + maxEntries: 5000, // client-side cap; UI lag is the bottleneck + droppedTotal: 0, + pollIntervalMs: 1500, + followTail: true, + initialised: false, +}; + +function startLogsPolling() { + initLogsTabOnce(); + // Reset window each time the tab opens so the user gets a + // fresh snapshot rather than picking up stale state. + logsState.entries = []; + logsState.nextSince = 0; + logsState.droppedTotal = 0; + renderLogs(); + pollLogsOnce(); + if (logsState.timerId !== null) clearInterval(logsState.timerId); + logsState.timerId = setInterval(pollLogsOnce, logsState.pollIntervalMs); +} + +function stopLogsPolling() { + if (logsState.timerId !== null) { + clearInterval(logsState.timerId); + logsState.timerId = null; + } +} + +function initLogsTabOnce() { + if (logsState.initialised) return; + logsState.initialised = true; + + const filterEl = document.getElementById("logs-filter"); + if (filterEl) { + filterEl.addEventListener("input", () => renderLogs()); + } + + const followEl = document.getElementById("logs-follow"); + if (followEl) { + followEl.addEventListener("change", () => { + logsState.followTail = followEl.checked; + if (logsState.followTail) scrollLogsToBottom(); + }); + } + + const viewEl = document.getElementById("logs-view"); + if (viewEl) { + // Disengage follow-tail when the user scrolls up; re-engage + // when they're back at the bottom. tail -f muscle memory. + viewEl.addEventListener("scroll", () => { + const distanceFromBottom = viewEl.scrollHeight - viewEl.scrollTop - viewEl.clientHeight; + const atBottom = distanceFromBottom < 8; + if (atBottom !== logsState.followTail) { + logsState.followTail = atBottom; + const followEl = document.getElementById("logs-follow"); + if (followEl) followEl.checked = atBottom; + } + }); + } +} + +async function pollLogsOnce() { + if (typeof document !== "undefined" && document.hidden) return; + + const statusEl = document.getElementById("logs-status"); + try { + const url = `/setup/logs?since=${encodeURIComponent(logsState.nextSince)}`; + const resp = await fetch(url); + if (!resp.ok) throw new Error(`HTTP ${resp.status}`); + const data = await resp.json(); + + if (Array.isArray(data.entries) && data.entries.length > 0) { + logsState.entries.push(...data.entries); + // Trim from the front when we exceed the client cap. + if (logsState.entries.length > logsState.maxEntries) { + logsState.entries.splice(0, logsState.entries.length - logsState.maxEntries); + } + } + + if (typeof data.nextSince === "number") { + logsState.nextSince = data.nextSince; + } + + if (typeof data.dropped === "number" && data.dropped > 0) { + logsState.droppedTotal += data.dropped; + } + + renderLogs(); + if (statusEl) { + const now = new Date().toLocaleTimeString(); + const droppedNote = logsState.droppedTotal > 0 + ? ` · ${logsState.droppedTotal} dropped` + : ""; + statusEl.textContent = `${logsState.entries.length} buffered${droppedNote} · last update ${now}`; + } + } catch (e) { + if (statusEl) statusEl.textContent = `Polling failed: ${e.message || e}`; + } +} + +function renderLogs() { + const viewEl = document.getElementById("logs-view"); + if (!viewEl) return; + + const filterEl = document.getElementById("logs-filter"); + const filter = filterEl ? filterEl.value.trim().toLowerCase() : ""; + + const lines = []; + for (const entry of logsState.entries) { + if (filter && entry.message.toLowerCase().indexOf(filter) === -1) continue; + const ts = entry.time ? entry.time.replace("T", " ").replace("Z", "") : ""; + lines.push(`${ts} ${entry.message}`); + } + + viewEl.textContent = lines.join("\n"); + + if (logsState.followTail) { + scrollLogsToBottom(); + } +} + +function scrollLogsToBottom() { + const viewEl = document.getElementById("logs-view"); + if (viewEl) viewEl.scrollTop = viewEl.scrollHeight; +} diff --git a/pkg/service/logbuf/logbuf.go b/pkg/service/logbuf/logbuf.go new file mode 100644 index 0000000..3855708 --- /dev/null +++ b/pkg/service/logbuf/logbuf.go @@ -0,0 +1,189 @@ +// Package logbuf provides a bounded, in-memory ring buffer for log +// lines. It implements io.Writer so it can be installed as a +// second sink under stdlib log via log.SetOutput(io.MultiWriter( +// os.Stderr, buf)). Each newline-terminated chunk written to the +// buffer becomes one Entry with a monotonic sequence number and a +// capture timestamp; the admin UI polls these entries and renders +// them as a live trace of the running service. +// +// The buffer holds at most Capacity entries; older entries are +// evicted FIFO when new ones arrive. Callers that fall behind can +// detect this via the dropped count returned by Since. +package logbuf + +import ( + "bytes" + "sync" + "time" +) + +// Entry is a single buffered log line. Message excludes the +// trailing newline. Seq is monotonic across the process lifetime +// of the Buffer (it keeps counting up even past evictions, so +// clients can use it as a high-water mark). +type Entry struct { + Seq uint64 `json:"seq"` + Time time.Time `json:"time"` + Message string `json:"message"` +} + +// Buffer is a bounded, concurrent-safe in-memory log ring. +type Buffer struct { + mu sync.Mutex + capacity int + entries []Entry // ring; oldest first + nextSeq uint64 // seq assigned to the next entry written + partial bytes.Buffer + now func() time.Time +} + +// New returns a Buffer that holds up to capacity entries. A +// non-positive capacity is clamped to a small default (16) — a +// zero-capacity buffer would silently drop every line and make +// the feature look broken, which is worse than picking a number. +func New(capacity int) *Buffer { + if capacity <= 0 { + capacity = 16 + } + + return &Buffer{ + capacity: capacity, + entries: make([]Entry, 0, capacity), + nextSeq: 1, + now: time.Now, + } +} + +// Write implements io.Writer. Bytes are split on '\n'; each +// complete line is appended as an Entry. A trailing partial line +// (no terminating newline) is held until the next Write supplies +// the rest. Returns len(p) on success (never reports a short +// write — the buffer always accepts the bytes, even when older +// entries get evicted). +func (b *Buffer) Write(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + + b.mu.Lock() + defer b.mu.Unlock() + + remaining := p + for { + idx := bytes.IndexByte(remaining, '\n') + if idx < 0 { + // No newline in the rest of p — stash it. + b.partial.Write(remaining) + break + } + + // We have a complete line: anything held in `partial` + + // remaining[:idx]. The newline itself is dropped. + var line string + + if b.partial.Len() > 0 { + b.partial.Write(remaining[:idx]) + line = b.partial.String() + + b.partial.Reset() + } else { + line = string(remaining[:idx]) + } + + b.append(line) + + remaining = remaining[idx+1:] + if len(remaining) == 0 { + break + } + } + + return len(p), nil +} + +// append adds one Entry under the lock, evicting the oldest when +// the buffer is at capacity. +func (b *Buffer) append(message string) { + entry := Entry{ + Seq: b.nextSeq, + Time: b.now().UTC(), + Message: message, + } + b.nextSeq++ + + if len(b.entries) < b.capacity { + b.entries = append(b.entries, entry) + return + } + + // At capacity: drop the oldest, slide left, append. A + // circular index would avoid the copy, but at 2k entries + // the copy is negligible and the slice API is simpler. + copy(b.entries, b.entries[1:]) + b.entries[len(b.entries)-1] = entry +} + +// Snapshot returns a copy of every currently buffered entry, in +// order of increasing Seq. +func (b *Buffer) Snapshot() []Entry { + b.mu.Lock() + defer b.mu.Unlock() + + out := make([]Entry, len(b.entries)) + copy(out, b.entries) + + return out +} + +// Since returns all entries with Seq strictly greater than since, +// up to limit. nextSince is the highest Seq among the returned +// entries (or the input `since` when none match), suitable for +// the caller's next poll. dropped is the count of entries the +// caller missed — i.e. entries with Seq > since that have already +// been evicted from the buffer. A limit <= 0 means "no limit". +func (b *Buffer) Since(since uint64, limit int) (entries []Entry, nextSince, dropped uint64) { + b.mu.Lock() + defer b.mu.Unlock() + + nextSince = since + + if len(b.entries) == 0 { + return nil, nextSince, 0 + } + + oldest := b.entries[0].Seq + if since+1 < oldest { + dropped = oldest - (since + 1) + } + + // Binary-walking the ring isn't worth it at 2k entries. + out := make([]Entry, 0, len(b.entries)) + + for i := range b.entries { + if b.entries[i].Seq <= since { + continue + } + + out = append(out, b.entries[i]) + nextSince = b.entries[i].Seq + + if limit > 0 && len(out) >= limit { + break + } + } + + return out, nextSince, dropped +} + +// Capacity returns the configured maximum number of entries. +func (b *Buffer) Capacity() int { + return b.capacity +} + +// Len returns the current count of buffered entries. +func (b *Buffer) Len() int { + b.mu.Lock() + defer b.mu.Unlock() + + return len(b.entries) +} diff --git a/pkg/service/logbuf/logbuf_test.go b/pkg/service/logbuf/logbuf_test.go new file mode 100644 index 0000000..2cff308 --- /dev/null +++ b/pkg/service/logbuf/logbuf_test.go @@ -0,0 +1,264 @@ +package logbuf + +import ( + "fmt" + "strings" + "sync" + "testing" + "time" +) + +func TestBuffer_SingleLineWrite(t *testing.T) { + b := New(8) + + n, err := b.Write([]byte("hello\n")) + if err != nil { + t.Fatalf("Write: %v", err) + } + + if n != len("hello\n") { + t.Errorf("expected %d bytes consumed, got %d", len("hello\n"), n) + } + + got := b.Snapshot() + if len(got) != 1 { + t.Fatalf("expected 1 entry, got %d", len(got)) + } + + if got[0].Message != "hello" { + t.Errorf("expected message 'hello', got %q", got[0].Message) + } + + if got[0].Seq != 1 { + t.Errorf("expected seq=1, got %d", got[0].Seq) + } +} + +func TestBuffer_MultiLineSingleWrite(t *testing.T) { + b := New(8) + _, _ = b.Write([]byte("one\ntwo\nthree\n")) + + got := b.Snapshot() + if len(got) != 3 { + t.Fatalf("expected 3 entries, got %d", len(got)) + } + + want := []string{"one", "two", "three"} + for i := range got { + if got[i].Message != want[i] { + t.Errorf("entry %d: want %q, got %q", i, want[i], got[i].Message) + } + + if got[i].Seq != uint64(i+1) { + t.Errorf("entry %d: want seq=%d, got %d", i, i+1, got[i].Seq) + } + } +} + +func TestBuffer_PartialLineBuffered(t *testing.T) { + b := New(8) + + _, _ = b.Write([]byte("hel")) + _, _ = b.Write([]byte("lo ")) + _, _ = b.Write([]byte("world\n")) + + got := b.Snapshot() + if len(got) != 1 { + t.Fatalf("expected 1 entry, got %d", len(got)) + } + + if got[0].Message != "hello world" { + t.Errorf("expected reassembled line, got %q", got[0].Message) + } +} + +func TestBuffer_PartialLineDoesNotLeakUntilNewline(t *testing.T) { + b := New(8) + _, _ = b.Write([]byte("no newline here")) + + if b.Len() != 0 { + t.Errorf("expected zero entries for an unterminated write, got %d", b.Len()) + } +} + +func TestBuffer_RingEvictionAtCapacity(t *testing.T) { + const capacity = 3 + + b := New(capacity) + for i := 1; i <= 5; i++ { + _, _ = fmt.Fprintf(b, "line %d\n", i) + } + + got := b.Snapshot() + if len(got) != capacity { + t.Fatalf("expected %d entries after eviction, got %d", capacity, len(got)) + } + + // Oldest surviving entry should be "line 3" with Seq=3. + if got[0].Message != "line 3" || got[0].Seq != 3 { + t.Errorf("oldest survivor: want line 3/seq 3, got %q/seq %d", got[0].Message, got[0].Seq) + } + + if got[capacity-1].Message != "line 5" || got[capacity-1].Seq != 5 { + t.Errorf("newest entry: want line 5/seq 5, got %q/seq %d", got[capacity-1].Message, got[capacity-1].Seq) + } +} + +func TestBuffer_SinceFilters(t *testing.T) { + b := New(8) + for i := 1; i <= 5; i++ { + _, _ = fmt.Fprintf(b, "line %d\n", i) + } + + entries, nextSince, dropped := b.Since(2, 0) + if len(entries) != 3 { + t.Fatalf("expected 3 entries since=2, got %d", len(entries)) + } + + if entries[0].Seq != 3 { + t.Errorf("first entry seq: want 3, got %d", entries[0].Seq) + } + + if nextSince != 5 { + t.Errorf("nextSince: want 5, got %d", nextSince) + } + + if dropped != 0 { + t.Errorf("dropped: want 0, got %d", dropped) + } +} + +func TestBuffer_SinceReportsDropped(t *testing.T) { + b := New(3) + for i := 1; i <= 10; i++ { + _, _ = fmt.Fprintf(b, "line %d\n", i) + } + + // Capacity 3 → only seq 8,9,10 remain. Polling with since=2 + // means seq 3..7 (five entries) were evicted before we saw them. + entries, nextSince, dropped := b.Since(2, 0) + if len(entries) != 3 { + t.Fatalf("expected 3 entries, got %d", len(entries)) + } + + if dropped != 5 { + t.Errorf("expected dropped=5, got %d", dropped) + } + + if nextSince != 10 { + t.Errorf("nextSince: want 10, got %d", nextSince) + } +} + +func TestBuffer_SinceRespectsLimit(t *testing.T) { + b := New(10) + for i := 1; i <= 5; i++ { + _, _ = fmt.Fprintf(b, "line %d\n", i) + } + + entries, nextSince, _ := b.Since(0, 2) + if len(entries) != 2 { + t.Fatalf("expected limit=2 to cap result, got %d", len(entries)) + } + + if nextSince != 2 { + t.Errorf("nextSince after limit: want 2, got %d", nextSince) + } +} + +func TestBuffer_SinceNoNewEntries(t *testing.T) { + b := New(4) + _, _ = b.Write([]byte("only\n")) + + entries, nextSince, dropped := b.Since(5, 0) + if len(entries) != 0 { + t.Errorf("expected no entries, got %d", len(entries)) + } + + if nextSince != 5 { + t.Errorf("nextSince should pass through since when no matches: want 5, got %d", nextSince) + } + + if dropped != 0 { + t.Errorf("dropped: want 0, got %d", dropped) + } +} + +func TestBuffer_TimestampMonotonic(t *testing.T) { + b := New(4) + // Inject a deterministic clock so the test isn't time-flaky. + tick := time.Unix(1_700_000_000, 0) + b.now = func() time.Time { + t := tick + tick = tick.Add(time.Millisecond) + + return t + } + + _, _ = b.Write([]byte("a\nb\n")) + + got := b.Snapshot() + if !got[1].Time.After(got[0].Time) { + t.Errorf("expected later entry to have a later timestamp, got %v vs %v", got[1].Time, got[0].Time) + } +} + +func TestBuffer_ConcurrentWrites(t *testing.T) { + b := New(10000) + + const writers = 8 + const perWriter = 500 + + var wg sync.WaitGroup + for w := 0; w < writers; w++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for i := 0; i < perWriter; i++ { + _, _ = fmt.Fprintf(b, "w%d-i%d\n", id, i) + } + }(w) + } + + wg.Wait() + + got := b.Snapshot() + if len(got) != writers*perWriter { + t.Fatalf("expected %d entries, got %d", writers*perWriter, len(got)) + } + + // Seq must be strictly increasing and contiguous from 1. + for i := range got { + if got[i].Seq != uint64(i+1) { + t.Fatalf("seq gap at index %d: got %d", i, got[i].Seq) + } + + if !strings.HasPrefix(got[i].Message, "w") { + t.Errorf("unexpected message: %q", got[i].Message) + } + } +} + +func TestBuffer_ZeroCapacityClamped(t *testing.T) { + b := New(0) + if b.Capacity() == 0 { + t.Errorf("zero capacity should be clamped to a positive default") + } +} + +func TestBuffer_EmptyWriteNoOp(t *testing.T) { + b := New(4) + + n, err := b.Write(nil) + if err != nil { + t.Errorf("nil write: %v", err) + } + + if n != 0 { + t.Errorf("nil write should consume 0 bytes, got %d", n) + } + + if b.Len() != 0 { + t.Errorf("buffer should be empty, got %d entries", b.Len()) + } +}