review: update eventsstorage comments and test error handling

Author: archandatta

server/e2e/e2e_s2_storage_test.go

@@ -56,7 +56,7 @@ func TestS2StorageWriter(t *testing.T) {
 
 	checkResp, err := streamClient.CheckTail(ctx)
 	require.NoError(t, err, "check tail before test")
-	startSeq := checkResp.SeqNum
+	startSeq := checkResp.Tail.SeqNum
 
 	// Start a capture session.
 	startResp, err := client.StartCaptureSessionWithResponse(ctx, instanceoapi.StartCaptureSessionJSONRequestBody{})
@@ -81,17 +81,17 @@ func TestS2StorageWriter(t *testing.T) {
 
 	// Read records written after the pre-test tail and verify at least one
 	// envelope is present.
-	readSession, err := streamClient.ReadSession(ctx, &s2.ReadOptions{
+	readCtx, readCancel := context.WithTimeout(ctx, 10*time.Second)
+	defer readCancel()
+
+	readSession, err := streamClient.ReadSession(readCtx, &s2.ReadOptions{
 		SeqNum: s2.Uint64(startSeq),
 	})
 	require.NoError(t, err, "open S2 read session")
 	defer readSession.Close()
 
-	readCtx, readCancel := context.WithTimeout(ctx, 10*time.Second)
-	defer readCancel()
-
 	var count int
-	for readSession.Next(readCtx) {
+	for readSession.Next() {
 		count++
 	}
 	// EOF is expected once we reach the tail — not an error.

server/lib/events/eventsstorage.go

@@ -8,23 +8,16 @@ import (
 	"sync/atomic"
 )
 
-// Storage is the durable storage backend for browser events.
-// Append is called serially from StorageWriter.Run and need not be thread-safe.
 type Storage interface {
 	Append(ctx context.Context, env Envelope) error
 	Close(ctx context.Context) error
 }
 
-// StorageWriter drains the ring buffer and forwards each envelope to the
-// configured Storage backend. Single-use: call Run once; it blocks until ctx
-// is cancelled. After ctx is cancelled, call Drain to flush events that
-// arrived before all publishers stopped, then call Close to flush the backend.
-//
-// Delivery is at-least-once: on process restart the ring is empty so no
-// cross-restart duplicates occur, but consumers should dedupe by env.Seq in
-// case StorageWriter is ever restarted within a process lifetime.
-//
-// Starts from the oldest available event in the ring, not the current tail.
+// StorageWriter reads from the ring buffer and forwards each envelope to
+// Storage. Single-use and not thread-safe: call Run once, then after
+// it returns call Drain followed by Close. Reads start from the oldest
+// available event in the ring, not the current tail. Delivery is
+// at-least-once; consumers should dedupe by env.Seq.
 type StorageWriter struct {
 	reader       *Reader
 	storage      Storage

server/lib/events/eventsstorage_writer_test.go

@@ -70,11 +70,8 @@ func TestStorageWriter_NormalAppend(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		w.Run(ctx) //nolint:errcheck
-	}()
+	done := make(chan error, 1)
+	go func() { done <- w.Run(ctx) }()
 
 	env1 := es.Publish(Envelope{Event: makeEvent("test.one")})
 	env2 := es.Publish(Envelope{Event: makeEvent("test.two")})
@@ -84,7 +81,7 @@ func TestStorageWriter_NormalAppend(t *testing.T) {
 	}, time.Second, 5*time.Millisecond)
 
 	cancel()
-	<-done
+	require.ErrorIs(t, <-done, context.Canceled)
 
 	got := backend.envelopes()
 	assert.Equal(t, env1.Seq, got[0].Seq)
@@ -104,18 +101,15 @@ func TestStorageWriter_DroppedEvents(t *testing.T) {
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		w.Run(ctx) //nolint:errcheck
-	}()
+	done := make(chan error, 1)
+	go func() { done <- w.Run(ctx) }()
 
 	require.Eventually(t, func() bool {
 		return len(backend.envelopes()) > 0
 	}, time.Second, 5*time.Millisecond)
 
 	cancel()
-	<-done
+	require.ErrorIs(t, <-done, context.Canceled)
 
 	// With ring capacity 4 and 8 publishes, the writer must have skipped at
 	// least 4 events via a drop gap — so fewer than 8 envelopes landed.
@@ -132,11 +126,8 @@ func TestStorageWriter_AppendError(t *testing.T) {
 	w := NewStorageWriter(es, backend, slog.Default())
 
 	ctx, cancel := context.WithCancel(context.Background())
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		w.Run(ctx) //nolint:errcheck
-	}()
+	done := make(chan error, 1)
+	go func() { done <- w.Run(ctx) }()
 
 	// Publish an event that will fail. Wait until the writer has attempted it
 	// (errCount > 0), then clear the error and publish a second event. The
@@ -153,7 +144,7 @@ func TestStorageWriter_AppendError(t *testing.T) {
 	}, time.Second, 5*time.Millisecond)
 
 	cancel()
-	<-done
+	require.ErrorIs(t, <-done, context.Canceled)
 
 	got := backend.envelopes()
 	require.Len(t, got, 1)
@@ -198,10 +189,11 @@ func TestStorageWriter_DrainFlushesRingAfterRunExits(t *testing.T) {
 	cancel()
 	require.ErrorIs(t, <-done, context.Canceled)
 
-	// Drain must flush whatever is left in the ring.
+	// Drain then Close mirrors the real shutdown sequence.
 	drainCtx, drainCancel := context.WithTimeout(context.Background(), time.Second)
 	defer drainCancel()
 	require.NoError(t, w.Drain(drainCtx))
+	require.NoError(t, w.Close(drainCtx))
 
 	got := backend.envelopes()
 	// All 8 published events must have been appended across Run + Drain.