review: remove file writer from EventStream

Author: archandatta

server/cmd/api/api/api.go

@@ -352,7 +352,6 @@ func (s *ApiService) Shutdown(ctx context.Context) error {
 	s.lifecycleCancel()
 	s.cdpMonitor.Stop()
 	s.captureSession.Stop()
-	_ = s.captureSession.Close()
 	s.monitorMu.Unlock()
 	return s.recordManager.StopAll(ctx)
 }

server/cmd/api/api/api_test.go

@@ -306,16 +306,11 @@ func newMockNekoClient(t *testing.T) *nekoclient.AuthClient {
 
 func newCaptureSession(t *testing.T) *events.CaptureSession {
 	t.Helper()
-	es, err := events.NewEventStream(events.EventStreamConfig{
-		LogDir:       t.TempDir(),
-		RingCapacity: 64,
-	})
+	es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 64})
 	if err != nil {
 		t.Fatal(err)
 	}
-	cs := events.NewCaptureSession(es)
-	t.Cleanup(func() { cs.Close() })
-	return cs
+	return events.NewCaptureSession(es)
 }
 
 func TestApiService_PatchChromiumFlags(t *testing.T) {

server/cmd/api/main.go

@@ -93,7 +93,6 @@ func main() {
 
 	// Construct events pipeline
 	eventStream, err := events.NewEventStream(events.EventStreamConfig{
-		LogDir:       "/var/log/kernel",
 		RingCapacity: 1024,
 	})
 	if err != nil {

server/lib/events/capturesession.go

@@ -171,7 +171,3 @@ func (s *CaptureSession) Stop() {
 	s.captureSessionID = ""
 }
 
-// Close releases resources held by the EventStream.
-func (s *CaptureSession) Close() error {
-	return s.es.Close()
-}

server/lib/events/events_test.go

@@ -1,11 +1,8 @@
 package events
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
-	"os"
-	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
@@ -330,144 +327,25 @@ func TestConcurrentReaders(t *testing.T) {
 	}
 }
 
-// TestFileWriter: per-category JSONL appender tests.
-func TestFileWriter(t *testing.T) {
-	t.Run("category_routing", func(t *testing.T) {
-		dir := t.TempDir()
-		fw, err := newFileWriter(dir)
-		require.NoError(t, err)
-		defer fw.Close()
-
-		envsToFile := []struct {
-			env      Envelope
-			file     string
-			category string
-		}{
-			{Envelope{Seq: 1, Event: Event{Type: "console.log", Category: CategoryConsole, Source: Source{Kind: KindCDP}, Ts: 1}}, "console.log", "console"},
-			{Envelope{Seq: 2, Event: Event{Type: "network.request", Category: CategoryNetwork, Source: Source{Kind: KindCDP}, Ts: 1}}, "network.log", "network"},
-			{Envelope{Seq: 3, Event: Event{Type: "liveview.click", Category: CategoryLiveview, Source: Source{Kind: KindKernelAPI}, Ts: 1}}, "liveview.log", "liveview"},
-			{Envelope{Seq: 4, Event: Event{Type: "captcha.solve", Category: CategoryCaptcha, Source: Source{Kind: KindExtension}, Ts: 1}}, "captcha.log", "captcha"},
-			{Envelope{Seq: 5, Event: Event{Type: "page.navigation", Category: CategoryPage, Source: Source{Kind: KindCDP}, Ts: 1}}, "page.log", "page"},
-			{Envelope{Seq: 6, Event: Event{Type: "input.click", Category: CategoryInteraction, Source: Source{Kind: KindCDP}, Ts: 1}}, "interaction.log", "interaction"},
-			{Envelope{Seq: 7, Event: Event{Type: "monitor.connected", Category: CategorySystem, Source: Source{Kind: KindKernelAPI}, Ts: 1}}, "system.log", "system"},
-		}
-
-		for _, e := range envsToFile {
-			data, err := json.Marshal(e.env)
-			require.NoError(t, err)
-			require.NoError(t, fw.Write(e.file, data))
-		}
-
-		for _, e := range envsToFile {
-			data, err := os.ReadFile(filepath.Join(dir, e.file))
-			require.NoError(t, err, "missing file %s for type %s", e.file, e.env.Event.Type)
-
-			line := bytes.TrimRight(data, "\n")
-			require.True(t, json.Valid(line), "invalid JSON in %s", e.file)
-
-			var decoded map[string]any
-			require.NoError(t, json.Unmarshal(line, &decoded))
-			inner, ok := decoded["event"].(map[string]any)
-			require.True(t, ok)
-			assert.Equal(t, e.category, inner["category"], "wrong category in %s", e.file)
-			srcMap, ok := inner["source"].(map[string]any)
-			require.True(t, ok, "source should be an object in %s", e.file)
-			assert.Equal(t, string(e.env.Event.Source.Kind), srcMap["kind"], "wrong source kind in %s", e.file)
-		}
-	})
-
-	t.Run("empty_filename_rejected", func(t *testing.T) {
-		dir := t.TempDir()
-		fw, err := newFileWriter(dir)
-		require.NoError(t, err)
-		defer fw.Close()
-
-		err = fw.Write("", []byte(`{"seq":1}`))
-		require.Error(t, err)
-		assert.Contains(t, err.Error(), "empty filename")
-	})
-
-	t.Run("concurrent_writes", func(t *testing.T) {
-		dir := t.TempDir()
-		fw, err := newFileWriter(dir)
-		require.NoError(t, err)
-		defer fw.Close()
-
-		const goroutines = 10
-		const eventsPerGoroutine = 100
-
-		var wg sync.WaitGroup
-		for i := 0; i < goroutines; i++ {
-			wg.Add(1)
-			go func(i int) {
-				defer wg.Done()
-				for j := 0; j < eventsPerGoroutine; j++ {
-					env := Envelope{
-						Seq:   uint64(i*eventsPerGoroutine + j),
-						Event: Event{Type: "console.log", Category: CategoryConsole, Source: Source{Kind: KindCDP}, Ts: 1},
-					}
-					envData, err := json.Marshal(env)
-					require.NoError(t, err)
-					require.NoError(t, fw.Write("console.log", envData))
-				}
-			}(i)
-		}
-		wg.Wait()
-
-		data, err := os.ReadFile(filepath.Join(dir, "console.log"))
-		require.NoError(t, err)
-
-		lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
-		assert.Len(t, lines, goroutines*eventsPerGoroutine)
-		for _, line := range lines {
-			assert.True(t, json.Valid([]byte(line)), "invalid JSON line: %s", line)
-		}
-	})
-
-	t.Run("lazy_open", func(t *testing.T) {
-		dir := t.TempDir()
-		fw, err := newFileWriter(dir)
-		require.NoError(t, err)
-		defer fw.Close()
-
-		entries, err := os.ReadDir(dir)
-		require.NoError(t, err)
-		assert.Empty(t, entries, "files opened before first Write")
-
-		env := Envelope{Seq: 1, Event: Event{Type: "console.log", Category: CategoryConsole, Source: Source{Kind: KindCDP}, Ts: 1}}
-		envData, err := json.Marshal(env)
-		require.NoError(t, err)
-		require.NoError(t, fw.Write("console.log", envData))
-
-		entries, err = os.ReadDir(dir)
-		require.NoError(t, err)
-		assert.Len(t, entries, 1, "expected exactly one file after first Write")
-		assert.Equal(t, "console.log", entries[0].Name())
-	})
-}
-
 func TestCaptureSession(t *testing.T) {
-	newCaptureSession := func(t *testing.T) (*CaptureSession, string) {
+	newCaptureSession := func(t *testing.T) *CaptureSession {
 		t.Helper()
-		dir := t.TempDir()
-		es, err := NewEventStream(EventStreamConfig{LogDir: dir, RingCapacity: 100})
+		es, err := NewEventStream(EventStreamConfig{RingCapacity: 100})
 		require.NoError(t, err)
 		p := NewCaptureSession(es)
 		p.Start("test-session", CaptureConfig{})
-		t.Cleanup(func() { p.Close() })
-		return p, dir
+		return p
 	}
 
 	t.Run("concurrent_publish_seq_order", func(t *testing.T) {
 		const goroutines = 8
 		const eventsEach = 50
 		const total = goroutines * eventsEach
 
-		es, err := NewEventStream(EventStreamConfig{LogDir: t.TempDir(), RingCapacity: total})
+		es, err := NewEventStream(EventStreamConfig{RingCapacity: total})
 		require.NoError(t, err)
 		p := NewCaptureSession(es)
 		p.Start("test-concurrent", CaptureConfig{})
-		t.Cleanup(func() { p.Close() })
 		reader := p.NewReader(0)
 
 		var wg sync.WaitGroup
@@ -492,11 +370,9 @@ func TestCaptureSession(t *testing.T) {
 	})
 
 	t.Run("seq_continues_across_sessions", func(t *testing.T) {
-		es, err := NewEventStream(EventStreamConfig{LogDir: t.TempDir(), RingCapacity: 100})
+		es, err := NewEventStream(EventStreamConfig{RingCapacity: 100})
 		require.NoError(t, err)
 		p := NewCaptureSession(es)
-		t.Cleanup(func() { p.Close() })
-
 		p.Start("session-1", CaptureConfig{})
 		p.Publish(cdpEvent("ev.one", CategorySystem))
 		p.Publish(cdpEvent("ev.two", CategorySystem))
@@ -520,7 +396,7 @@ func TestCaptureSession(t *testing.T) {
 	})
 
 	t.Run("publish_increments_seq", func(t *testing.T) {
-		p, _ := newCaptureSession(t)
+		p := newCaptureSession(t)
 		reader := p.NewReader(0)
 
 		for i := 0; i < 3; i++ {
@@ -537,7 +413,7 @@ func TestCaptureSession(t *testing.T) {
 	})
 
 	t.Run("publish_sets_ts", func(t *testing.T) {
-		p, _ := newCaptureSession(t)
+		p := newCaptureSession(t)
 		reader := p.NewReader(0)
 
 		before := time.Now().UnixMicro()
@@ -552,22 +428,8 @@ func TestCaptureSession(t *testing.T) {
 		assert.LessOrEqual(t, env.Event.Ts, after)
 	})
 
-	t.Run("publish_writes_file", func(t *testing.T) {
-		p, dir := newCaptureSession(t)
-
-		p.Publish(Event{Type: "console.log", Category: CategoryConsole, Source: Source{Kind: KindCDP}, Ts: 1})
-
-		data, err := os.ReadFile(filepath.Join(dir, "console.log"))
-		require.NoError(t, err)
-
-		lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
-		require.Len(t, lines, 1)
-		assert.True(t, json.Valid([]byte(lines[0])))
-		assert.Contains(t, lines[0], `"console.log"`)
-	})
-
 	t.Run("publish_writes_ring", func(t *testing.T) {
-		p, _ := newCaptureSession(t)
+		p := newCaptureSession(t)
 
 		reader := p.NewReader(0)
 		p.Publish(Event{Type: "page.navigation", Category: CategoryPage, Source: Source{Kind: KindCDP}, Ts: 1})
@@ -581,7 +443,7 @@ func TestCaptureSession(t *testing.T) {
 	})
 
 	t.Run("start_sets_capture_session_id", func(t *testing.T) {
-		p, _ := newCaptureSession(t)
+		p := newCaptureSession(t)
 		p.Start("test-uuid", CaptureConfig{})
 
 		reader := p.NewReader(0)
@@ -595,7 +457,7 @@ func TestCaptureSession(t *testing.T) {
 	})
 
 	t.Run("truncation_applied", func(t *testing.T) {
-		p, dir := newCaptureSession(t)
+		p := newCaptureSession(t)
 		reader := p.NewReader(0)
 
 		largeData := strings.Repeat("x", 1_100_000)
@@ -620,12 +482,6 @@ func TestCaptureSession(t *testing.T) {
 		marshaled, err := json.Marshal(env)
 		require.NoError(t, err)
 		assert.LessOrEqual(t, len(marshaled), maxS2RecordBytes)
-
-		data, err := os.ReadFile(filepath.Join(dir, "page.log"))
-		require.NoError(t, err)
-		lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
-		require.Len(t, lines, 1)
-		assert.Contains(t, lines[0], `"truncated":true`)
 	})
 
 }

server/lib/events/eventstream.go

@@ -2,56 +2,42 @@ package events
 
 import (
 	"fmt"
-	"log/slog"
 	"sync"
 )
 
-// EventStream is the process-lifetime event bus. It owns the ring buffer, file
-// writer, and sequence counter.
+// EventStream is the process-lifetime event bus. It owns the ring buffer and
+// sequence counter, which outlive individual capture sessions.
 type EventStream struct {
-	mu    sync.Mutex
-	seq   uint64
-	ring  *ringBuffer
-	files *fileWriter
+	mu   sync.Mutex
+	seq  uint64
+	ring *ringBuffer
 }
 
 // EventStreamConfig holds the parameters for creating an EventStream.
 type EventStreamConfig struct {
-	LogDir string
 	// RingCapacity is the number of envelopes the ring buffer holds.
 	RingCapacity int
 }
 
+// NewEventStream creates an EventStream.
 func NewEventStream(cfg EventStreamConfig) (*EventStream, error) {
 	rb, err := newRingBuffer(cfg.RingCapacity)
 	if err != nil {
 		return nil, fmt.Errorf("event stream: %w", err)
 	}
-	fw, err := newFileWriter(cfg.LogDir)
-	if err != nil {
-		return nil, fmt.Errorf("event stream: %w", err)
-	}
-	return &EventStream{ring: rb, files: fw}, nil
+	return &EventStream{ring: rb}, nil
 }
 
-// publish assigns a monotonically increasing seq to env, writes it to the
-// per-category JSONL file, and pushes it to the ring buffer. Called by
-// CaptureSession under its own lock; env must already have CaptureSessionID set.
+// publish assigns a monotonically increasing seq to env, truncates oversized
+// payloads, and pushes it to the ring buffer. Called by CaptureSession under
+// its own lock; env must already have CaptureSessionID set.
 func (es *EventStream) publish(env Envelope) Envelope {
 	es.mu.Lock()
 	es.seq++
 	env.Seq = es.seq
 	es.mu.Unlock()
 
-	env, data := truncateIfNeeded(env)
-	if data == nil {
-		slog.Error("event_stream: marshal failed, skipping file write", "seq", env.Seq, "category", env.Event.Category)
-	} else {
-		filename := string(env.Event.Category) + ".log"
-		if err := es.files.Write(filename, data); err != nil {
-			slog.Error("event_stream: file write failed", "seq", env.Seq, "category", env.Event.Category, "err", err)
-		}
-	}
+	env, _ = truncateIfNeeded(env)
 	es.ring.publish(env)
 	return env
 }
@@ -68,8 +54,3 @@ func (es *EventStream) Seq() uint64 {
 	defer es.mu.Unlock()
 	return es.seq
 }
-
-// Close flushes and releases all open file descriptors.
-func (es *EventStream) Close() error {
-	return es.files.Close()
-}