Bound network header capture and isolate expiry accounting

server/lib/cdpmonitor/README.md

@@ -111,10 +111,14 @@ targets, not requests issued before their capture domains finish initializing.
 
 ## Network capture fidelity
 
-Top-level navigation retains requests until their terminal events. Only requests
-belonging to the new loader seed its pending count; completions from the previous
+Top-level navigation retains requests until their terminal events or expiry. The
+collector sweeps once per minute and evicts requests older than five minutes,
+measured from the initial request even across redirects. A completion after
+eviction cannot emit a response. Only requests belonging to the new loader seed
+its pending count; completions from the previous
 navigation cannot make the new navigation idle. Same-process frame requests still
-participate in their page's network-idle accounting.
+participate in their page's network-idle accounting. Expiry updates the capture
+that counted the request, so delayed cleanup cannot make a replacement capture idle.
 
 The collector joins `requestWillBeSentExtraInfo` to requests
 by session, request ID, and redirect hop. Cached hops without ExtraInfo do not
@@ -128,6 +132,16 @@ best-effort when Chrome omits events or a target disconnects. Redirects emit a
 response for each hop, without a body. Request IDs remain the same across a
 redirect chain.
 
+The request-header correlation registry holds at most 8,192 exchanges. If admitting
+a new exchange would exceed that limit, new chains use base request headers and
+omit response `request_headers` until capture restarts (telemetry off/on or CDP
+reconnect). Resuming sooner could attach a later redirect hop's headers to a hop
+whose earlier header event was dropped. Request, response, and failure events
+still publish in order; already admitted chains continue collecting headers.
+Entries are released after terminal publications drain, on expiry, or when capture
+is cleared. This bounds the header registry, not the number of in-flight requests
+or publishers.
+
 Network event timestamps record wall-clock time when the CDP event is observed,
 before header collection, body retrieval, or publication waits. Response events
 use observation of `loadingFinished`, or the redirect notification for redirects.

server/lib/cdpmonitor/handlers.go

