devtoolsproxy: count frames queued after the worker stops

Pump reports a disconnect as soon as one direction fails, while the other can
still forward, so a frame can reach the queue after the final drain. Nothing
reads it, and it was not counted either, so cdp_disconnect under-reported loss
for exactly the commands at the end of a session.

Author: archandatta

server/lib/devtoolsproxy/cdpobserver.go

@@ -104,15 +104,26 @@ func (o *cdpObserver) Observe(msg []byte, ts int64) {
 }
 
 // Dropped reports how many forwarded frames the classifier never saw: queue
-// saturation and classification panics. Reported on cdp_disconnect so a reader
-// sees the loss rather than only the VM's log. A saturated queue rejects
-// whatever arrives next, which may be library traffic that would have produced
-// nothing, so this is an upper bound on commands lost rather than a count.
+// saturation, classification panics, and anything still queued once the worker
+// has stopped. Reported on cdp_disconnect so a reader sees the loss rather than
+// only the VM's log. A saturated queue rejects whatever arrives next, which may
+// be library traffic that would have produced nothing, so this is an upper
+// bound on commands lost rather than a count.
 func (o *cdpObserver) Dropped() int64 {
 	if o == nil {
 		return 0
 	}
-	return o.droppedQueued.Load() + o.droppedPanicked.Load()
+	dropped := o.droppedQueued.Load() + o.droppedPanicked.Load()
+	// Pump calls onClose as soon as one direction fails, while the other may
+	// still be forwarding, so a frame can be queued after the final drain. Once
+	// the worker has stopped nothing will read it, which makes it as lost as one
+	// the queue turned away — and silently so, unless it is counted here.
+	select {
+	case <-o.drained:
+		dropped += int64(len(o.frames))
+	default:
+	}
+	return dropped
 }
 
 func (o *cdpObserver) run(ctx context.Context) {

server/lib/devtoolsproxy/cdpobserver_test.go

@@ -215,3 +215,34 @@ func waitFor(t *testing.T, cond func() bool) {
 	}
 	t.Fatal("condition not met within 5s")
 }
+
+// Pump reports a disconnect as soon as one direction fails, while the other can
+// still forward, so a frame can reach the queue after the worker has drained
+// and stopped. Nothing will classify it, so it has to be counted rather than
+// quietly left behind: telemetry_dropped is what tells a reader the tail of the
+// session is incomplete.
+func TestFramesQueuedAfterTeardownAreCountedAsLoss(t *testing.T) {
+	pub := &countingPublisher{}
+	ctx, cancel := context.WithCancel(context.Background())
+	o := newCdpObserver(ctx, pub.publish, controlOn, nil, silentLogger())
+
+	o.Observe([]byte(clickFrame), testForwardTs)
+	cancel()
+	o.WaitDrained(5 * time.Second)
+	if got := pub.n.Load(); got != 1 {
+		t.Fatalf("published %d events before teardown, want 1", got)
+	}
+	if got := o.Dropped(); got != 0 {
+		t.Fatalf("dropped = %d before the late frame, want 0", got)
+	}
+
+	// The straggler the other pump direction forwarded on its way out.
+	o.Observe([]byte(clickFrame), testForwardTs)
+
+	if got := pub.n.Load(); got != 1 {
+		t.Fatalf("published %d events, want 1: the worker has stopped", got)
+	}
+	if got := o.Dropped(); got != 1 {
+		t.Fatalf("dropped = %d, want 1: a frame nothing will read is a loss", got)
+	}
+}