review: redact S2AccessToken in config logs, bound Close with context, return error from StorageWriter.Run double-call

Author: archandatta

server/cmd/api/main.go

@@ -102,7 +102,7 @@ func main() {
 	}
 	captureSession := capturesession.NewCaptureSession(eventStream)
 
-	// Optional S2 durable storage sink.
+	// Optional S2 storage sink.
 	var storageWriter *events.StorageWriter
 	if config.S2Basin != "" && config.S2AccessToken != "" && config.S2Stream != "" {
 		slogger.Info("S2 storage enabled", "basin", config.S2Basin, "stream", config.S2Stream)
@@ -261,7 +261,9 @@ func main() {
 	if storageWriter != nil {
 		go func() {
 			defer close(storageDone)
-			storageWriter.Run(ctx) //nolint:errcheck
+			if err := storageWriter.Run(ctx); err != nil && ctx.Err() == nil {
+				slogger.Error("storage writer failed", "err", err)
+			}
 		}()
 	} else {
 		close(storageDone)
@@ -271,9 +273,6 @@ func main() {
 	<-ctx.Done()
 	slogger.Info("shutdown signal received")
 
-	// 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)
@@ -296,16 +295,15 @@ func main() {
 		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.
+	// wait for Run to return, then drain any events that arrived and 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 {
+		if err := storageWriter.Close(drainCtx); err != nil {
 			slogger.Error("storage writer close failed", "err", err)
 		}
 	}

server/cmd/config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
@@ -42,6 +43,31 @@ type Config struct {
 	S2Stream      string `envconfig:"S2_STREAM"       default:""`
 }
 
+// LogValue implements slog.LogValuer, redacting secret fields.
+func (c *Config) LogValue() slog.Value {
+	s2AccessToken := ""
+	if c.S2AccessToken != "" {
+		s2AccessToken = "[redacted]"
+	}
+	return slog.GroupValue(
+		slog.Int("port", c.Port),
+		slog.Int("frame_rate", c.FrameRate),
+		slog.Int("display_num", c.DisplayNum),
+		slog.Int("max_size_mb", c.MaxSizeInMB),
+		slog.String("output_dir", c.OutputDir),
+		slog.String("ffmpeg_path", c.PathToFFmpeg),
+		slog.Int("devtools_proxy_port", c.DevToolsProxyPort),
+		slog.Bool("log_cdp_messages", c.LogCDPMessages),
+		slog.Duration("scale_to_zero_cooldown", c.ScaleToZeroCooldown),
+		slog.Int("chromedriver_proxy_port", c.ChromeDriverProxyPort),
+		slog.String("chromedriver_upstream_addr", c.ChromeDriverUpstreamAddr),
+		slog.String("devtools_proxy_addr", c.DevToolsProxyAddr),
+		slog.String("s2_basin", c.S2Basin),
+		slog.String("s2_access_token", s2AccessToken),
+		slog.String("s2_stream", c.S2Stream),
+	)
+}
+
 // Load loads configuration from environment variables
 func Load() (*Config, error) {
 	var config Config

server/lib/events/eventsstorage.go

@@ -2,6 +2,7 @@ package events
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"sync"
 	"sync/atomic"
@@ -11,7 +12,7 @@ import (
 // Append is called serially from StorageWriter.Run and need not be thread-safe.
 type Storage interface {
 	Append(ctx context.Context, env Envelope) error
-	Close() error
+	Close(ctx context.Context) error
 }
 
 // StorageWriter drains the ring buffer and forwards each envelope to the
@@ -43,12 +44,12 @@ 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 the context error on clean shutdown. Must be
-// called at most once; panics on a second call.
+// called at most once; returns an error on a second call.
 func (w *StorageWriter) Run(ctx context.Context) error {
 	firstCall := false
 	w.once.Do(func() { firstCall = true })
 	if !firstCall {
-		panic("events: StorageWriter.Run called more than once")
+		return fmt.Errorf("events: StorageWriter.Run called more than once")
 	}
 
 	for {
@@ -96,12 +97,7 @@ func (w *StorageWriter) processResult(ctx context.Context, res ReadResult) error
 	return nil
 }
 
-// AppendErrors returns the total number of Append failures since Run started.
-func (w *StorageWriter) AppendErrors() uint64 {
-	return w.appendErrors.Load()
-}
-
 // Close drains in-flight writes and releases backend resources.
-func (w *StorageWriter) Close() error {
-	return w.storage.Close()
+func (w *StorageWriter) Close(ctx context.Context) error {
+	return w.storage.Close(ctx)
 }

server/lib/events/eventsstorage_writer_test.go

@@ -30,7 +30,7 @@ func (m *mockBackend) Append(_ context.Context, env Envelope) error {
 	return nil
 }
 
-func (m *mockBackend) Close() error { return nil }
+func (m *mockBackend) Close(_ context.Context) error { return nil }
 
 func (m *mockBackend) envelopes() []Envelope {
 	m.mu.Lock()

server/lib/events/s2storage.go

@@ -117,7 +117,7 @@ func (s *S2Storage) Append(_ context.Context, env Envelope) error {
 
 // Close cancels in-flight ack goroutines, waits for them to drain, then closes
 // the producer (which flushes the S2 batcher to the network).
-func (s *S2Storage) Close() error {
+func (s *S2Storage) Close(ctx context.Context) error {
 	close(s.shutdownCh)
-	return s.producer.close(context.Background())
+	return s.producer.close(ctx)
 }