[kernel-1116] browser events add s2 storage support

images/chromium-headful/run-docker.sh

@@ -71,6 +71,14 @@ if [[ -n "${PLAYWRIGHT_ENGINE:-}" ]]; then
   RUN_ARGS+=( -e PLAYWRIGHT_ENGINE="$PLAYWRIGHT_ENGINE" )
 fi
 
+# S2 durable event storage
+if [[ -n "${S2_BASIN:-}" ]]; then
+  RUN_ARGS+=( -e S2_BASIN="$S2_BASIN" )
+fi
+if [[ -n "${S2_ACCESS_TOKEN:-}" ]]; then
+  RUN_ARGS+=( -e S2_ACCESS_TOKEN="$S2_ACCESS_TOKEN" )
+fi
+
 # WebRTC port mapping
 if [[ "${ENABLE_WEBRTC:-}" == "true" ]]; then
   echo "Running container with WebRTC"

images/chromium-headful/supervisor/services/kernel-images-api.conf

@@ -1,5 +1,5 @@
 [program:kernel-images-api]
-command=/bin/bash -lc 'mkdir -p "${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" && PORT="${KERNEL_IMAGES_API_PORT:-10001}" FRAME_RATE="${KERNEL_IMAGES_API_FRAME_RATE:-10}" DISPLAY_NUM="${KERNEL_IMAGES_API_DISPLAY_NUM:-${DISPLAY_NUM:-1}}" MAX_SIZE_MB="${KERNEL_IMAGES_API_MAX_SIZE_MB:-500}" OUTPUT_DIR="${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" LOG_CDP_MESSAGES="${LOG_CDP_MESSAGES:-false}" exec /usr/local/bin/kernel-images-api'
+command=/bin/bash -lc 'mkdir -p "${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" && PORT="${KERNEL_IMAGES_API_PORT:-10001}" FRAME_RATE="${KERNEL_IMAGES_API_FRAME_RATE:-10}" DISPLAY_NUM="${KERNEL_IMAGES_API_DISPLAY_NUM:-${DISPLAY_NUM:-1}}" MAX_SIZE_MB="${KERNEL_IMAGES_API_MAX_SIZE_MB:-500}" OUTPUT_DIR="${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" LOG_CDP_MESSAGES="${LOG_CDP_MESSAGES:-false}" S2_BASIN="${S2_BASIN:-}" S2_ACCESS_TOKEN="${S2_ACCESS_TOKEN:-}" exec /usr/local/bin/kernel-images-api'
 autostart=false
 autorestart=true
 startsecs=2

images/chromium-headless/image/supervisor/services/kernel-images-api.conf

@@ -1,5 +1,5 @@
 [program:kernel-images-api]
-command=/bin/bash -lc 'mkdir -p "${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" && PORT="${KERNEL_IMAGES_API_PORT:-10001}" FRAME_RATE="${KERNEL_IMAGES_API_FRAME_RATE:-10}" DISPLAY_NUM="${KERNEL_IMAGES_API_DISPLAY_NUM:-${DISPLAY_NUM:-1}}" MAX_SIZE_MB="${KERNEL_IMAGES_API_MAX_SIZE_MB:-500}" OUTPUT_DIR="${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" LOG_CDP_MESSAGES="${LOG_CDP_MESSAGES:-false}" exec /usr/local/bin/kernel-images-api'
+command=/bin/bash -lc 'mkdir -p "${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" && PORT="${KERNEL_IMAGES_API_PORT:-10001}" FRAME_RATE="${KERNEL_IMAGES_API_FRAME_RATE:-10}" DISPLAY_NUM="${KERNEL_IMAGES_API_DISPLAY_NUM:-${DISPLAY_NUM:-1}}" MAX_SIZE_MB="${KERNEL_IMAGES_API_MAX_SIZE_MB:-500}" OUTPUT_DIR="${KERNEL_IMAGES_API_OUTPUT_DIR:-/recordings}" LOG_CDP_MESSAGES="${LOG_CDP_MESSAGES:-false}" S2_BASIN="${S2_BASIN:-}" S2_ACCESS_TOKEN="${S2_ACCESS_TOKEN:-}" exec /usr/local/bin/kernel-images-api'
 autostart=false
 autorestart=true
 startsecs=2

images/chromium-headless/run-docker.sh

@@ -24,6 +24,14 @@ if [[ -n "${PLAYWRIGHT_ENGINE:-}" ]]; then
   RUN_ARGS+=( -e PLAYWRIGHT_ENGINE="$PLAYWRIGHT_ENGINE" )
 fi
 