@@ -472,6 +472,7 @@ func (m *Monitor) handleNetworkRequest(p cdpNetworkRequestWillBeSentParams, sess
 	} else {
 		navSeq = existing.navSeq
 		countedNavSeq = existing.countedNavSeq
+		cs = existing.countedState
 		m.resolveNetworkHeadersLocked(existing.hop, p.RedirectHasExtraInfo)
 	}
 	m.pendingRequests[key] = networkReqState{
@@ -485,6 +486,7 @@ func (m *Monitor) handleNetworkRequest(p cdpNetworkRequestWillBeSentParams, sess
 		frameID:       p.FrameID,
 		navSeq:        navSeq,
 		countedNavSeq: countedNavSeq,
+		countedState:  cs,
 		addedAt:       addedAt,
 	}
 	m.pendReqMu.Unlock()
@@ -848,6 +850,7 @@ func (m *Monitor) handleFrameNavigated(p cdpPageFrameNavigatedParams, sessionID
 			req.countedNavSeq = -1
 			if cs != nil && req.loaderID == p.Frame.LoaderID {
 				req.countedNavSeq = nextSeq
+				req.countedState = cs
 				inflight++
 			}
 			m.pendingRequests[key] = req

server/lib/cdpmonitor/monitor.go

@@ -68,9 +68,10 @@ type Monitor struct {
 	sessions      map[string]targetInfo // sessionID → targetInfo
 	mainSessionID atomic.Value          // string; set on first top-level frameNavigated, cleared on reconnect
 
-	pendReqMu        sync.Mutex
-	pendingRequests  map[networkRequestKey]networkReqState
-	networkExchanges map[networkRequestKey]*networkExchange
+	pendReqMu               sync.Mutex
+	pendingRequests         map[networkRequestKey]networkReqState
+	networkExchanges        map[networkRequestKey]*networkExchange
+	networkHeadersSaturated bool // pendReqMu; reset with capture
 
 	computedStates map[string]*computedState // sessionID → state machine; guarded by sessionsMu
 
@@ -363,6 +364,7 @@ func (m *Monitor) clearState() {
 	m.pendReqMu.Lock()
 	m.pendingRequests = make(map[networkRequestKey]networkReqState)
 	clear(m.networkExchanges)
+	m.networkHeadersSaturated = false
 	m.pendReqMu.Unlock()
 	m.bindingRateMu.Lock()
 	m.bindingLastSeen = make(map[string]time.Time)
@@ -384,27 +386,34 @@ func (m *Monitor) sweepPendingRequests(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case now := <-ticker.C:
-			var toSweep []networkReqState
-			m.pendReqMu.Lock()
-			for id, state := range m.pendingRequests {
-				if now.Sub(state.addedAt) > pendingRequestTTL {
-					delete(m.pendingRequests, id)
-					toSweep = append(toSweep, state)
-				}
-			}
-			for key, exchange := range m.networkExchanges {
-				if now.Sub(exchange.addedAt) > pendingRequestTTL {
-					delete(m.networkExchanges, key)
-				}
-			}
-			m.pendReqMu.Unlock()
+			toSweep := m.takeExpiredNetworkRequests(now)
 			for _, state := range toSweep {
 				m.finishNetworkRequest(state)
 			}
 		}
 	}
 }
 
+// takeExpiredNetworkRequests transfers accounting to the sweeper before releasing
+// the registry lock. The capture can be replaced before that accounting runs.
+func (m *Monitor) takeExpiredNetworkRequests(now time.Time) []networkReqState {
+	m.pendReqMu.Lock()
+	defer m.pendReqMu.Unlock()
+	var expired []networkReqState
+	for key, state := range m.pendingRequests {
+		if now.Sub(state.addedAt) > pendingRequestTTL {
+			delete(m.pendingRequests, key)
+			expired = append(expired, state)
+		}
+	}
+	for key, exchange := range m.networkExchanges {
+		if now.Sub(exchange.addedAt) > pendingRequestTTL {
+			delete(m.networkExchanges, key)
+		}
+	}
+	return expired
+}
+
 func (m *Monitor) computedFor(sessionID string) *computedState {
 	m.sessionsMu.RLock()
 	cs := m.computedStates[sessionID]

server/lib/cdpmonitor/network_capture.go

@@ -43,6 +43,16 @@ type networkHop struct {
 func (m *Monitor) exchangeLocked(key networkRequestKey) *networkExchange {
 	e := m.networkExchanges[key]
 	if e == nil {
+		if m.networkHeadersSaturated {
+			return nil
+		}
+		if len(m.networkExchanges) >= maxNetworkExchanges {
+			// ExtraInfo may precede its request. Once an event is dropped, new
+			// chains cannot safely resume correlation mid-capture: a later hop's
+			// headers could otherwise be mistaken for the missing first hop.
+			m.networkHeadersSaturated = true
+			return nil
+		}
 		e = &networkExchange{addedAt: time.Now()}
 		m.networkExchanges[key] = e
 	}
@@ -51,6 +61,9 @@ func (m *Monitor) exchangeLocked(key networkRequestKey) *networkExchange {
 
 func (m *Monitor) queueNetworkRequestLocked(key networkRequestKey, request oapi.BrowserNetworkRequestEventData, info targetInfo, observedAt time.Time) *networkHop {
 	e := m.exchangeLocked(key)
+	if e == nil {
+		e = &networkExchange{}
+	}
 	metadata := targetMetadata(info)
 	h := &networkHop{
 		exchange:         e,
@@ -60,7 +73,12 @@ func (m *Monitor) queueNetworkRequestLocked(key networkRequestKey, request oapi.
 		requestReady:     make(chan struct{}),
 		requestPublished: make(chan struct{}),
 	}
-	e.hops = append(e.hops, h)
+	if m.networkExchanges[key] == e {
+		e.hops = append(e.hops, h)
+	} else {
+		// Keep publication ordering even when header enrichment is unavailable.
+		close(h.requestReady)
+	}
 	e.publications++
 	return h
 }
@@ -69,11 +87,8 @@ func (m *Monitor) handleRequestExtraInfo(p cdpNetworkRequestWillBeSentExtraInfoP
 	key := networkRequestKey{sessionID, p.RequestID}
 	m.pendReqMu.Lock()
 	defer m.pendReqMu.Unlock()
-	if m.networkExchanges[key] == nil && len(m.networkExchanges) >= maxNetworkExchanges {
-		return
-	}
 	e := m.exchangeLocked(key)
-	if len(e.requests) >= maxExtraInfoPerExchange {
+	if e == nil || len(e.requests) >= maxExtraInfoPerExchange {
 		return
 	}
 	e.requests = append(e.requests, p.Headers)
@@ -253,7 +268,7 @@ func (m *Monitor) emitNetworkResponse(ctx context.Context, observedAt time.Time,
 }
 
 func (m *Monitor) finishNetworkRequest(state networkReqState) {
-	if cs := m.computedFor(state.sessionID); cs != nil {
+	if cs := state.countedState; cs != nil {
 		m.pendReqMu.Lock()
 		defer m.pendReqMu.Unlock()
 		cs.mu.Lock()

server/lib/cdpmonitor/network_limits_test.go

@@ -0,0 +1,197 @@
+package cdpmonitor
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/kernel/kernel-images/server/lib/oapi"
+	"github.com/stretchr/testify/require"
+)
+
+func TestExpiredRequestCannotMakeNewCaptureIdle(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		m, ec := newComputedMonitor(t)
+		defer m.Stop()
+		navigateMonitor(m, "https://example.com/old")
+		m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: "old"}, "s1")
+		time.Sleep(pendingRequestTTL + time.Second)
+		expired := m.takeExpiredNetworkRequests(time.Now())
+		require.Len(t, expired, 1)
+
+		// Reconnect replaces capture state after expiry removes the request but
+		// before the sweeper applies its pending-count decrement.
+		m.clearState()
+		navigateMonitor(m, "https://example.com/new")
+		m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: "new"}, "s1")
+		m.finishNetworkRequest(expired[0])
+		time.Sleep(networkIdleDebounce + time.Millisecond)
+		synctest.Wait()
+		ec.mu.Lock()
+		idleEvents := 0
+		for _, event := range ec.events {
+			if event.Type == EventNetworkIdle {
+				idleEvents++
+			}
+		}
+		ec.mu.Unlock()
+		require.Zero(t, idleEvents, "an old expiry made the new capture idle")
+
+		m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: "new"}, "s1")
+		time.Sleep(networkIdleDebounce + time.Millisecond)
+		event := ec.waitFor(t, EventNetworkIdle, time.Second)
+		var payload struct {
+			URL string `json:"url"`
+		}
+		require.NoError(t, json.Unmarshal(event.Data, &payload))
+		require.Equal(t, "https://example.com/new", payload.URL)
+	})
+}
+
+func TestNetworkExchangeLimitPreservesCapture(t *testing.T) {
+	for _, terminal := range []string{EventNetworkResponse, EventNetworkLoadingFailed} {
+		t.Run(terminal, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				m, ec := newComputedMonitor(t)
+				defer m.Stop()
+				for i := range maxNetworkExchanges {
+					m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: fmt.Sprint(i)}, "s")
+				}
+				m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: "overflow", Request: cdpNetworkRequest{URL: "https://example.com/start", Headers: json.RawMessage(`{"Base":"start"}`)}}, "s")
+				m.pendReqMu.Lock()
+				size := len(m.networkExchanges)
+				m.pendReqMu.Unlock()
+				require.Equal(t, maxNetworkExchanges, size, "ordinary requests must respect the header registry limit")
+
+				// Free a slot while the overflow chain remains active. Its later
+				// hops must not start consuming headers from the middle of the chain.
+				m.handleResponseReceived(cdpNetworkResponseReceivedParams{RequestID: "0"}, "s")
+				m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: "0"}, "s")
+				time.Sleep(networkHeaderWait + time.Millisecond)
+				synctest.Wait()
+				m.handleRequestExtraInfo(cdpNetworkRequestWillBeSentExtraInfoParams{RequestID: "overflow", Headers: json.RawMessage(`{"Cookie":"wrong-hop"}`)}, "s")
+				m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: "overflow", RedirectHasExtraInfo: true, RedirectResponse: json.RawMessage(`{"status":302}`), Request: cdpNetworkRequest{URL: "https://example.com/end", Headers: json.RawMessage(`{"Base":"end"}`)}}, "s")
+				m.pendReqMu.Lock()
+				_, readmitted := m.networkExchanges[networkRequestKey{"s", "overflow"}]
+				m.pendReqMu.Unlock()
+				require.False(t, readmitted, "an overflow redirect must not restart header correlation mid-chain")
+				if terminal == EventNetworkResponse {
+					m.handleResponseReceived(cdpNetworkResponseReceivedParams{RequestID: "overflow", HasExtraInfo: true, Response: cdpNetworkResponse{Status: 200}}, "s")
+					m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: "overflow"}, "s")
+				} else {
+					m.handleLoadingFailed(cdpNetworkLoadingFailedParams{RequestID: "overflow", ErrorText: "net::ERR_ABORTED", Canceled: true}, "s")
+				}
+				// An admitted chain still receives late headers at capacity.
+				m.handleRequestExtraInfo(cdpNetworkRequestWillBeSentExtraInfoParams{RequestID: "1", Headers: json.RawMessage(`{"Cookie":"retained"}`)}, "s")
+				m.handleResponseReceived(cdpNetworkResponseReceivedParams{RequestID: "1", HasExtraInfo: true}, "s")
+				m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: "1"}, "s")
+				time.Sleep(networkHeaderWait + time.Millisecond)
+				synctest.Wait()
+				ec.mu.Lock()
+				defer ec.mu.Unlock()
+				var sequence []string
+				retainedHeaders := false
+				for _, event := range ec.events {
+					var payload struct {
+						RequestID string `json:"request_id"`
+					}
+					require.NoError(t, json.Unmarshal(event.Data, &payload))
+					if payload.RequestID == "1" && event.Type == EventNetworkResponse {
+						var response oapi.BrowserNetworkResponseEventData
+						require.NoError(t, json.Unmarshal(event.Data, &response))
+						require.NotNil(t, response.RequestHeaders)
+						require.Equal(t, "retained", (*response.RequestHeaders)["Cookie"])
+						retainedHeaders = true
+					}
+					if payload.RequestID != "overflow" {
+						continue
+					}
+					sequence = append(sequence, event.Type)
+					if event.Type == EventNetworkRequest {
+						var request oapi.BrowserNetworkRequestEventData
+						require.NoError(t, json.Unmarshal(event.Data, &request))
+						require.NotNil(t, request.Url)
+						require.Equal(t, strings.TrimPrefix(*request.Url, "https://example.com/"), request.Headers["Base"])
+						require.NotContains(t, request.Headers, "Cookie")
+					}
+					if event.Type == EventNetworkResponse {
+						var response oapi.BrowserNetworkResponseEventData
+						require.NoError(t, json.Unmarshal(event.Data, &response))
+						require.Nil(t, response.RequestHeaders)
+					}
+				}
+				require.Equal(t, []string{EventNetworkRequest, EventNetworkResponse, EventNetworkRequest, terminal}, sequence)
+				require.True(t, retainedHeaders, "capacity fallback affected an admitted chain")
+				require.Less(t, len(m.networkExchanges), maxNetworkExchanges, "terminal publications must release slots")
+			})
+		})
+	}
+}
+
+func TestTelemetryRestartRestoresHeadersAfterSaturation(t *testing.T) {
+	srv := newTestServer(t)
+	defer srv.close()
+	m, ec, cleanup := startMonitor(t, srv, nil)
+	defer cleanup()
+	for i := range maxNetworkExchanges + 1 {
+		m.handleRequestExtraInfo(cdpNetworkRequestWillBeSentExtraInfoParams{RequestID: fmt.Sprint(i), Headers: json.RawMessage(`{}`)}, "")
+	}
+	require.NoError(t, m.SetTelemetry(false))
+	waitForTelemetryReconcile(t, m, false)
+	require.NoError(t, m.SetTelemetry(true))
+	waitForTelemetryReconcile(t, m, true)
+
+	sendNetworkFixture(t, srv, "Network.requestWillBeSentExtraInfo", cdpNetworkRequestWillBeSentExtraInfoParams{RequestID: "fresh", Headers: json.RawMessage(`{"Cookie":"fresh-capture"}`)})
+	sendNetworkFixture(t, srv, "Network.requestWillBeSent", cdpNetworkRequestWillBeSentParams{RequestID: "fresh", Request: cdpNetworkRequest{URL: "https://example.com/fresh"}})
+	sendNetworkFixture(t, srv, "Network.responseReceived", cdpNetworkResponseReceivedParams{RequestID: "fresh", HasExtraInfo: true})
+	sendNetworkFixture(t, srv, "Network.loadingFinished", cdpNetworkLoadingFinishedParams{RequestID: "fresh"})
+	event := ec.waitFor(t, EventNetworkResponse, 2*time.Second)
+	var response oapi.BrowserNetworkResponseEventData
+	require.NoError(t, json.Unmarshal(event.Data, &response))
+	require.NotNil(t, response.RequestHeaders)
+	require.Equal(t, "fresh-capture", (*response.RequestHeaders)["Cookie"])
+}
+
+func TestNetworkExtraInfoOverflowCannotShiftRedirectHeaders(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		m, ec := newComputedMonitor(t)
+		defer m.Stop()
+		for i := range maxNetworkExchanges {
+			m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: fmt.Sprint(i)}, "s")
+		}
+		// The first hop's actual headers arrive while all slots are occupied.
+		m.handleRequestExtraInfo(cdpNetworkRequestWillBeSentExtraInfoParams{RequestID: "late-base", Headers: json.RawMessage(`{"Cookie":"first-hop"}`)}, "s")
+		m.handleResponseReceived(cdpNetworkResponseReceivedParams{RequestID: "0"}, "s")
+		m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: "0"}, "s")
+		time.Sleep(networkHeaderWait + time.Millisecond)
+		synctest.Wait()
+		// Capacity becomes available before the chain's base event is received.
+		m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: "late-base", Request: cdpNetworkRequest{URL: "https://example.com/start", Headers: json.RawMessage(`{"Base":"start"}`)}}, "s")
+		m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: "late-base", RedirectHasExtraInfo: true, RedirectResponse: json.RawMessage(`{"status":302}`), Request: cdpNetworkRequest{URL: "https://example.com/end", Headers: json.RawMessage(`{"Base":"end"}`)}}, "s")
+		m.handleRequestExtraInfo(cdpNetworkRequestWillBeSentExtraInfoParams{RequestID: "late-base", Headers: json.RawMessage(`{"Cookie":"second-hop"}`)}, "s")
+		m.handleResponseReceived(cdpNetworkResponseReceivedParams{RequestID: "late-base", HasExtraInfo: true, Response: cdpNetworkResponse{Status: 200}}, "s")
+		m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: "late-base"}, "s")
+		time.Sleep(networkHeaderWait + time.Millisecond)
+		synctest.Wait()
+		event := waitForNetworkURL(t, ec, EventNetworkResponse, "https://example.com/start")
+		var response oapi.BrowserNetworkResponseEventData
+		require.NoError(t, json.Unmarshal(event.Data, &response))
+		require.Nil(t, response.RequestHeaders, "a rejected ExtraInfo-first chain must not resume correlation from its second hop")
+
+		// A fresh capture can safely enrich new chains again.
+		m.clearState()
+		m.handleRequestExtraInfo(cdpNetworkRequestWillBeSentExtraInfoParams{RequestID: "fresh", Headers: json.RawMessage(`{"Cookie":"fresh-capture"}`)}, "s")
+		m.handleNetworkRequest(cdpNetworkRequestWillBeSentParams{RequestID: "fresh", Request: cdpNetworkRequest{URL: "https://example.com/fresh"}}, "s")
+		m.handleResponseReceived(cdpNetworkResponseReceivedParams{RequestID: "fresh", HasExtraInfo: true}, "s")
+		m.handleLoadingFinished(context.Background(), cdpNetworkLoadingFinishedParams{RequestID: "fresh"}, "s")
+		synctest.Wait()
+		event = waitForNetworkURL(t, ec, EventNetworkResponse, "https://example.com/fresh")
+		require.NoError(t, json.Unmarshal(event.Data, &response))
+		require.NotNil(t, response.RequestHeaders)
+		require.Equal(t, "fresh-capture", (*response.RequestHeaders)["Cookie"])
+	})
+}

server/lib/cdpmonitor/telemetry.go

@@ -95,6 +95,7 @@ func (m *Monitor) applyTelemetry(state uint64) error {
 	m.pendReqMu.Lock()
 	clear(m.pendingRequests)
 	clear(m.networkExchanges)
+	m.networkHeadersSaturated = false
 	m.pendReqMu.Unlock()
 	m.mainSessionID.Store(mainSessionUnset)
 	m.bindingRateMu.Lock()