review: update monitor to capture ids from cdp to group request data

Author: archandatta

server/lib/cdpmonitor/cdp_test.go

@@ -345,7 +345,15 @@ func newComputedMonitor(t *testing.T) (*Monitor, *eventCollector) {
 }
 
 // navigateMonitor sends a Page.frameNavigated to reset computed state.
+// It ensures session "s1" has a page-like computedState before navigating,
+// mirroring what handleAttachedToTarget would do in production.
 func navigateMonitor(m *Monitor, url string) {
+	m.sessionsMu.Lock()
+	if _, ok := m.sessions["s1"]; !ok {
+		m.sessions["s1"] = targetInfo{targetID: "test-target", targetType: targetTypePage}
+		m.computedStates["s1"] = newComputedState(m.publish)
+	}
+	m.sessionsMu.Unlock()
 	m.handleFrameNavigated(cdpPageFrameNavigatedParams{
 		Frame: cdpPageFrame{ID: "f1", URL: url},
 	}, "s1")
@@ -366,7 +374,7 @@ func simulateRequest(m *Monitor, id string) {
 // simulateFinished stores minimal state and sends Network.loadingFinished.
 func simulateFinished(m *Monitor, id string) {
 	m.pendReqMu.Lock()
-	m.pendingRequests[id] = networkReqState{method: "GET", url: "https://example.com/" + id}
+	m.pendingRequests[id] = networkReqState{sessionID: "s1", method: "GET", url: "https://example.com/" + id}
 	m.pendReqMu.Unlock()
 	m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: id}, "s1")
 }

server/lib/cdpmonitor/computed.go