+# S2 durable event storage
+if [[ -n "${S2_BASIN:-}" ]]; then
+  RUN_ARGS+=( -e S2_BASIN="$S2_BASIN" )
+fi
+if [[ -n "${S2_ACCESS_TOKEN:-}" ]]; then
+  RUN_ARGS+=( -e S2_ACCESS_TOKEN="$S2_ACCESS_TOKEN" )
+fi
+
 # If a positional argument is given, use it as the entrypoint
 ENTRYPOINT_ARG=()
 if [[ $# -ge 1 && -n "$1" ]]; then

server/cmd/api/api/api.go

@@ -90,6 +90,7 @@ type ApiService struct {
 
 var _ oapi.StrictServerInterface = (*ApiService)(nil)
 
+// New constructs an ApiService.
 func New(
 	recordManager recorder.RecordManager,
 	factory recorder.FFmpegRecorderFactory,
@@ -116,8 +117,8 @@ func New(
 	ctx, cancel := context.WithCancel(context.Background())
 
 	return &ApiService{
-		recordManager:   recordManager,
-		factory:         factory,
+		recordManager:     recordManager,
+		factory:           factory,
 		defaultRecorderID: "default",
 		watches:           make(map[string]*fsWatch),
 		procs:             make(map[string]*processHandle),

server/cmd/api/api/capture_session.go

@@ -87,6 +87,9 @@ func (s *ApiService) StopCaptureSession(_ context.Context, _ oapi.StopCaptureSes
 	// tear down asynchronously, leaving IsRunning briefly true.
 	resp := s.buildSessionResponse()
 	resp.Status = oapi.CaptureSessionStatusStopped
+	// Session cleanup (Remove on the S2 producer) happens automatically in
+	// EventsStorageWriter.Run when it processes the SessionEnded event, ensuring
+	// all pending writes are flushed before the producer is torn down.
 	s.captureSession.Stop()
 
 	return oapi.StopCaptureSession200JSONResponse(resp), nil

server/cmd/api/api/capture_session_test.go

@@ -244,6 +244,8 @@ func (m *mockRecordManager) ListActiveRecorders(_ context.Context) []recorder.Re
 func (m *mockRecordManager) StopAll(_ context.Context) error                           { return nil }
 
 // newTestService builds an ApiService with minimal dependencies for capture session tests.
+// The RemoveSession path (triggered by SessionEnded events via the writer's Run loop) is
+// not exercised here — it lives in eventsstorage_writer_test.go.
 func newTestService(t *testing.T, mgr recorder.RecordManager) *ApiService {
 	t.Helper()
 	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)

server/cmd/api/api/events.go

@@ -26,7 +26,7 @@ func (s *ApiService) PublishEvent(_ context.Context, req oapi.PublishEventReques
 	if body == nil || body.Type == "" {
 		return oapi.PublishEvent400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "type is required"}}, nil
 	}
-	if body.Type == events.TypeSessionEnded || body.Type == events.TypeEventsDropped {
+	if body.Type == events.SessionEnded || body.Type == events.EventsDropped || body.Type == events.EventsStorageError {
 		return oapi.PublishEvent400JSONResponse{BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{Message: "type is reserved"}}, nil
 	}
 
@@ -120,7 +120,7 @@ func (s *ApiService) StreamEvents(ctx context.Context, req oapi.StreamEventsRequ
 					Seq:              0,
 					Event: events.Event{
 						Ts:       time.Now().UnixMicro(),
-						Type:     events.TypeEventsDropped,
+						Type:     events.EventsDropped,
 						Category: events.CategorySystem,
 						Source:   events.Source{Kind: events.KindKernelAPI},
 						Data:     json.RawMessage(fmt.Sprintf(`{"dropped":%d}`, result.Dropped)),
@@ -137,7 +137,7 @@ func (s *ApiService) StreamEvents(ctx context.Context, req oapi.StreamEventsRequ
 			if err := writeEnvelopeFrame(pw, &env.Seq, *env); err != nil {
 				return
 			}
-			if env.Event.Type == events.TypeSessionEnded {
+			if env.Event.Type == events.SessionEnded {
 				return
 			}
 		}

server/cmd/api/api/events_test.go

@@ -77,7 +77,7 @@ func TestEventLifecycle(t *testing.T) {
 	// Verify session_ended arrives on the stream.
 	select {
 	case env := <-received:
-		assert.Equal(t, events.TypeSessionEnded, env.Event.Type)
+		assert.Equal(t, events.SessionEnded, env.Event.Type)
 	case <-time.After(2 * time.Second):
 		t.Fatal("timed out waiting for session_ended")
 	}

server/cmd/api/main.go

@@ -101,6 +101,19 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Optionally connect S2 durable storage sink (requires S2_BASIN + S2_TOKEN).
+	var storageWriter *events.EventsStorageWriter
+	if config.S2Basin != "" && config.S2Token != "" {
+		s2backend, err := events.NewS2Storage(config.S2Basin, config.S2Token,
+			config.S2BatcherLingerMs, config.S2BatcherMaxRecs)
+		if err != nil {
+			slogger.Warn("s2 storage unavailable, running without durable event storage", "err", err)
+		} else {
+			storageWriter = events.NewEventsStorageWriter(captureSession, s2backend)
+			slogger.Info("s2 durable event storage enabled", "basin", config.S2Basin)
+		}
+	}
+
 	apiService, err := api.New(
 		recorder.NewFFmpegManager(),
 		recorder.NewFFmpegRecorderFactory(config.PathToFFmpeg, defaultParams, stz),
@@ -242,31 +255,80 @@ func main() {
 		}
 	}()
 
+	// Give the storage writer its own cancellable context so we can stop it
+	// independently of the signal context. This lets us call captureSession.Stop()
+	// (which publishes SessionEnded to the ring) before cancelling the writer,
+	// ensuring the writer can process and flush SessionEnded to S2.
+	writerCtx, writerCancel := context.WithCancel(context.Background())
+	defer writerCancel()
+
+	// Start the S2 storage writer goroutine (no-op if S2 not configured).
+	storageDone := make(chan struct{})
+	if storageWriter != nil {
+		go func() {
+			defer close(storageDone)
+			storageWriter.Run(writerCtx)
+		}()
+	} else {
+		close(storageDone)
+	}
+
 	// graceful shutdown
 	<-ctx.Done()
 	slogger.Info("shutdown signal received")
 
 	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer shutdownCancel()
+
+	// Stop the active capture session now, while the storage writer is still
+	// alive (writerCtx not yet cancelled), so the writer can process the
+	// SessionEnded event and flush it to S2 before tearing down.
+	captureSession.Stop()
+
+	// Cancel the writer so it exits after draining remaining events.
+	writerCancel()
+
+	// Drain storage writer, close it (bounded by shutdownCtx), and shut down
+	// all HTTP servers in parallel so the full 10s budget is available to each.
 	g, _ := errgroup.WithContext(shutdownCtx)
 
+	if storageWriter != nil {
+		g.Go(func() error {
+			<-storageDone
+			closeDone := make(chan error, 1)
+			go func() { closeDone <- storageWriter.Close() }()
+			select {
+			case err := <-closeDone:
+				if err != nil {
+					slogger.Error("storage writer close failed", "err", err)
+				}
+			case <-shutdownCtx.Done():
+				slogger.Error("storage writer close timed out, forcing shutdown")
+			}
+			return nil
+		})
+	}
+
 	g.Go(func() error {
 		return srv.Shutdown(shutdownCtx)
 	})
-	g.Go(func() error {
-		return apiService.Shutdown(shutdownCtx)
-	})
 	g.Go(func() error {
 		upstreamMgr.Stop()
 		return srvDevtools.Shutdown(shutdownCtx)
 	})
 	g.Go(func() error {
 		return srvChromeDriver.Shutdown(shutdownCtx)
 	})
-
 	if err := g.Wait(); err != nil {
 		slogger.Error("server failed to shutdown", "err", err)
 	}
+
+	// Close CaptureSession last, after storage is fully drained.
+	// apiService.Shutdown calls captureSession.Close, which must come after
+	// the storage writer has finished consuming events.
+	if err := apiService.Shutdown(shutdownCtx); err != nil {
+		slogger.Error("api service failed to shutdown", "err", err)
+	}
 }
 
 func mustFFmpeg() {

server/cmd/config/config.go

@@ -35,6 +35,15 @@ type Config struct {
 	// DevTools proxy address passed to ChromeDriver as goog:chromeOptions.debuggerAddress.
 	// If empty, it is derived from DevToolsProxyPort as 127.0.0.1:<port>.
 	DevToolsProxyAddr string `envconfig:"DEVTOOLS_PROXY_ADDR" default:""`
+
+	// S2 durable event storage. Both fields must be non-empty to enable the sink.
+	S2Basin string `envconfig:"S2_BASIN"        default:""`
+	S2Token string `envconfig:"S2_ACCESS_TOKEN" default:""`
+	// S2BatcherLingerMs and S2BatcherMaxRecs control the batcher's flush triggers.
+	// 100ms linger keeps event latency low for near-real-time replay; 50 records
+	// per batch keeps individual S2 append payloads small.
+	S2BatcherLingerMs int `envconfig:"S2_BATCHER_LINGER_MS"  default:"100"`
+	S2BatcherMaxRecs  int `envconfig:"S2_BATCHER_MAX_RECORDS" default:"50"`
 }
 
 // Load loads configuration from environment variables

server/cmd/config/config_test.go

@@ -29,6 +29,8 @@ func TestLoad(t *testing.T) {
 				ChromeDriverProxyPort:    9224,
 				ChromeDriverUpstreamAddr: "127.0.0.1:9225",
 				DevToolsProxyAddr:        "127.0.0.1:9222",
+				S2BatcherLingerMs:        100,
+				S2BatcherMaxRecs:         50,
 			},
 		},
 		{
@@ -57,6 +59,8 @@ func TestLoad(t *testing.T) {
 				ChromeDriverProxyPort:    5432,
 				ChromeDriverUpstreamAddr: "127.0.0.1:9999",
 				DevToolsProxyAddr:        "127.0.0.1:9876",
+				S2BatcherLingerMs:        100,
+				S2BatcherMaxRecs:         50,
 			},
 		},
 		{
@@ -77,6 +81,8 @@ func TestLoad(t *testing.T) {
 				ChromeDriverProxyPort:    9224,
 				ChromeDriverUpstreamAddr: "127.0.0.1:9225",
 				DevToolsProxyAddr:        "10.0.0.1:1234",
+				S2BatcherLingerMs:        100,
+				S2BatcherMaxRecs:         50,
 			},
 		},
 		{

server/go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/m1k1o/neko/server v0.0.0-20251008185748-46e2fc7d3866
 	github.com/nrednav/cuid2 v1.1.0
 	github.com/oapi-codegen/runtime v1.2.0
+	github.com/s2-streamstore/s2-sdk-go v0.14.0
 	github.com/samber/lo v1.52.0
 	github.com/stretchr/testify v1.11.1
 	github.com/testcontainers/testcontainers-go v0.40.0
@@ -99,6 +100,7 @@ require (
 	go.opentelemetry.io/proto/otlp v1.9.0 // indirect
 	golang.org/x/crypto v0.43.0 // indirect
 	golang.org/x/mod v0.28.0 // indirect
+	golang.org/x/net v0.45.0 // indirect
 	golang.org/x/text v0.30.0 // indirect
 	golang.org/x/tools v0.37.0 // indirect
 	google.golang.org/protobuf v1.36.10 // indirect

server/go.sum

@@ -198,6 +198,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
+github.com/s2-streamstore/s2-sdk-go v0.14.0 h1:YqAqUpyeaf6XBA2gPjIyHhVMOnClhFAd8etckUbmSE4=
+github.com/s2-streamstore/s2-sdk-go v0.14.0/go.mod h1:1a+v2sGqU+s5neI8XwqRJz78ktStkR+mZH/JEi9HNSo=
 github.com/samber/lo v1.52.0 h1:Rvi+3BFHES3A8meP33VPAxiBZX/Aws5RxrschYGjomw=
 github.com/samber/lo v1.52.0/go.mod h1:4+MXEGsJzbKGaUEQFKBq2xtfuznW9oz/WrgyzMzRoM0=
 github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=

server/lib/events/capturesession.go

@@ -208,7 +208,7 @@ func (s *CaptureSession) Stop() {
 		return
 	}
 	s.publishLocked(Event{
-		Type:     TypeSessionEnded,
+		Type:     SessionEnded,
 		Category: CategorySystem,
 		Source:   Source{Kind: KindKernelAPI},
 	})

server/lib/events/event.go

@@ -44,8 +44,9 @@ func ValidCategory(c EventCategory) bool {
 
 // System event types emitted by the pipeline itself.
 const (
-	TypeSessionEnded  = "session_ended"
-	TypeEventsDropped = "events_dropped"
+	SessionEnded   = "session_ended"
+	EventsDropped  = "events_dropped"
+	EventsStorageError = "storage_error"
 )
 
 type SourceKind string