[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

@@ -10,7 +10,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/kernel/kernel-images/server/lib/capturesession"
 	"github.com/kernel/kernel-images/server/lib/cdpmonitor"
 	"github.com/kernel/kernel-images/server/lib/devtoolsproxy"
 	"github.com/kernel/kernel-images/server/lib/events"
@@ -82,8 +81,7 @@ type ApiService struct {
 	xvfbResizeMu sync.Mutex
 
 	// CDP event pipeline and cdpMonitor.
-	eventStream     *events.EventStream
-	captureSession  *capturesession.CaptureSession
+	captureSession  *events.CaptureSession
 	cdpMonitor      cdpMonitorController
 	monitorMu       sync.Mutex
 	lifecycleCtx    context.Context
@@ -92,14 +90,14 @@ type ApiService struct {
 
 var _ oapi.StrictServerInterface = (*ApiService)(nil)
 
+// New constructs an ApiService.
 func New(
 	recordManager recorder.RecordManager,
 	factory recorder.FFmpegRecorderFactory,
 	upstreamMgr *devtoolsproxy.UpstreamManager,
 	stz scaletozero.Controller,
 	nekoAuthClient *nekoclient.AuthClient,
-	captureSession *capturesession.CaptureSession,
-	eventStream    *events.EventStream,
+	captureSession *events.CaptureSession,
 	displayNum int,
 ) (*ApiService, error) {
 	switch {
@@ -113,24 +111,21 @@ func New(
 		return nil, fmt.Errorf("nekoAuthClient cannot be nil")
 	case captureSession == nil:
 		return nil, fmt.Errorf("captureSession cannot be nil")
-	case eventStream == nil:
-		return nil, fmt.Errorf("eventStream cannot be nil")
 	}
 
-	mon := cdpmonitor.New(upstreamMgr, captureSession.Publish, displayNum, slog.Default())
+	mon := cdpmonitor.New(upstreamMgr, func(ev events.Event) { captureSession.Publish(ev) }, displayNum, slog.Default())
 	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),
 		upstreamMgr:       upstreamMgr,
 		stz:               stz,
 		nekoAuthClient:    nekoAuthClient,
 		policy:            &policy.Policy{},
-		eventStream:       eventStream,
 		captureSession:    captureSession,
 		cdpMonitor:        mon,
 		lifecycleCtx:      ctx,
@@ -358,6 +353,7 @@ 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

@@ -11,7 +11,6 @@ import (
 
 	"log/slog"
 
-	"github.com/kernel/kernel-images/server/lib/capturesession"
 	"github.com/kernel/kernel-images/server/lib/devtoolsproxy"
 	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/kernel/kernel-images/server/lib/nekoclient"
@@ -27,7 +26,7 @@ func TestApiService_StartRecording(t *testing.T) {
 
 	t.Run("success", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 
 		resp, err := svc.StartRecording(ctx, oapi.StartRecordingRequestObject{})
@@ -41,7 +40,7 @@ func TestApiService_StartRecording(t *testing.T) {
 
 	t.Run("already recording", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 
 		// First start should succeed
@@ -56,7 +55,7 @@ func TestApiService_StartRecording(t *testing.T) {
 
 	t.Run("custom ids don't collide", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 
 		for i := 0; i < 5; i++ {
@@ -89,7 +88,7 @@ func TestApiService_StopRecording(t *testing.T) {
 
 	t.Run("no active recording", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 
 		resp, err := svc.StopRecording(ctx, oapi.StopRecordingRequestObject{})
@@ -102,7 +101,7 @@ func TestApiService_StopRecording(t *testing.T) {
 		rec := &mockRecorder{id: "default", isRecordingFlag: true}
 		require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 		resp, err := svc.StopRecording(ctx, oapi.StopRecordingRequestObject{})
 		require.NoError(t, err)
@@ -117,7 +116,7 @@ func TestApiService_StopRecording(t *testing.T) {
 
 		force := true
 		req := oapi.StopRecordingRequestObject{Body: &oapi.StopRecordingJSONRequestBody{ForceStop: &force}}
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 		resp, err := svc.StopRecording(ctx, req)
 		require.NoError(t, err)
@@ -131,7 +130,7 @@ func TestApiService_DownloadRecording(t *testing.T) {
 
 	t.Run("not found", func(t *testing.T) {
 		mgr := recorder.NewFFmpegManager()
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
 		require.NoError(t, err)
@@ -151,7 +150,7 @@ func TestApiService_DownloadRecording(t *testing.T) {
 		rec := &mockRecorder{id: "default", isRecordingFlag: true, recordingData: randomBytes(minRecordingSizeInBytes - 1)}
 		require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 		// will return a 202 when the recording is too small
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
@@ -181,7 +180,7 @@ func TestApiService_DownloadRecording(t *testing.T) {
 		rec := &mockRecorder{id: "default", recordingData: data}
 		require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-		svc, err := newSvc(t, mgr)
+		svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 		require.NoError(t, err)
 		resp, err := svc.DownloadRecording(ctx, oapi.DownloadRecordingRequestObject{})
 		require.NoError(t, err)
@@ -201,7 +200,7 @@ func TestApiService_Shutdown(t *testing.T) {
 	rec := &mockRecorder{id: "default", isRecordingFlag: true}
 	require.NoError(t, mgr.RegisterRecorder(ctx, rec), "failed to register recorder")
 
-	svc, err := newSvc(t, mgr)
+	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 	require.NoError(t, err)
 
 	require.NoError(t, svc.Shutdown(ctx))
@@ -305,26 +304,23 @@ func newMockNekoClient(t *testing.T) *nekoclient.AuthClient {
 	return client
 }
 
-func newCaptureSession(t *testing.T) (*capturesession.CaptureSession, *events.EventStream) {
+func newCaptureSession(t *testing.T) *events.CaptureSession {
 	t.Helper()
-	es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 64})
+	cs, err := events.NewCaptureSession(events.CaptureSessionConfig{
+		LogDir:       t.TempDir(),
+		RingCapacity: 64,
+	})
 	if err != nil {
 		t.Fatal(err)
 	}
-	return capturesession.NewCaptureSession(es), es
-}
-
-// newSvc constructs an ApiService with a fresh capture session and event stream.
-func newSvc(t *testing.T, mgr recorder.RecordManager) (*ApiService, error) {
-	t.Helper()
-	cs, es := newCaptureSession(t)
-	return New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), cs, es, 0)
+	t.Cleanup(func() { cs.Close() })
+	return cs
 }
 
 func TestApiService_PatchChromiumFlags(t *testing.T) {
 	ctx := context.Background()
 	mgr := recorder.NewFFmpegManager()
-	svc, err := newSvc(t, mgr)
+	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 	require.NoError(t, err)
 
 	// Test with valid flags

server/cmd/api/api/capture_session.go

@@ -8,7 +8,6 @@ import (
 	"github.com/nrednav/cuid2"
 	oapi "github.com/kernel/kernel-images/server/lib/oapi"
 
-	"github.com/kernel/kernel-images/server/lib/capturesession"
 	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/kernel/kernel-images/server/lib/logger"
 )
@@ -88,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
@@ -120,26 +122,26 @@ func (s *ApiService) buildSessionResponse() oapi.CaptureSession {
 }
 
 // captureConfigFrom converts the optional StartCaptureSessionRequest body
-// into a capturesession.CaptureConfig.
-func captureConfigFrom(body *oapi.StartCaptureSessionRequest) (capturesession.CaptureConfig, error) {
+// into an events.CaptureConfig.
+func captureConfigFrom(body *oapi.StartCaptureSessionRequest) (events.CaptureConfig, error) {
 	if body == nil {
-		return capturesession.CaptureConfig{}, nil
+		return events.CaptureConfig{}, nil
 	}
 	return captureConfigFromOAPI(body.Config)
 }
 
-// captureConfigFromOAPI converts an oapi.CaptureConfig to capturesession.CaptureConfig.
-func captureConfigFromOAPI(cfg *oapi.CaptureConfig) (capturesession.CaptureConfig, error) {
+// captureConfigFromOAPI converts an oapi.CaptureConfig to events.CaptureConfig.
+func captureConfigFromOAPI(cfg *oapi.CaptureConfig) (events.CaptureConfig, error) {
 	if cfg == nil || cfg.Categories == nil {
-		return capturesession.CaptureConfig{}, nil
+		return events.CaptureConfig{}, nil
 	}
-	out := capturesession.CaptureConfig{
+	out := events.CaptureConfig{
 		Categories: make([]events.EventCategory, 0, len(*cfg.Categories)),
 	}
 	for _, c := range *cfg.Categories {
 		cat := events.EventCategory(c)
 		if !events.ValidCategory(cat) {
-			return capturesession.CaptureConfig{}, fmt.Errorf("unknown category: %q", c)
+			return events.CaptureConfig{}, fmt.Errorf("unknown category: %q", c)
 		}
 		out.Categories = append(out.Categories, cat)
 	}

server/cmd/api/api/capture_session_test.go

@@ -244,10 +244,11 @@ 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()
-	cs, es := newCaptureSession(t)
-	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), cs, es, 0)
+	svc, err := New(mgr, newMockFactory(), newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 	require.NoError(t, err)
 	svc.cdpMonitor = &stubCdpMonitor{}
 	return svc
@@ -258,3 +259,4 @@ type stubCdpMonitor struct{}
 func (s *stubCdpMonitor) Start(_ context.Context) error { return nil }
 func (s *stubCdpMonitor) Stop()                         {}
 func (s *stubCdpMonitor) IsRunning() bool               { return false }
+

server/cmd/api/api/display_test.go

@@ -34,8 +34,7 @@ func testFFmpegFactory(t *testing.T, tempDir string) recorder.FFmpegRecorderFact
 
 func newTestServiceWithFactory(t *testing.T, mgr recorder.RecordManager, factory recorder.FFmpegRecorderFactory) *ApiService {
 	t.Helper()
-	cs, es := newCaptureSession(t)
-	svc, err := New(mgr, factory, newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), cs, es, 0)
+	svc, err := New(mgr, factory, newTestUpstreamManager(), scaletozero.NewNoopController(), newMockNekoClient(t), newCaptureSession(t), 0)
 	require.NoError(t, err)
 	return svc
 }