fix: drain ring after HTTP shutdown to prevent data loss on exit

Two bugs combined caused silent event loss on shutdown:
1. StorageWriter.Run exited immediately on ctx cancellation with no drain.
2. HTTP servers were shut down after waiting on storageDone, so new events
   could arrive in the ring after Run had already exited.

Fix:
- Add Reader.TryRead for non-blocking ring reads.
- Add StorageWriter.Drain(ctx) which flushes remaining ring contents after
  all publishers have stopped, with a bounded 5s deadline.
- Reorder main.go shutdown: HTTP servers (and all publishers) stop first,
  then Run exits, then Drain flushes the remaining tail, then Close flushes
  the S2 batcher to the network.
- Document at-least-once delivery and dedupe-by-seq contract on StorageWriter.

Author: archandatta

server/cmd/api/main.go

@@ -271,15 +271,9 @@ func main() {
 	<-ctx.Done()
 	slogger.Info("shutdown signal received")
 
-	// Wait for the storage writer to drain from the ring, then flush S2 before
-	// shutting down the HTTP servers and closing the capture session.
-	<-storageDone
-	if storageWriter != nil {
-		if err := storageWriter.Close(); err != nil {
-			slogger.Error("storage writer close failed", "err", err)
-		}
-	}
-
+	// Step 1: shut down all HTTP servers and stop all event publishers (cdpmonitor,
+	// captureSession) before draining the ring. This bounds the set of events that
+	// can arrive after Run exits so Drain sees a stable tail.
 	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer shutdownCancel()
 	g, _ := errgroup.WithContext(shutdownCtx)
@@ -301,6 +295,20 @@ func main() {
 	if err := g.Wait(); err != nil {
 		slogger.Error("server failed to shutdown", "err", err)
 	}
+
+	// Step 2: wait for Run to return (it exits on ctx cancellation), then drain any
+	// events that arrived between the last Read and HTTP shutdown, then flush S2.
+	<-storageDone
+	if storageWriter != nil {
+		drainCtx, drainCancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer drainCancel()
+		if err := storageWriter.Drain(drainCtx); err != nil {
+			slogger.Warn("storage writer drain incomplete", "err", err)
+		}
+		if err := storageWriter.Close(); err != nil {
+			slogger.Error("storage writer close failed", "err", err)
+		}
+	}
 }
 
 func mustFFmpeg() {

server/lib/events/eventsstorage.go

@@ -16,7 +16,13 @@ type Storage interface {
 
 // 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. Call Close after Run returns to flush in-flight writes.
+// 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.
 type StorageWriter struct {
 	reader       *Reader
@@ -36,8 +42,8 @@ func NewStorageWriter(es *EventStream, storage Storage, log *slog.Logger) *Stora
 }
 
 // Run reads from the ring buffer and appends each envelope to storage until
-// ctx is cancelled. Returns ctx.Err() on clean shutdown. Must be called at
-// most once; panics on a second call.
+// ctx is cancelled. Returns the context error on clean shutdown. Must be
+// called at most once; panics on a second call.
 func (w *StorageWriter) Run(ctx context.Context) error {
 	firstCall := false
 	w.once.Do(func() { firstCall = true })
@@ -50,15 +56,44 @@ func (w *StorageWriter) Run(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
-		if res.Dropped > 0 {
-			w.log.Warn("storage writer: dropped events", "count", res.Dropped, "from_seq", res.DroppedFrom, "to_seq", res.DroppedTo)
-			continue
+		if err := w.processResult(ctx, res); err != nil {
+			return err
 		}
-		if err := w.storage.Append(ctx, *res.Envelope); err != nil {
-			total := w.appendErrors.Add(1)
-			w.log.Error("storage writer: append failed", "seq", res.Envelope.Seq, "err", err, "total_append_errors", total)
+	}
+}
+
+// Drain reads any events still in the ring non-blockingly until caught up or
+// ctx expires. Call after all publishers have stopped and Run has returned to
+// ensure no events are silently skipped on shutdown.
+func (w *StorageWriter) Drain(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			w.log.Warn("storage writer: drain deadline exceeded, ring may have unread events")
+			return ctx.Err()
+		default:
+		}
+
+		res, ok := w.reader.TryRead()
+		if !ok {
+			return nil
 		}
+		if err := w.processResult(ctx, res); err != nil {
+			return err
+		}
+	}
+}
+
+func (w *StorageWriter) processResult(ctx context.Context, res ReadResult) error {
+	if res.Dropped > 0 {
+		w.log.Warn("storage writer: dropped events", "count", res.Dropped, "from_seq", res.DroppedFrom, "to_seq", res.DroppedTo)
+		return nil
+	}
+	if err := w.storage.Append(ctx, *res.Envelope); err != nil {
+		total := w.appendErrors.Add(1)
+		w.log.Error("storage writer: append failed", "seq", res.Envelope.Seq, "err", err, "total_append_errors", total)
 	}
+	return nil
 }
 
 // AppendErrors returns the total number of Append failures since Run started.

server/lib/events/eventsstorage_writer_test.go

@@ -171,3 +171,39 @@ func TestStorageWriter_ContextCancelled(t *testing.T) {
 	err := w.Run(ctx)
 	assert.ErrorIs(t, err, context.Canceled)
 }
+
+// TestStorageWriter_DrainFlushesRingAfterRunExits verifies that events
+// published before Drain is called are not lost even after Run has returned.
+func TestStorageWriter_DrainFlushesRingAfterRunExits(t *testing.T) {
+	es := newTestStream(t, 64)
+	backend := &mockBackend{}
+	w := NewStorageWriter(es, backend, slog.Default())
+
+	// Publish events before Run starts so the ring is non-empty.
+	for range 5 {
+		es.Publish(Envelope{Event: Event{Type: "pre"}})
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- w.Run(ctx) }()
+
+	// Let Run consume some events, then cancel.
+	time.Sleep(20 * time.Millisecond)
+
+	// Publish more events that may arrive while Run is winding down.
+	for range 3 {
+		es.Publish(Envelope{Event: Event{Type: "post"}})
+	}
+	cancel()
+	require.ErrorIs(t, <-done, context.Canceled)
+
+	// Drain must flush whatever is left in the ring.
+	drainCtx, drainCancel := context.WithTimeout(context.Background(), time.Second)
+	defer drainCancel()
+	require.NoError(t, w.Drain(drainCtx))
+
+	got := backend.envelopes()
+	// All 8 published events must have been appended across Run + Drain.
+	assert.Len(t, got, 8)
+}

server/lib/events/ringbuffer.go

@@ -83,6 +83,32 @@ type Reader struct {
 	nextSeq uint64
 }
 
+// TryRead returns the next available result without blocking. Returns
+// (result, true) if data is available, (ReadResult{}, false) if the reader
+// has caught up to the latest published seq.
+func (r *Reader) TryRead() (ReadResult, bool) {
+	r.rb.mu.RLock()
+	defer r.rb.mu.RUnlock()
+
+	latest := r.rb.latestSeq
+	oldest := r.rb.oldestSeq()
+
+	if latest == 0 || r.nextSeq > latest {
+		return ReadResult{}, false
+	}
+
+	if r.nextSeq < oldest {
+		from := r.nextSeq
+		dropped := oldest - r.nextSeq
+		r.nextSeq = oldest
+		return ReadResult{Dropped: dropped, DroppedFrom: from, DroppedTo: oldest - 1}, true
+	}
+
+	env := r.rb.buf[r.nextSeq%r.rb.cap]
+	r.nextSeq++
+	return ReadResult{Envelope: &env}, true
+}
+
 // Read blocks until the next envelope is available or ctx is cancelled.
 func (r *Reader) Read(ctx context.Context) (ReadResult, error) {
 	for {