diff --git a/server/lib/cdpmonitor/README.md b/server/lib/cdpmonitor/README.md index 807cf56b..c7a460d6 100644 --- a/server/lib/cdpmonitor/README.md +++ b/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. diff --git a/server/lib/cdpmonitor/handlers.go b/server/lib/cdpmonitor/handlers.go index 7a672dd8..60bf7adb 100644 --- a/server/lib/cdpmonitor/handlers.go +++ b/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 diff --git a/server/lib/cdpmonitor/monitor.go b/server/lib/cdpmonitor/monitor.go index f66e7809..f8691754 100644 --- a/server/lib/cdpmonitor/monitor.go +++ b/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,20 +386,7 @@ 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) } @@ -405,6 +394,26 @@ func (m *Monitor) sweepPendingRequests(ctx context.Context) { } } +// 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] diff --git a/server/lib/cdpmonitor/network_capture.go b/server/lib/cdpmonitor/network_capture.go index 192b6e7b..ec07890f 100644 --- a/server/lib/cdpmonitor/network_capture.go +++ b/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() diff --git a/server/lib/cdpmonitor/network_limits_test.go b/server/lib/cdpmonitor/network_limits_test.go new file mode 100644 index 00000000..5ec10a87 --- /dev/null +++ b/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"]) + }) +} diff --git a/server/lib/cdpmonitor/telemetry.go b/server/lib/cdpmonitor/telemetry.go index 8a9b20dd..292f4ddc 100644 --- a/server/lib/cdpmonitor/telemetry.go +++ b/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() diff --git a/server/lib/cdpmonitor/types.go b/server/lib/cdpmonitor/types.go index f9905c1a..0c82b601 100644 --- a/server/lib/cdpmonitor/types.go +++ b/server/lib/cdpmonitor/types.go @@ -122,6 +122,7 @@ type networkReqState struct { frameID string navSeq int64 countedNavSeq int + countedState *computedState // accounting owner, retained across asynchronous expiry status int statusText string resHeaders json.RawMessage