@@ -2,6 +2,7 @@ package cdpmonitor
 
 import (
 	"encoding/json"
+	"fmt"
 	"sync"
 	"time"
 
@@ -23,11 +24,23 @@ type computedState struct {
 	mu      sync.Mutex
 	publish PublishFunc
 
+	// dead is set by stop(). Timer callbacks check it under mu and bail,
+	// preventing orphaned events after a session is detached or cleared.
+	dead bool
+
 	// navSeq is incremented on every resetOnNavigation. AfterFunc callbacks
 	// capture their navSeq at creation and bail if it has changed, preventing
 	// stale timers from publishing events for a previous navigation.
 	navSeq int
 
+	// navCtx is the navigation identity stamped at the last Page.frameNavigated.
+	// navData and navMeta are its precomputed JSON payload and Source.Metadata.
+	// Maps are replaced (not mutated) on each reset, so in-flight events holding
+	// a pointer to old navMeta are safe.
+	navCtx  navContext
+	navData json.RawMessage
+	navMeta map[string]string
+
 	// network_idle: 500 ms debounce after all pending requests finish.
 	netPending int
 	netTimer   *time.Timer
@@ -47,8 +60,14 @@ type computedState struct {
 }
 
 // newComputedState creates a fresh computedState backed by the given publish func.
+// navData is initialized to {} and navMeta to an empty map so events emitted
+// before the first frameNavigated carry consistent empty payloads rather than null.
 func newComputedState(publish PublishFunc) *computedState {
-	return &computedState{publish: publish}
+	return &computedState{
+		publish: publish,
+		navData: json.RawMessage(`{}`),
+		navMeta: make(map[string]string),
+	}
 }
 
 func stopTimer(t *time.Timer) {
@@ -63,17 +82,45 @@ func stopTimer(t *time.Timer) {
 	}
 }
 
+// stop marks the state machine dead and cancels pending timers. Called when the
+// owning session detaches or the monitor reconnects. Any AfterFunc goroutine
+// already running will check dead under mu and discard its result.
+func (s *computedState) stop() {
+	s.mu.Lock()
+	s.dead = true
+	stopTimer(s.netTimer)
+	stopTimer(s.layoutTimer)
+	s.mu.Unlock()
+}
+
 // resetOnNavigation resets all state machines. Called on Page.frameNavigated.
 // Increments navSeq so any AfterFunc callbacks already running will discard their results.
-// inflight is the number of in-flight requests from other sessions
-// (e.g. subframes) that were not cleared by the navigation; netPending is set
-// to this value instead of zero so that their eventual loadingFinished events
-// decrement correctly.
-func (s *computedState) resetOnNavigation(inflight int) {
+// inflight seeds netPending; callers pass 0 because each session only tracks its
+// own requests and starts fresh on navigation.
+func (s *computedState) resetOnNavigation(inflight int, ctx navContext) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.navSeq++
+	s.navCtx = ctx
+	var err error
+	s.navData, err = json.Marshal(map[string]any{
+		"session_id":  ctx.sessionID,
+		"target_id":   ctx.targetID,
+		"target_type": ctx.targetType,
+		"frame_id":    ctx.frameID,
+		"loader_id":   ctx.loaderID,
+		"url":         ctx.url,
+		"nav_seq":     s.navSeq,
+	})
+	if err != nil {
+		panic(fmt.Sprintf("cdpmonitor: navData marshal: %v", err))
+	}
+	s.navMeta = map[string]string{
+		MetadataKeyCDPSessionID: ctx.sessionID,
+		MetadataKeyTargetID:     ctx.targetID,
+		MetadataKeyTargetType:   ctx.targetType,
+	}
 
 	stopTimer(s.netTimer)
 	s.netTimer = nil
@@ -97,6 +144,9 @@ func (s *computedState) resetOnNavigation(inflight int) {
 func (s *computedState) onRequest() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if s.dead {
+		return
+	}
 	s.netPending++
 	// A new request invalidates any pending network_idle timer
 	stopTimer(s.netTimer)
@@ -107,6 +157,9 @@ func (s *computedState) onRequest() {
 func (s *computedState) onLoadingFinished() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if s.dead {
+		return
+	}
 
 	s.netPending--
 	if s.netPending < 0 {
@@ -123,11 +176,16 @@ func (s *computedState) onLoadingFinished() {
 
 // startNetIdleTimer arms the network_idle debounce timer. Must be called with s.mu held.
 func (s *computedState) startNetIdleTimer() {
+	if s.dead {
+		return
+	}
 	stopTimer(s.netTimer)
 	navSeq := s.navSeq
+	navData := s.navData
+	navMeta := s.navMeta
 	s.netTimer = time.AfterFunc(networkIdleDebounce, func() {
 		s.mu.Lock()
-		if s.navSeq != navSeq || s.netFired || s.netPending > 0 {
+		if s.dead || s.navSeq != navSeq || s.netFired || s.netPending > 0 {
 			s.mu.Unlock()
 			return
 		}
@@ -137,8 +195,11 @@ func (s *computedState) startNetIdleTimer() {
 			Ts:       time.Now().UnixMicro(),
 			Type:     EventNetworkIdle,
 			Category: events.CategoryNetwork,
-			Source:   events.Source{Kind: events.KindCDP},
-			Data:     json.RawMessage(`{}`),
+			Source: events.Source{
+				Kind:     events.KindCDP,
+				Metadata: navMeta,
+			},
+			Data: navData,
 		}}
 		evs = append(evs, s.pendingNavigationSettled()...)
 		s.mu.Unlock()
@@ -152,6 +213,9 @@ func (s *computedState) startNetIdleTimer() {
 func (s *computedState) onPageLoad() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	if s.dead {
+		return
+	}
 	s.pageLoadSeen = true
 	if s.layoutFired {
 		return
@@ -166,7 +230,7 @@ func (s *computedState) onPageLoad() {
 func (s *computedState) onLayoutShift() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if s.layoutFired || !s.pageLoadSeen {
+	if s.dead || s.layoutFired || !s.pageLoadSeen {
 		return
 	}
 	// Reset the timer to 1s from now.
@@ -178,18 +242,23 @@ func (s *computedState) onLayoutShift() {
 // emitLayoutSettled is called from the layout timer's AfterFunc goroutine.
 func (s *computedState) emitLayoutSettled(navSeq int) {
 	s.mu.Lock()
-	if s.navSeq != navSeq || s.layoutFired || !s.pageLoadSeen {
+	if s.dead || s.navSeq != navSeq || s.layoutFired || !s.pageLoadSeen {
 		s.mu.Unlock()
 		return
 	}
 	s.layoutFired = true
 	s.navLayoutSettled = true
+	navData := s.navData
+	navMeta := s.navMeta
 	evs := []events.Event{{
 		Ts:       time.Now().UnixMicro(),
 		Type:     EventLayoutSettled,
 		Category: events.CategoryPage,
-		Source:   events.Source{Kind: events.KindCDP},
-		Data:     json.RawMessage(`{}`),
+		Source: events.Source{
+			Kind:     events.KindCDP,
+			Metadata: navMeta,
+		},
+		Data: navData,
 	}}
 	evs = append(evs, s.pendingNavigationSettled()...)
 	s.mu.Unlock()
@@ -212,14 +281,20 @@ func (s *computedState) onDOMContentLoaded() {
 // pendingNavigationSettled returns a navigation_settled event if all three
 // conditions are met. Must be called with s.mu held.
 func (s *computedState) pendingNavigationSettled() []events.Event {
+	if s.dead {
+		return nil
+	}
 	if s.navDOMLoaded && s.navNetIdle && s.navLayoutSettled && !s.navFired {
 		s.navFired = true
 		return []events.Event{{
 			Ts:       time.Now().UnixMicro(),
 			Type:     EventNavigationSettled,
 			Category: events.CategoryPage,
-			Source:   events.Source{Kind: events.KindCDP},
-			Data:     json.RawMessage(`{}`),
+			Source: events.Source{
+				Kind:     events.KindCDP,
+				Metadata: s.navMeta,
+			},
+			Data: s.navData,
 		}}
 	}
 	return nil

server/lib/cdpmonitor/computed_test.go

@@ -1,11 +1,13 @@
 package cdpmonitor
 
 import (
+	"encoding/json"
 	"testing"
 	"time"
 
 	"github.com/kernel/kernel-images/server/lib/events"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // newTestComputed creates a computedState with an eventCollector for testing.
@@ -19,7 +21,7 @@ func newTestComputed(t *testing.T) (*computedState, *eventCollector) {
 func TestNetworkIdle(t *testing.T) {
 	t.Run("debounce_500ms", func(t *testing.T) {
 		cs, ec := newTestComputed(t)
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 
 		cs.onRequest()
 		cs.onRequest()
@@ -37,7 +39,7 @@ func TestNetworkIdle(t *testing.T) {
 
 	t.Run("timer_reset_on_new_request", func(t *testing.T) {
 		cs, ec := newTestComputed(t)
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 
 		cs.onRequest()
 		cs.onLoadingFinished()
@@ -55,7 +57,7 @@ func TestNetworkIdle(t *testing.T) {
 func TestLayoutSettled(t *testing.T) {
 	t.Run("debounce_1s_after_page_load", func(t *testing.T) {
 		cs, ec := newTestComputed(t)
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 
 		t0 := time.Now()
 		cs.onPageLoad()
@@ -67,7 +69,7 @@ func TestLayoutSettled(t *testing.T) {
 
 	t.Run("layout_shift_before_page_load_ignored", func(t *testing.T) {
 		cs, ec := newTestComputed(t)
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 
 		// layout_shift before page_load should be ignored; layout_settled must
 		// still fire after page_load's 1s debounce.
@@ -81,7 +83,7 @@ func TestLayoutSettled(t *testing.T) {
 
 	t.Run("layout_shift_resets_timer", func(t *testing.T) {
 		cs, ec := newTestComputed(t)
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 		cs.onPageLoad()
 
 		time.Sleep(600 * time.Millisecond)
@@ -96,7 +98,7 @@ func TestLayoutSettled(t *testing.T) {
 func TestNavigationSettled(t *testing.T) {
 	t.Run("fires_when_all_three_flags_set", func(t *testing.T) {
 		cs, ec := newTestComputed(t)
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 
 		cs.onDOMContentLoaded()
 		cs.onRequest()
@@ -109,15 +111,81 @@ func TestNavigationSettled(t *testing.T) {
 
 	t.Run("interrupted_by_new_navigation", func(t *testing.T) {
 		cs, ec := newTestComputed(t)
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 
 		cs.onDOMContentLoaded()
 		cs.onRequest()
 		cs.onLoadingFinished()
 
 		// Interrupt before layout_settled fires.
-		cs.resetOnNavigation(0)
+		cs.resetOnNavigation(0, navContext{})
 
 		ec.assertNone(t, "navigation_settled", 1500*time.Millisecond)
 	})
 }
+
+func TestNavDataMetadata(t *testing.T) {
+	ctx := navContext{
+		sessionID:  "s1",
+		targetID:   "t1",
+		targetType: "page",
+		frameID:    "f1",
+		loaderID:   "l1",
+		url:        "https://example.com",
+	}
+
+	t.Run("layout_settled_carries_navData_and_navMeta", func(t *testing.T) {
+		cs, ec := newTestComputed(t)
+		cs.resetOnNavigation(0, ctx)
+		cs.onPageLoad()
+
+		ev := ec.waitFor(t, "layout_settled", 3*time.Second)
+		assert.Equal(t, events.CategoryPage, ev.Category)
+		assert.Equal(t, "s1", ev.Source.Metadata[MetadataKeyCDPSessionID])
+		assert.Equal(t, "t1", ev.Source.Metadata[MetadataKeyTargetID])
+		assert.Equal(t, "page", ev.Source.Metadata[MetadataKeyTargetType])
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Equal(t, "s1", data["session_id"])
+		assert.Equal(t, "l1", data["loader_id"])
+		assert.Equal(t, "https://example.com", data["url"])
+	})
+
+	t.Run("navigation_settled_carries_navData_and_navMeta", func(t *testing.T) {
+		cs, ec := newTestComputed(t)
+		cs.resetOnNavigation(0, ctx)
+
+		cs.onDOMContentLoaded()
+		cs.onRequest()
+		cs.onLoadingFinished()
+		cs.onPageLoad()
+
+		ev := ec.waitFor(t, "navigation_settled", 3*time.Second)
+		assert.Equal(t, events.CategoryPage, ev.Category)
+		assert.Equal(t, "s1", ev.Source.Metadata[MetadataKeyCDPSessionID])
+		assert.Equal(t, "t1", ev.Source.Metadata[MetadataKeyTargetID])
+		var data map[string]any
+		require.NoError(t, json.Unmarshal(ev.Data, &data))
+		assert.Equal(t, "s1", data["session_id"])
+		assert.Equal(t, "l1", data["loader_id"])
+	})
+}
+
+func TestStopSuppressesTimers(t *testing.T) {
+	t.Run("stop_suppresses_network_idle", func(t *testing.T) {
+		cs, ec := newTestComputed(t)
+		cs.resetOnNavigation(0, navContext{})
+		cs.onRequest()
+		cs.onLoadingFinished() // arms 500ms network_idle timer
+		cs.stop()
+		ec.assertNone(t, "network_idle", 1200*time.Millisecond)
+	})
+
+	t.Run("stop_suppresses_layout_settled", func(t *testing.T) {
+		cs, ec := newTestComputed(t)
+		cs.resetOnNavigation(0, navContext{})
+		cs.onPageLoad() // arms 1s layout_settled timer
+		cs.stop()
+		ec.assertNone(t, "layout_settled", 1500*time.Millisecond)
+	})
+}