feat(stats): log command, routing and state commit latency every 10 s

Each stream that must not wait on another reports how long it did:
- core "outbox stats": direct vs queued commands, merges, backlog, the
  longest queue wait and flush wait (Core.CommandStats, logged by app);
- wayland "command stats": applied commands, ring length, merges, the
  forwarder's longest push block and push-to-apply latency;
- router "routing stats": scenes and cursors replaced unread per output,
  captures refused on a full queue, and the backend loop's longest wait
  sending an output event to core;
- drm stats: state_commits (state committed alone) and state_carried
  (deferred state a frame carried).

The counters allocate nothing; the allocation guards cover them.

commit: 97393eeba7a1d533e972646e9118a42878c6c4cf
author: brice · 4d ago · 2026-10-18 06:01 UTC
parent: c0f82f3
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

18 files · +599 -10

M internal/adapters/drm/flip_stats.go +4 -0


@@ -26,6 +26,10 @@ type flipStats struct {
26	26		// vblank the frame targeted (never under VRR). Fence times are only
27	27		// read while flip tracing is on.
28	28		lateFences int
	29	+	// stateCommits counts state-only commits (cursor, VRR, content type);
	30	+	// stateCarried counts deferred state changes a frame carried instead.
	31	+	// Many state commits while a video plays point at a lost vblank each.
	32	+	stateCommits, stateCarried int
29	33		}
30	34
31	35		// nominalPeriod is the time between two vblanks at the mode's refresh


@@ -26,6 +26,10 @@ type flipStats struct {
26	// vblank the frame targeted (never under VRR). Fence times are only	26		// vblank the frame targeted (never under VRR). Fence times are only
27	// read while flip tracing is on.	27		// read while flip tracing is on.
28	lateFences int	28		lateFences int
		29	+	// stateCommits counts state-only commits (cursor, VRR, content type);
		30	+	// stateCarried counts deferred state changes a frame carried instead.
		31	+	// Many state commits while a video plays point at a lost vblank each.
		32	+	stateCommits, stateCarried int
29	}	33		}
30		34
31	// nominalPeriod is the time between two vblanks at the mode's refresh	35		// nominalPeriod is the time between two vblanks at the mode's refresh

M internal/adapters/drm/output.go +5 -0


@@ -730,6 +730,9 @@ func (o Output) commitWithRect(fb uint32, fence os.File, async bool, vrr bool,
730	730		o.planeRect = rect
731	731		o.lastFrame, o.cursorHeld = time.Now(), false
732	732		// The frame carries every deferred state change.
	733	+	if !o.stateAt.IsZero() {
	734	+	o.flipStats.stateCarried++
	735	+	}
733	736		o.stateAt = time.Time{}
734	737		f.frame, f.async = true, async
735	738		f.wantedAt, o.wantedAt = o.wantedAt, 0

@@ -857,6 +860,7 @@ func (o *Output) commitState(vrr bool) error {
857	860		}
858	861		o.setVRR(vrr)
859	862		o.contentValue = o.contentWanted
	863	+	o.flipStats.stateCommits++
860	864		o.frame.begin(pendingFrame{}, time.Now())
861	865		return nil
862	866		}

@@ -1493,6 +1497,7 @@ func (o *Output) Run(ctx context.Context, newRenderer func(w, h int) (ports.Rend
1493	1497		fs := o.takeFlipStats()
1494	1498		ev = ev.Int("missed_vblanks", fs.missedVblanks).Float64("max_flip_interval_ms", ms(fs.maxInterval))
1495	1499		ev = ev.Float64("max_commit_delay_ms", ms(fs.maxCommitDelay)).Float64("max_flip_to_read_ms", ms(fs.maxFlipToRead))
	1500	+	ev = ev.Int("state_commits", fs.stateCommits).Int("state_carried", fs.stateCarried)
1496	1501		ev = ev.Int("redrawn_pixels", r.TakeRedrawn())
1497	1502		if o.traceFlips {
1498	1503		ev = ev.Int("late_fences", fs.lateFences)


@@ -730,6 +730,9 @@ func (o Output) commitWithRect(fb uint32, fence os.File, async bool, vrr bool,
730	o.planeRect = rect	730		o.planeRect = rect
731	o.lastFrame, o.cursorHeld = time.Now(), false	731		o.lastFrame, o.cursorHeld = time.Now(), false
732	// The frame carries every deferred state change.	732		// The frame carries every deferred state change.
		733	+	if !o.stateAt.IsZero() {
		734	+	o.flipStats.stateCarried++
		735	+	}
733	o.stateAt = time.Time{}	736		o.stateAt = time.Time{}
734	f.frame, f.async = true, async	737		f.frame, f.async = true, async
735	f.wantedAt, o.wantedAt = o.wantedAt, 0	738		f.wantedAt, o.wantedAt = o.wantedAt, 0

@@ -857,6 +860,7 @@ func (o *Output) commitState(vrr bool) error {
857	}	860		}
858	o.setVRR(vrr)	861		o.setVRR(vrr)
859	o.contentValue = o.contentWanted	862		o.contentValue = o.contentWanted
		863	+	o.flipStats.stateCommits++
860	o.frame.begin(pendingFrame{}, time.Now())	864		o.frame.begin(pendingFrame{}, time.Now())
861	return nil	865		return nil
862	}	866		}

@@ -1493,6 +1497,7 @@ func (o *Output) Run(ctx context.Context, newRenderer func(w, h int) (ports.Rend
1493	fs := o.takeFlipStats()	1497		fs := o.takeFlipStats()
1494	ev = ev.Int("missed_vblanks", fs.missedVblanks).Float64("max_flip_interval_ms", ms(fs.maxInterval))	1498		ev = ev.Int("missed_vblanks", fs.missedVblanks).Float64("max_flip_interval_ms", ms(fs.maxInterval))
1495	ev = ev.Float64("max_commit_delay_ms", ms(fs.maxCommitDelay)).Float64("max_flip_to_read_ms", ms(fs.maxFlipToRead))	1499		ev = ev.Float64("max_commit_delay_ms", ms(fs.maxCommitDelay)).Float64("max_flip_to_read_ms", ms(fs.maxFlipToRead))
		1500	+	ev = ev.Int("state_commits", fs.stateCommits).Int("state_carried", fs.stateCarried)
1496	ev = ev.Int("redrawn_pixels", r.TakeRedrawn())	1501		ev = ev.Int("redrawn_pixels", r.TakeRedrawn())
1497	if o.traceFlips {	1502		if o.traceFlips {
1498	ev = ev.Int("late_fences", fs.lateFences)	1503		ev = ev.Int("late_fences", fs.lateFences)

M internal/adapters/drm/state_schedule_test.go +6 -0


@@ -52,6 +52,9 @@ func TestDeferredCursorRidesOnFrame(t *testing.T) {
52	52		if v, _ := (*commits)[0].req.value(tCursor, 14); v != 7 \|\| !o.stateAt.IsZero() {
53	53		t.Fatalf("frame cursor x %d, due %v", v, o.stateAt)
54	54		}
	55	+	if s := o.takeFlipStats(); s.stateCarried != 1 \|\| s.stateCommits != 0 {
	56	+	t.Fatalf("carried %d, state commits %d, want 1, 0", s.stateCarried, s.stateCommits)
	57	+	}
55	58		}
56	59
57	60		// With no frame, the deferred move commits alone once due, and a late

@@ -71,6 +74,9 @@ func TestDeferredCursorCommitsWhenDue(t *testing.T) {
71	74		if v, _ := (*commits)[0].req.value(tCursor, 14); v != 4 \|\| !o.stateAt.IsZero() {
72	75		t.Fatalf("cursor x %d, due %v", v, o.stateAt)
73	76		}
	77	+	if s := o.takeFlipStats(); s.stateCommits != 1 \|\| s.stateCarried != 0 {
	78	+	t.Fatalf("state commits %d, carried %d, want 1, 0", s.stateCommits, s.stateCarried)
	79	+	}
74	80		}
75	81
76	82		// A move undone before its deadline commits nothing and clears the wait.


@@ -52,6 +52,9 @@ func TestDeferredCursorRidesOnFrame(t *testing.T) {
52	if v, _ := (*commits)[0].req.value(tCursor, 14); v != 7 \|\| !o.stateAt.IsZero() {	52		if v, _ := (*commits)[0].req.value(tCursor, 14); v != 7 \|\| !o.stateAt.IsZero() {
53	t.Fatalf("frame cursor x %d, due %v", v, o.stateAt)	53		t.Fatalf("frame cursor x %d, due %v", v, o.stateAt)
54	}	54		}
		55	+	if s := o.takeFlipStats(); s.stateCarried != 1 \|\| s.stateCommits != 0 {
		56	+	t.Fatalf("carried %d, state commits %d, want 1, 0", s.stateCarried, s.stateCommits)
		57	+	}
55	}	58		}
56		59
57	// With no frame, the deferred move commits alone once due, and a late	60		// With no frame, the deferred move commits alone once due, and a late

@@ -71,6 +74,9 @@ func TestDeferredCursorCommitsWhenDue(t *testing.T) {
71	if v, _ := (*commits)[0].req.value(tCursor, 14); v != 4 \|\| !o.stateAt.IsZero() {	74		if v, _ := (*commits)[0].req.value(tCursor, 14); v != 4 \|\| !o.stateAt.IsZero() {
72	t.Fatalf("cursor x %d, due %v", v, o.stateAt)	75		t.Fatalf("cursor x %d, due %v", v, o.stateAt)
73	}	76		}
		77	+	if s := o.takeFlipStats(); s.stateCommits != 1 \|\| s.stateCarried != 0 {
		78	+	t.Fatalf("state commits %d, carried %d, want 1, 0", s.stateCommits, s.stateCarried)
		79	+	}
74	}	80		}
75		81
76	// A move undone before its deadline commits nothing and clears the wait.	82		// A move undone before its deadline commits nothing and clears the wait.

M internal/adapters/wayland/alloc_hotpath_test.go +6 -1


@@ -225,6 +225,7 @@ func TestCommandForwardAllocations(t *testing.T) {
225	225		var wrapped ports.ClientCommand = ports.SecurityCommand{State: state, Command: ports.PointerMotionTo{ID: 1, DX: 1}}
226	226		ran := make(chan struct{}, 1)
227	227		marker := func() { ran <- struct{}{} }
	228	+	var stats commandStats
228	229		cycle := func() {
229	230		r.Push(queue(button))
230	231		r.Push(queue(motion))

@@ -233,11 +234,15 @@ func TestCommandForwardAllocations(t *testing.T) {
233	234		r.Push(queue(wrapped)) // enveloped, same state: folded as well
234	235		s.display.Post(marker)
235	236		<-ran
	237	+	// The latency stats path runs per command and per report.
	238	+	stats.observe(time.Time{}, time.Now())
	239	+	stats.take()
	240	+	r.takeStats()
236	241		}
237	242		for range 4 {
238	243		cycle()
239	244		}
240	245		if allocs := testing.AllocsPerRun(200, cycle); allocs != 0 {
241		-	t.Errorf("command push + merge + drain: %.1f allocs, want 0", allocs)
	246	+	t.Errorf("command push + merge + drain + stats: %.1f allocs, want 0", allocs)
242	247		}
243	248		}

 	var wrapped ports.ClientCommand = ports.SecurityCommand{State: state, Command: ports.PointerMotionTo{ID: 1, DX: 1}}
 	ran := make(chan struct{}, 1)
 	marker := func() { ran <- struct{}{} }
+	var stats commandStats
 	cycle := func() {
 		r.Push(queue(button))
 		r.Push(queue(motion))
 		r.Push(queue(wrapped)) // enveloped, same state: folded as well
 		s.display.Post(marker)
 		<-ran
+		// The latency stats path runs per command and per report.
+		stats.observe(time.Time{}, time.Now())
+		stats.take()
+		r.takeStats()
 	}
 	for range 4 {
 		cycle()
 	}
 	if allocs := testing.AllocsPerRun(200, cycle); allocs != 0 {
-		t.Errorf("command push + merge + drain: %.1f allocs, want 0", allocs)
+		t.Errorf("command push + merge + drain + stats: %.1f allocs, want 0", allocs)
 	}
 }

M internal/adapters/wayland/commandring.go +70 -2

@@ -1,6 +1,10 @@
1	1		package wayland
2	2
3	3		import (
	4	+	"context"
	5	+	"sync/atomic"
	6	+	"time"
	7	+
4	8		"github.com/bnema/neferwl/internal/ports"
5	9		"github.com/bnema/purego-libwayland/server"
6	10		)

@@ -26,10 +30,13 @@ type queuedCommand struct {
26	30		isMotion, enveloped bool
27	31		// merged: motion holds the sum of several motions and cmd is stale.
28	32		merged bool
	33	+	// at is when the command entered the ring path. A merge keeps the oldest
	34	+	// value, so a folded motion reports the wait of the first motion in it.
	35	+	at time.Time
29	36		}
30	37
31	38		func queue(cmd ports.ClientCommand) queuedCommand {
32		-	q := queuedCommand{cmd: cmd}
	39	+	q := queuedCommand{cmd: cmd, at: time.Now()}
33	40		if env, ok := cmd.(ports.SecurityCommand); ok {
34	41		q.state, q.enveloped = env.State, true
35	42		cmd = env.Command

@@ -53,7 +60,7 @@ func (q *queuedCommand) command() ports.ClientCommand {
53	60		// last is the most recent queued command, so no other command is ever
54	61		// overtaken and a bare motion never takes the place of an enveloped one. The
55	62		// position is the latest; relative deltas add up so locked pointers (games)
56		-	// lose no movement.
	63	+	// lose no movement. last.at stays: the merge reports its first motion's wait.
57	64		func mergeMotion(last *queuedCommand, v queuedCommand) bool {
58	65		if !last.isMotion \|\| !v.isMotion \|\| last.motion.ID != v.motion.ID \|\|
59	66		last.enveloped != v.enveloped \|\| last.state != v.state {

@@ -77,11 +84,72 @@ func newCommandRing(d server.Display, apply func(queuedCommand)) postRing[queu
77	84		return newMergingPostRing(d, commandRingSize, apply, mergeMotion)
78	85		}
79	86
	87	+	// commandStats counts what the display goroutine applied. The display
	88	+	// goroutine writes and the stats logger takes, so the fields are atomic.
	89	+	type commandStats struct {
	90	+	applied atomic.Int64
	91	+	maxLatency atomic.Int64 // nanoseconds from push to apply
	92	+	}
	93	+
	94	+	// observe records one applied command pushed at at.
	95	+	func (c *commandStats) observe(at time.Time, now time.Time) {
	96	+	c.applied.Add(1)
	97	+	d := int64(now.Sub(at))
	98	+	for {
	99	+	old := c.maxLatency.Load()
	100	+	if d <= old \|\| c.maxLatency.CompareAndSwap(old, d) {
	101	+	return
	102	+	}
	103	+	}
	104	+	}
	105	+
	106	+	// take returns the applied count and the largest latency, and resets both.
	107	+	func (c *commandStats) take() (applied int64, maxLatency time.Duration) {
	108	+	return c.applied.Swap(0), time.Duration(c.maxLatency.Swap(0))
	109	+	}
	110	+
	111	+	// commandStatsInterval is how often logCommandStats reports.
	112	+	const commandStatsInterval = 10 * time.Second
	113	+
	114	+	// logCommandStats reports command path latency every interval until ctx or the
	115	+	// display ends, so a live log shows whether the display goroutine delays input.
	116	+	// A period without applied commands logs nothing.
	117	+	func (s *Server) logCommandStats(ctx context.Context) {
	118	+	t := time.NewTicker(commandStatsInterval)
	119	+	defer t.Stop()
	120	+	for {
	121	+	select {
	122	+	case <-ctx.Done():
	123	+	return
	124	+	case <-s.display.Stopped():
	125	+	return
	126	+	case <-t.C:
	127	+	s.reportCommandStats()
	128	+	}
	129	+	}
	130	+	}
	131	+
	132	+	func (s *Server) reportCommandStats() {
	133	+	rs := s.commands.takeStats()
	134	+	applied, latency := s.commandStats.take()
	135	+	if applied == 0 {
	136	+	return
	137	+	}
	138	+	s.log.Info().
	139	+	Int64("commands", applied).
	140	+	Int("merged", rs.merged).
	141	+	Int("max_queued", rs.maxLen).
	142	+	Float64("max_push_block_ms", float64(rs.maxPushBlock)/float64(time.Millisecond)).
	143	+	Float64("max_latency_ms", float64(latency)/float64(time.Millisecond)).
	144	+	Msg("command stats")
	145	+	}
	146	+
80	147		// applyQueued runs on the display goroutine; it stops applying once the
81	148		// server context is done, like the per-command check of the old batch.
82	149		func (s *Server) applyQueued(q queuedCommand) {
83	150		if s.ctx != nil && s.ctx.Err() != nil {
84	151		return
85	152		}
	153	+	s.commandStats.observe(q.at, time.Now())
86	154		s.apply(q.command())
87	155		}

@@ -1,6 +1,10 @@
1		package wayland	1		package wayland
2			2
3		import (	3		import (
			4	+	"context"
			5	+	"sync/atomic"
			6	+	"time"
			7	+
4		"github.com/bnema/neferwl/internal/ports"	8		"github.com/bnema/neferwl/internal/ports"
5		"github.com/bnema/purego-libwayland/server"	9		"github.com/bnema/purego-libwayland/server"
6		)	10		)

@@ -26,10 +30,13 @@ type queuedCommand struct {
26		isMotion, enveloped bool	30		isMotion, enveloped bool
27		// merged: motion holds the sum of several motions and cmd is stale.	31		// merged: motion holds the sum of several motions and cmd is stale.
28		merged bool	32		merged bool
			33	+	// at is when the command entered the ring path. A merge keeps the oldest
			34	+	// value, so a folded motion reports the wait of the first motion in it.
			35	+	at time.Time
29		}	36		}
30			37
31		func queue(cmd ports.ClientCommand) queuedCommand {	38		func queue(cmd ports.ClientCommand) queuedCommand {
32	-	q := queuedCommand{cmd: cmd}	39	+	q := queuedCommand{cmd: cmd, at: time.Now()}
33		if env, ok := cmd.(ports.SecurityCommand); ok {	40		if env, ok := cmd.(ports.SecurityCommand); ok {
34		q.state, q.enveloped = env.State, true	41		q.state, q.enveloped = env.State, true
35		cmd = env.Command	42		cmd = env.Command

@@ -53,7 +60,7 @@ func (q *queuedCommand) command() ports.ClientCommand {
53		// last is the most recent queued command, so no other command is ever	60		// last is the most recent queued command, so no other command is ever
54		// overtaken and a bare motion never takes the place of an enveloped one. The	61		// overtaken and a bare motion never takes the place of an enveloped one. The
55		// position is the latest; relative deltas add up so locked pointers (games)	62		// position is the latest; relative deltas add up so locked pointers (games)
56	-	// lose no movement.	63	+	// lose no movement. last.at stays: the merge reports its first motion's wait.
57		func mergeMotion(last *queuedCommand, v queuedCommand) bool {	64		func mergeMotion(last *queuedCommand, v queuedCommand) bool {
58		if !last.isMotion \|\| !v.isMotion \|\| last.motion.ID != v.motion.ID \|\|	65		if !last.isMotion \|\| !v.isMotion \|\| last.motion.ID != v.motion.ID \|\|
59		last.enveloped != v.enveloped \|\| last.state != v.state {	66		last.enveloped != v.enveloped \|\| last.state != v.state {

@@ -77,11 +84,72 @@ func newCommandRing(d server.Display, apply func(queuedCommand)) postRing[queu
77		return newMergingPostRing(d, commandRingSize, apply, mergeMotion)	84		return newMergingPostRing(d, commandRingSize, apply, mergeMotion)
78		}	85		}
79			86
			87	+	// commandStats counts what the display goroutine applied. The display
			88	+	// goroutine writes and the stats logger takes, so the fields are atomic.
			89	+	type commandStats struct {
			90	+	applied atomic.Int64
			91	+	maxLatency atomic.Int64 // nanoseconds from push to apply
			92	+	}
			93	+
			94	+	// observe records one applied command pushed at at.
			95	+	func (c *commandStats) observe(at time.Time, now time.Time) {
			96	+	c.applied.Add(1)
			97	+	d := int64(now.Sub(at))
			98	+	for {
			99	+	old := c.maxLatency.Load()
			100	+	if d <= old \|\| c.maxLatency.CompareAndSwap(old, d) {
			101	+	return
			102	+	}
			103	+	}
			104	+	}
			105	+
			106	+	// take returns the applied count and the largest latency, and resets both.
			107	+	func (c *commandStats) take() (applied int64, maxLatency time.Duration) {
			108	+	return c.applied.Swap(0), time.Duration(c.maxLatency.Swap(0))
			109	+	}
			110	+
			111	+	// commandStatsInterval is how often logCommandStats reports.
			112	+	const commandStatsInterval = 10 * time.Second
			113	+
			114	+	// logCommandStats reports command path latency every interval until ctx or the
			115	+	// display ends, so a live log shows whether the display goroutine delays input.
			116	+	// A period without applied commands logs nothing.
			117	+	func (s *Server) logCommandStats(ctx context.Context) {
			118	+	t := time.NewTicker(commandStatsInterval)
			119	+	defer t.Stop()
			120	+	for {
			121	+	select {
			122	+	case <-ctx.Done():
			123	+	return
			124	+	case <-s.display.Stopped():
			125	+	return
			126	+	case <-t.C:
			127	+	s.reportCommandStats()
			128	+	}
			129	+	}
			130	+	}
			131	+
			132	+	func (s *Server) reportCommandStats() {
			133	+	rs := s.commands.takeStats()
			134	+	applied, latency := s.commandStats.take()
			135	+	if applied == 0 {
			136	+	return
			137	+	}
			138	+	s.log.Info().
			139	+	Int64("commands", applied).
			140	+	Int("merged", rs.merged).
			141	+	Int("max_queued", rs.maxLen).
			142	+	Float64("max_push_block_ms", float64(rs.maxPushBlock)/float64(time.Millisecond)).
			143	+	Float64("max_latency_ms", float64(latency)/float64(time.Millisecond)).
			144	+	Msg("command stats")
			145	+	}
			146	+
80		// applyQueued runs on the display goroutine; it stops applying once the	147		// applyQueued runs on the display goroutine; it stops applying once the
81		// server context is done, like the per-command check of the old batch.	148		// server context is done, like the per-command check of the old batch.
82		func (s *Server) applyQueued(q queuedCommand) {	149		func (s *Server) applyQueued(q queuedCommand) {
83		if s.ctx != nil && s.ctx.Err() != nil {	150		if s.ctx != nil && s.ctx.Err() != nil {
84		return	151		return
85		}	152		}
			153	+	s.commandStats.observe(q.at, time.Now())
86		s.apply(q.command())	154		s.apply(q.command())
87		}	155		}

M internal/adapters/wayland/commands_test.go +85 -0


@@ -3,12 +3,14 @@ package wayland
3	3		import (
4	4		"context"
5	5		"reflect"
	6	+	"strings"
6	7		"sync"
7	8		"testing"
8	9		"time"
9	10
10	11		"github.com/bnema/neferwl/internal/adapters/logging"
11	12		"github.com/bnema/neferwl/internal/ports"
	13	+	"github.com/bnema/zerowrap"
12	14		)
13	15
14	16		// Adjacent motions for one window fold into the last queued command, summing

@@ -205,3 +207,86 @@ func releaseOnce(t testing.T) release {
205	207		t.Cleanup(r.do)
206	208		return r
207	209		}
	210	+
	211	+	// A merged motion keeps the push time of the first motion folded into it, so
	212	+	// the reported latency is the longest wait any of its motions had.
	213	+	func TestMergeMotionKeepsOldestPushTime(t *testing.T) {
	214	+	t0 := time.Unix(100, 0)
	215	+	first := queue(ports.PointerMotionTo{ID: 1, DX: 1})
	216	+	first.at = t0
	217	+	second := queue(ports.PointerMotionTo{ID: 1, DX: 2})
	218	+	second.at = t0.Add(5 * time.Millisecond)
	219	+	if !mergeMotion(&first, second) {
	220	+	t.Fatal("adjacent motions did not merge")
	221	+	}
	222	+	if !first.at.Equal(t0) {
	223	+	t.Fatalf("merged at = %v, want the oldest %v", first.at, t0)
	224	+	}
	225	+	}
	226	+
	227	+	// commandStats keeps the largest push-to-apply wait and the applied count,
	228	+	// and take resets both.
	229	+	func TestCommandStatsObserveAndTake(t *testing.T) {
	230	+	var c commandStats
	231	+	t0 := time.Unix(100, 0)
	232	+	c.observe(t0, t0.Add(3*time.Millisecond))
	233	+	c.observe(t0, t0.Add(9*time.Millisecond))
	234	+	c.observe(t0, t0.Add(time.Millisecond))
	235	+	if n, d := c.take(); n != 3 \|\| d != 9*time.Millisecond {
	236	+	t.Fatalf("take = %d, %v; want 3, 9ms", n, d)
	237	+	}
	238	+	if n, d := c.take(); n != 0 \|\| d != 0 {
	239	+	t.Fatalf("second take = %d, %v; want zeros", n, d)
	240	+	}
	241	+	}
	242	+
	243	+	// The ring counts the largest queued length and the folded pushes, and
	244	+	// takeStats returns then resets them.
	245	+	func TestCommandRingStats(t *testing.T) {
	246	+	s, _, _, _, _, _ := presentServer(t)
	247	+	r := newCommandRing(s.display, func(queuedCommand) {})
	248	+	release := releaseOnce(t)
	249	+	stalled := make(chan struct{})
	250	+	s.display.Post(func() { close(stalled); <-release.ch })
	251	+	<-stalled
	252	+	r.Push(queue(ports.PointerFocus{ID: 1}))
	253	+	r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1}))
	254	+	r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1})) // folded
	255	+	r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1})) // folded
	256	+	r.Push(queue(ports.PointerFocus{ID: 2}))
	257	+	st := r.takeStats()
	258	+	if st.maxLen != 3 \|\| st.merged != 2 \|\| st.maxPushBlock != 0 {
	259	+	t.Fatalf("stats = %+v, want maxLen 3, merged 2, no block", st)
	260	+	}
	261	+	if st := r.takeStats(); st != (ringStats{}) {
	262	+	t.Fatalf("stats after take = %+v, want zero", st)
	263	+	}
	264	+	release.do()
	265	+	}
	266	+
	267	+	// The report logs the period's figures and resets them; a period without
	268	+	// applied commands logs nothing.
	269	+	func TestReportCommandStats(t *testing.T) {
	270	+	var out lockedBuffer
	271	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out}).WithField("component", "wayland")
	272	+	h := newAdmissionHarnessWith(t, 1, Options{}, log)
	273	+	s := h.s
	274	+	s.reportCommandStats()
	275	+	if strings.Contains(out.String(), "command stats") {
	276	+	t.Fatalf("idle period logged: %s", out.String())
	277	+	}
	278	+	t0 := time.Unix(100, 0)
	279	+	s.commandStats.observe(t0, t0.Add(1500*time.Microsecond))
	280	+	s.reportCommandStats()
	281	+	got := out.String()
	282	+	for _, want := range []string{`"message":"command stats"`, `"commands":1`, `"merged":0`, `"max_queued":0`, `"max_push_block_ms":0`, `"max_latency_ms":1.5`} {
	283	+	if !strings.Contains(got, want) {
	284	+	t.Fatalf("log lacks %s: %s", want, got)
	285	+	}
	286	+	}
	287	+	n := strings.Count(got, "command stats")
	288	+	s.reportCommandStats()
	289	+	if strings.Count(out.String(), "command stats") != n {
	290	+	t.Fatalf("stats not reset: %s", out.String())
	291	+	}
	292	+	}


@@ -3,12 +3,14 @@ package wayland
3	import (	3		import (
4	"context"	4		"context"
5	"reflect"	5		"reflect"
		6	+	"strings"
6	"sync"	7		"sync"
7	"testing"	8		"testing"
8	"time"	9		"time"
9		10
10	"github.com/bnema/neferwl/internal/adapters/logging"	11		"github.com/bnema/neferwl/internal/adapters/logging"
11	"github.com/bnema/neferwl/internal/ports"	12		"github.com/bnema/neferwl/internal/ports"
		13	+	"github.com/bnema/zerowrap"
12	)	14		)
13		15
14	// Adjacent motions for one window fold into the last queued command, summing	16		// Adjacent motions for one window fold into the last queued command, summing

@@ -205,3 +207,86 @@ func releaseOnce(t testing.T) release {
205	t.Cleanup(r.do)	207		t.Cleanup(r.do)
206	return r	208		return r
207	}	209		}
		210	+
		211	+	// A merged motion keeps the push time of the first motion folded into it, so
		212	+	// the reported latency is the longest wait any of its motions had.
		213	+	func TestMergeMotionKeepsOldestPushTime(t *testing.T) {
		214	+	t0 := time.Unix(100, 0)
		215	+	first := queue(ports.PointerMotionTo{ID: 1, DX: 1})
		216	+	first.at = t0
		217	+	second := queue(ports.PointerMotionTo{ID: 1, DX: 2})
		218	+	second.at = t0.Add(5 * time.Millisecond)
		219	+	if !mergeMotion(&first, second) {
		220	+	t.Fatal("adjacent motions did not merge")
		221	+	}
		222	+	if !first.at.Equal(t0) {
		223	+	t.Fatalf("merged at = %v, want the oldest %v", first.at, t0)
		224	+	}
		225	+	}
		226	+
		227	+	// commandStats keeps the largest push-to-apply wait and the applied count,
		228	+	// and take resets both.
		229	+	func TestCommandStatsObserveAndTake(t *testing.T) {
		230	+	var c commandStats
		231	+	t0 := time.Unix(100, 0)
		232	+	c.observe(t0, t0.Add(3*time.Millisecond))
		233	+	c.observe(t0, t0.Add(9*time.Millisecond))
		234	+	c.observe(t0, t0.Add(time.Millisecond))
		235	+	if n, d := c.take(); n != 3 \|\| d != 9*time.Millisecond {
		236	+	t.Fatalf("take = %d, %v; want 3, 9ms", n, d)
		237	+	}
		238	+	if n, d := c.take(); n != 0 \|\| d != 0 {
		239	+	t.Fatalf("second take = %d, %v; want zeros", n, d)
		240	+	}
		241	+	}
		242	+
		243	+	// The ring counts the largest queued length and the folded pushes, and
		244	+	// takeStats returns then resets them.
		245	+	func TestCommandRingStats(t *testing.T) {
		246	+	s, _, _, _, _, _ := presentServer(t)
		247	+	r := newCommandRing(s.display, func(queuedCommand) {})
		248	+	release := releaseOnce(t)
		249	+	stalled := make(chan struct{})
		250	+	s.display.Post(func() { close(stalled); <-release.ch })
		251	+	<-stalled
		252	+	r.Push(queue(ports.PointerFocus{ID: 1}))
		253	+	r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1}))
		254	+	r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1})) // folded
		255	+	r.Push(queue(ports.PointerMotionTo{ID: 1, DX: 1})) // folded
		256	+	r.Push(queue(ports.PointerFocus{ID: 2}))
		257	+	st := r.takeStats()
		258	+	if st.maxLen != 3 \|\| st.merged != 2 \|\| st.maxPushBlock != 0 {
		259	+	t.Fatalf("stats = %+v, want maxLen 3, merged 2, no block", st)
		260	+	}
		261	+	if st := r.takeStats(); st != (ringStats{}) {
		262	+	t.Fatalf("stats after take = %+v, want zero", st)
		263	+	}
		264	+	release.do()
		265	+	}
		266	+
		267	+	// The report logs the period's figures and resets them; a period without
		268	+	// applied commands logs nothing.
		269	+	func TestReportCommandStats(t *testing.T) {
		270	+	var out lockedBuffer
		271	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out}).WithField("component", "wayland")
		272	+	h := newAdmissionHarnessWith(t, 1, Options{}, log)
		273	+	s := h.s
		274	+	s.reportCommandStats()
		275	+	if strings.Contains(out.String(), "command stats") {
		276	+	t.Fatalf("idle period logged: %s", out.String())
		277	+	}
		278	+	t0 := time.Unix(100, 0)
		279	+	s.commandStats.observe(t0, t0.Add(1500*time.Microsecond))
		280	+	s.reportCommandStats()
		281	+	got := out.String()
		282	+	for _, want := range []string{`"message":"command stats"`, `"commands":1`, `"merged":0`, `"max_queued":0`, `"max_push_block_ms":0`, `"max_latency_ms":1.5`} {
		283	+	if !strings.Contains(got, want) {
		284	+	t.Fatalf("log lacks %s: %s", want, got)
		285	+	}
		286	+	}
		287	+	n := strings.Count(got, "command stats")
		288	+	s.reportCommandStats()
		289	+	if strings.Count(out.String(), "command stats") != n {
		290	+	t.Fatalf("stats not reset: %s", out.String())
		291	+	}
		292	+	}

M internal/adapters/wayland/postring.go +28 -0


@@ -2,6 +2,7 @@ package wayland
2	2
3	3		import (
4	4		"sync"
	5	+	"time"
5	6
6	7		"github.com/bnema/purego-libwayland/server"
7	8		)

@@ -31,6 +32,25 @@ type postRing[T any] struct {
31	32		ring, spare []T
32	33		limit int
33	34		pending bool // a drain is posted and has not taken the ring yet
	35	+	stats ringStats
	36	+	}
	37	+
	38	+	// ringStats is what a ring saw since the last takeStats. It lets a live log
	39	+	// tell a slow display goroutine (long queue, long push block) from a slow
	40	+	// producer.
	41	+	type ringStats struct {
	42	+	maxLen int // largest queued length
	43	+	merged int // pushes folded into a queued value
	44	+	maxPushBlock time.Duration // longest wait of one Push on a full ring
	45	+	}
	46	+
	47	+	// takeStats returns the counters and resets them.
	48	+	func (r *postRing[T]) takeStats() ringStats {
	49	+	r.mu.Lock()
	50	+	st := r.stats
	51	+	r.stats = ringStats{}
	52	+	r.mu.Unlock()
	53	+	return st
34	54		}
35	55
36	56		// newPostRing returns a ring of limit values whose values apply runs on the

@@ -59,16 +79,21 @@ func newMergingPostRing[T any](d *server.Display, limit int, apply func(T), merg
59	79		// the caller, which must not be the display goroutine, until a drain made
60	80		// room; it reports false, with v dropped, once the display has stopped.
61	81		func (r *postRing[T]) Push(v T) bool {
	82	+	var blocked time.Duration // total wait of this push on a full ring
62	83		for {
63	84		r.mu.Lock()
64	85		// A non-empty ring always has its drain posted, so the last value is
65	86		// still queued: merging into it needs neither room nor a new post.
66	87		if n := len(r.ring); r.merge != nil && n > 0 && r.merge(&r.ring[n-1], v) {
	88	+	r.stats.merged++
	89	+	r.stats.maxPushBlock = max(r.stats.maxPushBlock, blocked)
67	90		r.mu.Unlock()
68	91		return true
69	92		}
70	93		if len(r.ring) < r.limit {
71	94		r.ring = append(r.ring, v)
	95	+	r.stats.maxLen = max(r.stats.maxLen, len(r.ring))
	96	+	r.stats.maxPushBlock = max(r.stats.maxPushBlock, blocked)
72	97		post := !r.pending
73	98		r.pending = true
74	99		room := len(r.ring) < r.limit

@@ -82,8 +107,11 @@ func (r *postRing[T]) Push(v T) bool {
82	107		return true
83	108		}
84	109		r.mu.Unlock()
	110	+	// Only the blocking path reads the clock: the fast path stays free.
	111	+	start := time.Now()
85	112		select {
86	113		case <-r.space:
	114	+	blocked += time.Since(start)
87	115		case <-r.display.Stopped():
88	116		return false
89	117		}


@@ -2,6 +2,7 @@ package wayland
2		2
3	import (	3		import (
4	"sync"	4		"sync"
		5	+	"time"
5		6
6	"github.com/bnema/purego-libwayland/server"	7		"github.com/bnema/purego-libwayland/server"
7	)	8		)

@@ -31,6 +32,25 @@ type postRing[T any] struct {
31	ring, spare []T	32		ring, spare []T
32	limit int	33		limit int
33	pending bool // a drain is posted and has not taken the ring yet	34		pending bool // a drain is posted and has not taken the ring yet
		35	+	stats ringStats
		36	+	}
		37	+
		38	+	// ringStats is what a ring saw since the last takeStats. It lets a live log
		39	+	// tell a slow display goroutine (long queue, long push block) from a slow
		40	+	// producer.
		41	+	type ringStats struct {
		42	+	maxLen int // largest queued length
		43	+	merged int // pushes folded into a queued value
		44	+	maxPushBlock time.Duration // longest wait of one Push on a full ring
		45	+	}
		46	+
		47	+	// takeStats returns the counters and resets them.
		48	+	func (r *postRing[T]) takeStats() ringStats {
		49	+	r.mu.Lock()
		50	+	st := r.stats
		51	+	r.stats = ringStats{}
		52	+	r.mu.Unlock()
		53	+	return st
34	}	54		}
35		55
36	// newPostRing returns a ring of limit values whose values apply runs on the	56		// newPostRing returns a ring of limit values whose values apply runs on the

@@ -59,16 +79,21 @@ func newMergingPostRing[T any](d *server.Display, limit int, apply func(T), merg
59	// the caller, which must not be the display goroutine, until a drain made	79		// the caller, which must not be the display goroutine, until a drain made
60	// room; it reports false, with v dropped, once the display has stopped.	80		// room; it reports false, with v dropped, once the display has stopped.
61	func (r *postRing[T]) Push(v T) bool {	81		func (r *postRing[T]) Push(v T) bool {
		82	+	var blocked time.Duration // total wait of this push on a full ring
62	for {	83		for {
63	r.mu.Lock()	84		r.mu.Lock()
64	// A non-empty ring always has its drain posted, so the last value is	85		// A non-empty ring always has its drain posted, so the last value is
65	// still queued: merging into it needs neither room nor a new post.	86		// still queued: merging into it needs neither room nor a new post.
66	if n := len(r.ring); r.merge != nil && n > 0 && r.merge(&r.ring[n-1], v) {	87		if n := len(r.ring); r.merge != nil && n > 0 && r.merge(&r.ring[n-1], v) {
		88	+	r.stats.merged++
		89	+	r.stats.maxPushBlock = max(r.stats.maxPushBlock, blocked)
67	r.mu.Unlock()	90		r.mu.Unlock()
68	return true	91		return true
69	}	92		}
70	if len(r.ring) < r.limit {	93		if len(r.ring) < r.limit {
71	r.ring = append(r.ring, v)	94		r.ring = append(r.ring, v)
		95	+	r.stats.maxLen = max(r.stats.maxLen, len(r.ring))
		96	+	r.stats.maxPushBlock = max(r.stats.maxPushBlock, blocked)
72	post := !r.pending	97		post := !r.pending
73	r.pending = true	98		r.pending = true
74	room := len(r.ring) < r.limit	99		room := len(r.ring) < r.limit

@@ -82,8 +107,11 @@ func (r *postRing[T]) Push(v T) bool {
82	return true	107		return true
83	}	108		}
84	r.mu.Unlock()	109		r.mu.Unlock()
		110	+	// Only the blocking path reads the clock: the fast path stays free.
		111	+	start := time.Now()
85	select {	112		select {
86	case <-r.space:	113		case <-r.space:
		114	+	blocked += time.Since(start)
87	case <-r.display.Stopped():	115		case <-r.display.Stopped():
88	return false	116		return false
89	}	117		}

M internal/adapters/wayland/postring_test.go +41 -0


@@ -2,6 +2,8 @@ package wayland
2	2
3	3		import (
4	4		"context"
	5	+	"runtime"
	6	+	"strings"
5	7		"sync"
6	8		"testing"
7	9		"time"

@@ -158,3 +160,42 @@ func TestPostRingPushStopsWithDisplay(t *testing.T) {
158	160		t.Fatal("push blocked after the display stopped")
159	161		}
160	162		}
	163	+
	164	+	// A push that waits on a full ring reports how long, and a drain-side push
	165	+	// that finds room does not.
	166	+	func TestPostRingStatsPushBlock(t *testing.T) {
	167	+	s, _, _, _, _, _ := presentServer(t)
	168	+	r := newPostRing(s.display, 1, func(int) {})
	169	+	release := make(chan struct{})
	170	+	stalled := make(chan struct{})
	171	+	s.display.Post(func() { close(stalled); <-release })
	172	+	<-stalled
	173	+	r.Push(1)
	174	+	if st := r.takeStats(); st.maxLen != 1 \|\| st.maxPushBlock != 0 {
	175	+	t.Fatalf("stats = %+v, want maxLen 1 and no block", st)
	176	+	}
	177	+	done := make(chan struct{})
	178	+	go func() { defer close(done); r.Push(2) }() // full: waits for the drain
	179	+	waitPushWaiting(t)
	180	+	close(release)
	181	+	<-done
	182	+	if st := r.takeStats(); st.maxPushBlock <= 0 {
	183	+	t.Fatalf("stats = %+v, want a recorded push block", st)
	184	+	}
	185	+	}
	186	+
	187	+	// waitPushWaiting returns once a goroutine waits inside postRing.Push, read
	188	+	// from the goroutine dump, so the test needs no sleep to know the push blocks.
	189	+	func waitPushWaiting(t *testing.T) {
	190	+	t.Helper()
	191	+	buf := make([]byte, 1<<20)
	192	+	for {
	193	+	dump := string(buf[:runtime.Stack(buf, true)])
	194	+	for _, g := range strings.Split(dump, "\n\n") {
	195	+	if strings.Contains(g, "[select") && strings.Contains(g, "postRing") && strings.Contains(g, ").Push(") {
	196	+	return
	197	+	}
	198	+	}
	199	+	runtime.Gosched()
	200	+	}
	201	+	}


@@ -2,6 +2,8 @@ package wayland
2		2
3	import (	3		import (
4	"context"	4		"context"
		5	+	"runtime"
		6	+	"strings"
5	"sync"	7		"sync"
6	"testing"	8		"testing"
7	"time"	9		"time"

@@ -158,3 +160,42 @@ func TestPostRingPushStopsWithDisplay(t *testing.T) {
158	t.Fatal("push blocked after the display stopped")	160		t.Fatal("push blocked after the display stopped")
159	}	161		}
160	}	162		}
		163	+
		164	+	// A push that waits on a full ring reports how long, and a drain-side push
		165	+	// that finds room does not.
		166	+	func TestPostRingStatsPushBlock(t *testing.T) {
		167	+	s, _, _, _, _, _ := presentServer(t)
		168	+	r := newPostRing(s.display, 1, func(int) {})
		169	+	release := make(chan struct{})
		170	+	stalled := make(chan struct{})
		171	+	s.display.Post(func() { close(stalled); <-release })
		172	+	<-stalled
		173	+	r.Push(1)
		174	+	if st := r.takeStats(); st.maxLen != 1 \|\| st.maxPushBlock != 0 {
		175	+	t.Fatalf("stats = %+v, want maxLen 1 and no block", st)
		176	+	}
		177	+	done := make(chan struct{})
		178	+	go func() { defer close(done); r.Push(2) }() // full: waits for the drain
		179	+	waitPushWaiting(t)
		180	+	close(release)
		181	+	<-done
		182	+	if st := r.takeStats(); st.maxPushBlock <= 0 {
		183	+	t.Fatalf("stats = %+v, want a recorded push block", st)
		184	+	}
		185	+	}
		186	+
		187	+	// waitPushWaiting returns once a goroutine waits inside postRing.Push, read
		188	+	// from the goroutine dump, so the test needs no sleep to know the push blocks.
		189	+	func waitPushWaiting(t *testing.T) {
		190	+	t.Helper()
		191	+	buf := make([]byte, 1<<20)
		192	+	for {
		193	+	dump := string(buf[:runtime.Stack(buf, true)])
		194	+	for _, g := range strings.Split(dump, "\n\n") {
		195	+	if strings.Contains(g, "[select") && strings.Contains(g, "postRing") && strings.Contains(g, ").Push(") {
		196	+	return
		197	+	}
		198	+	}
		199	+	runtime.Gosched()
		200	+	}
		201	+	}

M internal/adapters/wayland/server.go +3 -1


@@ -137,6 +137,7 @@ type Server struct {
137	137		castRing *postRing[ports.CastCommand]
138	138		// commands carries core's client commands to the display goroutine.
139	139		commands *postRing[queuedCommand]
	140	+	commandStats commandStats
140	141		castInvFn func()
141	142		castInvPosted bool
142	143		castInvSent bool

@@ -535,7 +536,7 @@ func (s *Server) Run(ctx context.Context) error {
535	536		s.started = time.Now()
536	537		s.ctx = ctx
537	538		var wg sync.WaitGroup
538		-	wg.Add(13)
	539	+	wg.Add(14)
539	540		go func() { defer wg.Done(); s.forwardBackendSecurity(ctx) }()
540	541		go func() { defer wg.Done(); s.forwardSecurityEvents(ctx) }()
541	542		go func() { defer wg.Done(); s.forwardLeases(ctx) }()

@@ -566,6 +567,7 @@ func (s *Server) Run(ctx context.Context) error {
566	567		}
567	568		}
568	569		}()
	570	+	go func() { defer wg.Done(); s.logCommandStats(ctx) }()
569	571		go func() { defer wg.Done(); s.forwardOutputFormats(ctx) }()
570	572		go func() { defer wg.Done(); s.forwardOutputHeads(ctx) }()
571	573		go func() { defer wg.Done(); s.forwardOutputApplied(ctx) }()

 	castRing     *postRing[ports.CastCommand]
 	// commands carries core's client commands to the display goroutine.
 	commands      *postRing[queuedCommand]
+	commandStats  commandStats
 	castInvFn     func()
 	castInvPosted bool
 	castInvSent   bool
 	s.started = time.Now()
 	s.ctx = ctx
 	var wg sync.WaitGroup
-	wg.Add(13)
+	wg.Add(14)
 	go func() { defer wg.Done(); s.forwardBackendSecurity(ctx) }()
 	go func() { defer wg.Done(); s.forwardSecurityEvents(ctx) }()
 	go func() { defer wg.Done(); s.forwardLeases(ctx) }()
 			}
 		}
 	}()
+	go func() { defer wg.Done(); s.logCommandStats(ctx) }()
 	go func() { defer wg.Done(); s.forwardOutputFormats(ctx) }()
 	go func() { defer wg.Done(); s.forwardOutputHeads(ctx) }()
 	go func() { defer wg.Done(); s.forwardOutputApplied(ctx) }()

M internal/app/app.go +6 -3


@@ -214,14 +214,16 @@ func renderNode(dev uint64) string {
214	214		}
215	215
216	216		// latest sends v without blocking and replaces a value the consumer has not
217		-	// read yet: the newest wins. Only one goroutine may send on ch; consumers only
218		-	// receive, so the retry after draining cannot find the buffer full again.
219		-	func latest[T any](ch chan T, v T) {
	217	+	// read yet: the newest wins. It reports whether it dropped an unread value.
	218	+	// Only one goroutine may send on ch; consumers only receive, so the retry
	219	+	// after draining cannot find the buffer full again.
	220	+	func latest[T any](ch chan T, v T) (replaced bool) {
220	221		select {
221	222		case ch <- v:
222	223		default:
223	224		select {
224	225		case <-ch:
	226	+	replaced = true
225	227		default:
226	228		}
227	229		select {

@@ -229,4 +231,5 @@ func latest[T any](ch chan T, v T) {
229	231		default:
230	232		}
231	233		}
	234	+	return replaced
232	235		}

 }
 // latest sends v without blocking and replaces a value the consumer has not
-// read yet: the newest wins. Only one goroutine may send on ch; consumers only
+// read yet: the newest wins. It reports whether it dropped an unread value.
-// receive, so the retry after draining cannot find the buffer full again.
+// Only one goroutine may send on ch; consumers only receive, so the retry
-func latest[T any](ch chan T, v T) {
+// after draining cannot find the buffer full again.
+func latest[T any](ch chan T, v T) (replaced bool) {
 	select {
 	case ch <- v:
 	default:
 		select {
 		case <-ch:
+			replaced = true
 		default:
 		}
 		select {
 		default:
 		}
 	}
+	return replaced
 }

A internal/app/corestats.go +36 -0

@@ -0,0 +1,36 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"time"
	6	+
	7	+	"github.com/bnema/neferwl/internal/core"
	8	+	"github.com/bnema/zerowrap"
	9	+	)
	10	+
	11	+	// commandStatsEvery is how often the command outbox counters are logged,
	12	+	// like the output and input stats.
	13	+	const commandStatsEvery = 10 * time.Second
	14	+
	15	+	// logCommandStats logs core's command counters every commandStatsEvery
	16	+	// until ctx ends. A queued command or a flush wait means the display goroutine
	17	+	// held core's commands up; an idle period logs nothing.
	18	+	func logCommandStats(ctx context.Context, c *core.Core, log zerowrap.Logger) {
	19	+	t := time.NewTicker(commandStatsEvery)
	20	+	defer t.Stop()
	21	+	for {
	22	+	select {
	23	+	case <-ctx.Done():
	24	+	return
	25	+	case <-t.C:
	26	+	}
	27	+	s := c.CommandStats()
	28	+	if s.Direct == 0 && s.Queued == 0 {
	29	+	continue
	30	+	}
	31	+	log.Info().Int("direct", s.Direct).Int("queued", s.Queued).Int("merged", s.Merged).
	32	+	Uint64("max_backlog", s.MaxBacklog).
	33	+	Float64("max_queue_wait_ms", s.MaxQueueWait.Seconds()*1e3).
	34	+	Float64("max_flush_wait_ms", s.MaxFlushWait.Seconds()*1e3).Msg("outbox stats")
	35	+	}
	36	+	}

@@ -0,0 +1,36 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"time"
	6	+
	7	+	"github.com/bnema/neferwl/internal/core"
	8	+	"github.com/bnema/zerowrap"
	9	+	)
	10	+
	11	+	// commandStatsEvery is how often the command outbox counters are logged,
	12	+	// like the output and input stats.
	13	+	const commandStatsEvery = 10 * time.Second
	14	+
	15	+	// logCommandStats logs core's command counters every commandStatsEvery
	16	+	// until ctx ends. A queued command or a flush wait means the display goroutine
	17	+	// held core's commands up; an idle period logs nothing.
	18	+	func logCommandStats(ctx context.Context, c *core.Core, log zerowrap.Logger) {
	19	+	t := time.NewTicker(commandStatsEvery)
	20	+	defer t.Stop()
	21	+	for {
	22	+	select {
	23	+	case <-ctx.Done():
	24	+	return
	25	+	case <-t.C:
	26	+	}
	27	+	s := c.CommandStats()
	28	+	if s.Direct == 0 && s.Queued == 0 {
	29	+	continue
	30	+	}
	31	+	log.Info().Int("direct", s.Direct).Int("queued", s.Queued).Int("merged", s.Merged).
	32	+	Uint64("max_backlog", s.MaxBacklog).
	33	+	Float64("max_queue_wait_ms", s.MaxQueueWait.Seconds()*1e3).
	34	+	Float64("max_flush_wait_ms", s.MaxFlushWait.Seconds()*1e3).Msg("outbox stats")
	35	+	}
	36	+	}

M internal/app/drm.go +4 -0


@@ -347,10 +347,14 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
347	347		cards := map[string]*drmCard{}
348	348		outputs := map[string]*drm.Output{} // owner result, read only after done
349	349		send := func(ev ports.OutputEvent) {
	350	+	// Only the select is timed: the router logs the longest wait, which
	351	+	// tells whether core holds up the backend loop.
	352	+	start := time.Now()
350	353		select {
351	354		case ch.events <- ev:
352	355		case <-ctx.Done():
353	356		}
	357	+	set.router.noteEventSend(time.Since(start))
354	358		}
355	359		// stopping holds outputs asked to stop: true when they restart with a
356	360		// new mode (core keeps the screen), false when they are gone.


@@ -347,10 +347,14 @@ func (b *drmBackend) runOutputs(ctx context.Context, want func(ports.Config) drm
347	cards := map[string]*drmCard{}	347		cards := map[string]*drmCard{}
348	outputs := map[string]*drm.Output{} // owner result, read only after done	348		outputs := map[string]*drm.Output{} // owner result, read only after done
349	send := func(ev ports.OutputEvent) {	349		send := func(ev ports.OutputEvent) {
		350	+	// Only the select is timed: the router logs the longest wait, which
		351	+	// tells whether core holds up the backend loop.
		352	+	start := time.Now()
350	select {	353		select {
351	case ch.events <- ev:	354		case ch.events <- ev:
352	case <-ctx.Done():	355		case <-ctx.Done():
353	}	356		}
		357	+	set.router.noteEventSend(time.Since(start))
354	}	358		}
355	// stopping holds outputs asked to stop: true when they restart with a	359		// stopping holds outputs asked to stop: true when they restart with a
356	// new mode (core keeps the screen), false when they are gone.	360		// new mode (core keeps the screen), false when they are gone.

M internal/app/router.go +67 -3


@@ -3,7 +3,12 @@ package app
3	3		import (
4	4		"context"
5	5		"fmt"
	6	+	"sync/atomic"
	7	+	"time"
6	8
	9	+	"github.com/bnema/zerowrap"
	10	+
	11	+	"github.com/bnema/neferwl/internal/adapters/logging"
7	12		"github.com/bnema/neferwl/internal/adapters/outputkit/capture"
8	13		"github.com/bnema/neferwl/internal/ports"
9	14		)

@@ -31,6 +36,25 @@ type outputRouter struct {
31	36		in outputRoutes
32	37		ops chan routerOp
33	38		done chan struct{} // closed when run returns
	39	+
	40	+	log zerowrap.Logger
	41	+	// stats counts since the last report; only run's goroutine touches it.
	42	+	stats routerStats
	43	+	// eventWait is the longest time (ns) the backend loop blocked sending an
	44	+	// OutputEvent to core since the last report. The backend loop writes it,
	45	+	// so it is atomic.
	46	+	eventWait atomic.Int64
	47	+	}
	48	+
	49	+	// routerStatsEvery is the period of the "routing stats" log entry.
	50	+	const routerStatsEvery = 10 * time.Second
	51	+
	52	+	// routerStats tells whether outputs keep up: a scene or cursor replaced
	53	+	// unread means an output did not consume the previous one before the next.
	54	+	type routerStats struct {
	55	+	scenes, scenesReplaced uint64
	56	+	cursor, cursorReplaced uint64
	57	+	capturesFull uint64 // captures failed because the queue was full
34	58		}
35	59
36	60		// outputRoutes are the router's inputs: core's scenes, wayland's cursor and

@@ -67,16 +91,20 @@ type routerOp struct {
67	91		}
68	92
69	93		func newOutputRouter(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputRouter {
70		-	return &outputRouter{outs: map[string]*mailbox{}, off: map[string]bool{}, captured: captured, ctx: ctx, in: in, ops: make(chan routerOp), done: make(chan struct{})}
	94	+	return &outputRouter{outs: map[string]*mailbox{}, off: map[string]bool{}, captured: captured, ctx: ctx, in: in, ops: make(chan routerOp), done: make(chan struct{}), log: logging.For(ctx, "router")}
71	95		}
72	96
73	97		// run routes until ctx ends.
74	98		func (r *outputRouter) run(ctx context.Context) {
75	99		defer close(r.done)
	100	+	report := time.NewTicker(routerStatsEvery)
	101	+	defer report.Stop()
76	102		for {
77	103		select {
78	104		case <-ctx.Done():
79	105		return
	106	+	case <-report.C:
	107	+	r.report()
80	108		case op := <-r.ops:
81	109		r.apply(op)
82	110		case set := <-r.in.scenes:

@@ -89,6 +117,35 @@ func (r *outputRouter) run(ctx context.Context) {
89	117		}
90	118		}
91	119
	120	+	// report logs and resets the counters; it stays quiet when nothing happened.
	121	+	func (r *outputRouter) report() {
	122	+	st, wait := r.stats, time.Duration(r.eventWait.Swap(0))
	123	+	r.stats = routerStats{}
	124	+	if st == (routerStats{}) && wait == 0 {
	125	+	return
	126	+	}
	127	+	r.log.Info().
	128	+	Uint64("scenes", st.scenes).
	129	+	Uint64("scenes_replaced", st.scenesReplaced).
	130	+	Uint64("cursor", st.cursor).
	131	+	Uint64("cursor_replaced", st.cursorReplaced).
	132	+	Uint64("captures_full", st.capturesFull).
	133	+	Float64("max_event_send_ms", float64(wait)/float64(time.Millisecond)).
	134	+	Msg("routing stats")
	135	+	}
	136	+
	137	+	// noteEventSend records how long the backend loop waited for core to take an
	138	+	// event. Backend loop only; lock-free so it never adds to the wait.
	139	+	func (r *outputRouter) noteEventSend(d time.Duration) {
	140	+	n := int64(d)
	141	+	for {
	142	+	cur := r.eventWait.Load()
	143	+	if n <= cur \|\| r.eventWait.CompareAndSwap(cur, n) {
	144	+	return
	145	+	}
	146	+	}
	147	+	}
	148	+
92	149		func (r *outputRouter) apply(op routerOp) {
93	150		if op.add != nil {
94	151		op.reply <- r.register(op.name, op.add)

@@ -116,8 +173,11 @@ func (r *outputRouter) unregister(name string, instance ports.OutputInstance) {
116	173		// setCursor gives every output the latest cursor, replacing an unread one.
117	174		func (r *outputRouter) setCursor(c ports.CursorChange) {
118	175		r.cursor = c
	176	+	r.stats.cursor++
119	177		for _, m := range r.outs {
120		-	latest(m.cursor, c)
	178	+	if latest(m.cursor, c) {
	179	+	r.stats.cursorReplaced++
	180	+	}
121	181		}
122	182		}
123	183

@@ -130,7 +190,10 @@ func (r *outputRouter) scenes(set []ports.Scene) {
130	190		delete(r.off, sc.Output)
131	191		}
132	192		if m := r.outs[sc.Output]; m != nil {
133		-	latest(m.scenes, sc)
	193	+	r.stats.scenes++
	194	+	if latest(m.scenes, sc) {
	195	+	r.stats.scenesReplaced++
	196	+	}
134	197		}
135	198		}
136	199		}

@@ -151,6 +214,7 @@ func (r *outputRouter) routeCapture(req ports.CaptureRequest) {
151	214		case <-m.ctx.Done():
152	215		capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
153	216		default:
	217	+	r.stats.capturesFull++
154	218		capture.Fail(r.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), r.captured)
155	219		}
156	220		}


@@ -3,7 +3,12 @@ package app
3		import (	3		import (
4		"context"	4		"context"
5		"fmt"	5		"fmt"
			6	+	"sync/atomic"
			7	+	"time"
6			8
			9	+	"github.com/bnema/zerowrap"
			10	+
			11	+	"github.com/bnema/neferwl/internal/adapters/logging"
7		"github.com/bnema/neferwl/internal/adapters/outputkit/capture"	12		"github.com/bnema/neferwl/internal/adapters/outputkit/capture"
8		"github.com/bnema/neferwl/internal/ports"	13		"github.com/bnema/neferwl/internal/ports"
9		)	14		)

@@ -31,6 +36,25 @@ type outputRouter struct {
31		in outputRoutes	36		in outputRoutes
32		ops chan routerOp	37		ops chan routerOp
33		done chan struct{} // closed when run returns	38		done chan struct{} // closed when run returns
			39	+
			40	+	log zerowrap.Logger
			41	+	// stats counts since the last report; only run's goroutine touches it.
			42	+	stats routerStats
			43	+	// eventWait is the longest time (ns) the backend loop blocked sending an
			44	+	// OutputEvent to core since the last report. The backend loop writes it,
			45	+	// so it is atomic.
			46	+	eventWait atomic.Int64
			47	+	}
			48	+
			49	+	// routerStatsEvery is the period of the "routing stats" log entry.
			50	+	const routerStatsEvery = 10 * time.Second
			51	+
			52	+	// routerStats tells whether outputs keep up: a scene or cursor replaced
			53	+	// unread means an output did not consume the previous one before the next.
			54	+	type routerStats struct {
			55	+	scenes, scenesReplaced uint64
			56	+	cursor, cursorReplaced uint64
			57	+	capturesFull uint64 // captures failed because the queue was full
34		}	58		}
35			59
36		// outputRoutes are the router's inputs: core's scenes, wayland's cursor and	60		// outputRoutes are the router's inputs: core's scenes, wayland's cursor and

@@ -67,16 +91,20 @@ type routerOp struct {
67		}	91		}
68			92
69		func newOutputRouter(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputRouter {	93		func newOutputRouter(ctx context.Context, captured chan<- ports.CaptureDone, in outputRoutes) *outputRouter {
70	-	return &outputRouter{outs: map[string]*mailbox{}, off: map[string]bool{}, captured: captured, ctx: ctx, in: in, ops: make(chan routerOp), done: make(chan struct{})}	94	+	return &outputRouter{outs: map[string]*mailbox{}, off: map[string]bool{}, captured: captured, ctx: ctx, in: in, ops: make(chan routerOp), done: make(chan struct{}), log: logging.For(ctx, "router")}
71		}	95		}
72			96
73		// run routes until ctx ends.	97		// run routes until ctx ends.
74		func (r *outputRouter) run(ctx context.Context) {	98		func (r *outputRouter) run(ctx context.Context) {
75		defer close(r.done)	99		defer close(r.done)
			100	+	report := time.NewTicker(routerStatsEvery)
			101	+	defer report.Stop()
76		for {	102		for {
77		select {	103		select {
78		case <-ctx.Done():	104		case <-ctx.Done():
79		return	105		return
			106	+	case <-report.C:
			107	+	r.report()
80		case op := <-r.ops:	108		case op := <-r.ops:
81		r.apply(op)	109		r.apply(op)
82		case set := <-r.in.scenes:	110		case set := <-r.in.scenes:

@@ -89,6 +117,35 @@ func (r *outputRouter) run(ctx context.Context) {
89		}	117		}
90		}	118		}
91			119
			120	+	// report logs and resets the counters; it stays quiet when nothing happened.
			121	+	func (r *outputRouter) report() {
			122	+	st, wait := r.stats, time.Duration(r.eventWait.Swap(0))
			123	+	r.stats = routerStats{}
			124	+	if st == (routerStats{}) && wait == 0 {
			125	+	return
			126	+	}
			127	+	r.log.Info().
			128	+	Uint64("scenes", st.scenes).
			129	+	Uint64("scenes_replaced", st.scenesReplaced).
			130	+	Uint64("cursor", st.cursor).
			131	+	Uint64("cursor_replaced", st.cursorReplaced).
			132	+	Uint64("captures_full", st.capturesFull).
			133	+	Float64("max_event_send_ms", float64(wait)/float64(time.Millisecond)).
			134	+	Msg("routing stats")
			135	+	}
			136	+
			137	+	// noteEventSend records how long the backend loop waited for core to take an
			138	+	// event. Backend loop only; lock-free so it never adds to the wait.
			139	+	func (r *outputRouter) noteEventSend(d time.Duration) {
			140	+	n := int64(d)
			141	+	for {
			142	+	cur := r.eventWait.Load()
			143	+	if n <= cur \|\| r.eventWait.CompareAndSwap(cur, n) {
			144	+	return
			145	+	}
			146	+	}
			147	+	}
			148	+
92		func (r *outputRouter) apply(op routerOp) {	149		func (r *outputRouter) apply(op routerOp) {
93		if op.add != nil {	150		if op.add != nil {
94		op.reply <- r.register(op.name, op.add)	151		op.reply <- r.register(op.name, op.add)

@@ -116,8 +173,11 @@ func (r *outputRouter) unregister(name string, instance ports.OutputInstance) {
116		// setCursor gives every output the latest cursor, replacing an unread one.	173		// setCursor gives every output the latest cursor, replacing an unread one.
117		func (r *outputRouter) setCursor(c ports.CursorChange) {	174		func (r *outputRouter) setCursor(c ports.CursorChange) {
118		r.cursor = c	175		r.cursor = c
			176	+	r.stats.cursor++
119		for _, m := range r.outs {	177		for _, m := range r.outs {
120	-	latest(m.cursor, c)	178	+	if latest(m.cursor, c) {
			179	+	r.stats.cursorReplaced++
			180	+	}
121		}	181		}
122		}	182		}
123			183

@@ -130,7 +190,10 @@ func (r *outputRouter) scenes(set []ports.Scene) {
130		delete(r.off, sc.Output)	190		delete(r.off, sc.Output)
131		}	191		}
132		if m := r.outs[sc.Output]; m != nil {	192		if m := r.outs[sc.Output]; m != nil {
133	-	latest(m.scenes, sc)	193	+	r.stats.scenes++
			194	+	if latest(m.scenes, sc) {
			195	+	r.stats.scenesReplaced++
			196	+	}
134		}	197		}
135		}	198		}
136		}	199		}

@@ -151,6 +214,7 @@ func (r *outputRouter) routeCapture(req ports.CaptureRequest) {
151		case <-m.ctx.Done():	214		case <-m.ctx.Done():
152		capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)	215		capture.Fail(r.ctx, req, fmt.Errorf("output %q stopped", req.Output), r.captured)
153		default:	216		default:
			217	+	r.stats.capturesFull++
154		capture.Fail(r.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), r.captured)	218		capture.Fail(r.ctx, req, fmt.Errorf("output %q capture queue full", req.Output), r.captured)
155		}	219		}
156		}	220		}

A internal/app/router_stats_test.go +141 -0

@@ -0,0 +1,141 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"strings"
	7	+	"testing"
	8	+	"testing/synctest"
	9	+	"time"
	10	+
	11	+	"github.com/bnema/zerowrap"
	12	+
	13	+	"github.com/bnema/neferwl/internal/ports"
	14	+	)
	15	+
	16	+	// statsRegistered returns a router with output A registered and its initial
	17	+	// cursor drained.
	18	+	func statsRegistered(ctx context.Context, in outputRoutes) (outputRouter, mailbox) {
	19	+	r := newOutputRouter(ctx, nil, in)
	20	+	mctx, stop := context.WithCancel(ctx)
	21	+	m := newMailbox(1, mctx, stop, make(chan struct{}))
	22	+	r.register("A", m)
	23	+	<-m.cursor
	24	+	return r, m
	25	+	}
	26	+
	27	+	// A scene or cursor sent before the output read the previous one counts as
	28	+	// replaced: the output is behind.
	29	+	func TestRouterStatsCountReplaced(t *testing.T) {
	30	+	r, m := statsRegistered(t.Context(), outputRoutes{})
	31	+	r.scenes([]ports.Scene{{Output: "A", Seq: 1}})
	32	+	if r.stats.scenes != 1 \|\| r.stats.scenesReplaced != 0 {
	33	+	t.Fatalf("one scene: %+v", r.stats)
	34	+	}
	35	+	r.scenes([]ports.Scene{{Output: "A", Seq: 2}, {Output: "missing"}})
	36	+	if r.stats.scenes != 2 \|\| r.stats.scenesReplaced != 1 {
	37	+	t.Fatalf("two scenes unread: %+v", r.stats)
	38	+	}
	39	+	<-m.scenes
	40	+	r.scenes([]ports.Scene{{Output: "A", Seq: 3}})
	41	+	if r.stats.scenesReplaced != 1 {
	42	+	t.Fatalf("a drained mailbox is not behind: %+v", r.stats)
	43	+	}
	44	+
	45	+	r.setCursor(ports.CursorChange{Shape: "text"})
	46	+	r.setCursor(ports.CursorChange{Shape: "pointer"})
	47	+	if r.stats.cursor != 2 \|\| r.stats.cursorReplaced != 1 {
	48	+	t.Fatalf("cursor: %+v", r.stats)
	49	+	}
	50	+	if got := <-m.cursor; got.Shape != "pointer" {
	51	+	t.Fatalf("cursor %+v", got)
	52	+	}
	53	+	}
	54	+
	55	+	// A full capture queue fails the request fast and counts it.
	56	+	func TestRouterStatsCapturesFull(t *testing.T) {
	57	+	replies := make(chan ports.CaptureDone, 2)
	58	+	r, m := statsRegistered(t.Context(), outputRoutes{})
	59	+	r.captured = replies
	60	+	for id := range uint64(cap(m.captures)) {
	61	+	r.routeCapture(ports.CaptureRequest{ID: id, Output: "A"})
	62	+	}
	63	+	if r.stats.capturesFull != 0 {
	64	+	t.Fatalf("queue not full yet: %+v", r.stats)
	65	+	}
	66	+	r.routeCapture(ports.CaptureRequest{ID: 99, Output: "A"})
	67	+	r.routeCapture(ports.CaptureRequest{ID: 100, Output: "missing"})
	68	+	if r.stats.capturesFull != 1 {
	69	+	t.Fatalf("captures full: %+v", r.stats)
	70	+	}
	71	+	if got := <-replies; got.ID != 99 \|\| got.Err == nil {
	72	+	t.Fatalf("rejected capture: %+v", got)
	73	+	}
	74	+	}
	75	+
	76	+	// The backend loop reports its longest wait on core; shorter ones keep it.
	77	+	func TestRouterNoteEventSendKeepsMax(t *testing.T) {
	78	+	r := newOutputRouter(t.Context(), nil, outputRoutes{})
	79	+	for _, d := range []time.Duration{3 * time.Millisecond, 12 * time.Millisecond, time.Millisecond} {
	80	+	r.noteEventSend(d)
	81	+	}
	82	+	if got := time.Duration(r.eventWait.Load()); got != 12*time.Millisecond {
	83	+	t.Fatalf("max wait %v, want 12ms", got)
	84	+	}
	85	+	}
	86	+
	87	+	// Every 10 s the router logs and resets its counters, and stays quiet when
	88	+	// nothing happened.
	89	+	func TestRouterLogsStatsEvery10s(t *testing.T) {
	90	+	synctest.Test(t, func(t *testing.T) {
	91	+	var out lockedBuffer
	92	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out})
	93	+	ctx, cancel := context.WithCancel(zerowrap.WithCtx(context.Background(), log))
	94	+	in := newRouteInputs()
	95	+	r := newOutputRouter(ctx, nil, in.routes())
	96	+	go r.run(ctx)
	97	+	m := newMailbox(1, ctx, func() {}, make(chan struct{}))
	98	+	if _, ok := r.add("A", m); !ok {
	99	+	t.Fatal("router stopped")
	100	+	}
	101	+	<-m.cursor
	102	+
	103	+	time.Sleep(routerStatsEvery)
	104	+	synctest.Wait()
	105	+	if got := out.String(); got != "" {
	106	+	t.Fatalf("idle router logged: %s", got)
	107	+	}
	108	+
	109	+	in.scenes <- []ports.Scene{{Output: "A", Seq: 1}}
	110	+	in.scenes <- []ports.Scene{{Output: "A", Seq: 2}}
	111	+	in.cursor <- ports.CursorChange{Shape: "text"}
	112	+	in.cursor <- ports.CursorChange{Shape: "pointer"}
	113	+	in.scenes <- nil // the router took both cursors
	114	+	r.noteEventSend(1500 * time.Microsecond)
	115	+	time.Sleep(routerStatsEvery)
	116	+	synctest.Wait()
	117	+
	118	+	lines := strings.Split(strings.TrimSpace(out.String()), "\n")
	119	+	if len(lines) != 1 {
	120	+	t.Fatalf("log lines: %q", lines)
	121	+	}
	122	+	var entry map[string]any
	123	+	if err := json.Unmarshal([]byte(lines[0]), &entry); err != nil {
	124	+	t.Fatal(err)
	125	+	}
	126	+	want := map[string]any{"message": "routing stats", "component": "router", "scenes": 2.0, "scenes_replaced": 1.0, "cursor": 2.0, "cursor_replaced": 1.0, "captures_full": 0.0, "max_event_send_ms": 1.5}
	127	+	for k, v := range want {
	128	+	if entry[k] != v {
	129	+	t.Errorf("%s = %v, want %v (%s)", k, entry[k], v, lines[0])
	130	+	}
	131	+	}
	132	+
	133	+	time.Sleep(routerStatsEvery)
	134	+	synctest.Wait()
	135	+	if n := strings.Count(out.String(), "routing stats"); n != 1 {
	136	+	t.Fatalf("counters were not reset: %d entries", n)
	137	+	}
	138	+	cancel()
	139	+	<-r.done
	140	+	})
	141	+	}

@@ -0,0 +1,141 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"strings"
	7	+	"testing"
	8	+	"testing/synctest"
	9	+	"time"
	10	+
	11	+	"github.com/bnema/zerowrap"
	12	+
	13	+	"github.com/bnema/neferwl/internal/ports"
	14	+	)
	15	+
	16	+	// statsRegistered returns a router with output A registered and its initial
	17	+	// cursor drained.
	18	+	func statsRegistered(ctx context.Context, in outputRoutes) (outputRouter, mailbox) {
	19	+	r := newOutputRouter(ctx, nil, in)
	20	+	mctx, stop := context.WithCancel(ctx)
	21	+	m := newMailbox(1, mctx, stop, make(chan struct{}))
	22	+	r.register("A", m)
	23	+	<-m.cursor
	24	+	return r, m
	25	+	}
	26	+
	27	+	// A scene or cursor sent before the output read the previous one counts as
	28	+	// replaced: the output is behind.
	29	+	func TestRouterStatsCountReplaced(t *testing.T) {
	30	+	r, m := statsRegistered(t.Context(), outputRoutes{})
	31	+	r.scenes([]ports.Scene{{Output: "A", Seq: 1}})
	32	+	if r.stats.scenes != 1 \|\| r.stats.scenesReplaced != 0 {
	33	+	t.Fatalf("one scene: %+v", r.stats)
	34	+	}
	35	+	r.scenes([]ports.Scene{{Output: "A", Seq: 2}, {Output: "missing"}})
	36	+	if r.stats.scenes != 2 \|\| r.stats.scenesReplaced != 1 {
	37	+	t.Fatalf("two scenes unread: %+v", r.stats)
	38	+	}
	39	+	<-m.scenes
	40	+	r.scenes([]ports.Scene{{Output: "A", Seq: 3}})
	41	+	if r.stats.scenesReplaced != 1 {
	42	+	t.Fatalf("a drained mailbox is not behind: %+v", r.stats)
	43	+	}
	44	+
	45	+	r.setCursor(ports.CursorChange{Shape: "text"})
	46	+	r.setCursor(ports.CursorChange{Shape: "pointer"})
	47	+	if r.stats.cursor != 2 \|\| r.stats.cursorReplaced != 1 {
	48	+	t.Fatalf("cursor: %+v", r.stats)
	49	+	}
	50	+	if got := <-m.cursor; got.Shape != "pointer" {
	51	+	t.Fatalf("cursor %+v", got)
	52	+	}
	53	+	}
	54	+
	55	+	// A full capture queue fails the request fast and counts it.
	56	+	func TestRouterStatsCapturesFull(t *testing.T) {
	57	+	replies := make(chan ports.CaptureDone, 2)
	58	+	r, m := statsRegistered(t.Context(), outputRoutes{})
	59	+	r.captured = replies
	60	+	for id := range uint64(cap(m.captures)) {
	61	+	r.routeCapture(ports.CaptureRequest{ID: id, Output: "A"})
	62	+	}
	63	+	if r.stats.capturesFull != 0 {
	64	+	t.Fatalf("queue not full yet: %+v", r.stats)
	65	+	}
	66	+	r.routeCapture(ports.CaptureRequest{ID: 99, Output: "A"})
	67	+	r.routeCapture(ports.CaptureRequest{ID: 100, Output: "missing"})
	68	+	if r.stats.capturesFull != 1 {
	69	+	t.Fatalf("captures full: %+v", r.stats)
	70	+	}
	71	+	if got := <-replies; got.ID != 99 \|\| got.Err == nil {
	72	+	t.Fatalf("rejected capture: %+v", got)
	73	+	}
	74	+	}
	75	+
	76	+	// The backend loop reports its longest wait on core; shorter ones keep it.
	77	+	func TestRouterNoteEventSendKeepsMax(t *testing.T) {
	78	+	r := newOutputRouter(t.Context(), nil, outputRoutes{})
	79	+	for _, d := range []time.Duration{3 * time.Millisecond, 12 * time.Millisecond, time.Millisecond} {
	80	+	r.noteEventSend(d)
	81	+	}
	82	+	if got := time.Duration(r.eventWait.Load()); got != 12*time.Millisecond {
	83	+	t.Fatalf("max wait %v, want 12ms", got)
	84	+	}
	85	+	}
	86	+
	87	+	// Every 10 s the router logs and resets its counters, and stays quiet when
	88	+	// nothing happened.
	89	+	func TestRouterLogsStatsEvery10s(t *testing.T) {
	90	+	synctest.Test(t, func(t *testing.T) {
	91	+	var out lockedBuffer
	92	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out})
	93	+	ctx, cancel := context.WithCancel(zerowrap.WithCtx(context.Background(), log))
	94	+	in := newRouteInputs()
	95	+	r := newOutputRouter(ctx, nil, in.routes())
	96	+	go r.run(ctx)
	97	+	m := newMailbox(1, ctx, func() {}, make(chan struct{}))
	98	+	if _, ok := r.add("A", m); !ok {
	99	+	t.Fatal("router stopped")
	100	+	}
	101	+	<-m.cursor
	102	+
	103	+	time.Sleep(routerStatsEvery)
	104	+	synctest.Wait()
	105	+	if got := out.String(); got != "" {
	106	+	t.Fatalf("idle router logged: %s", got)
	107	+	}
	108	+
	109	+	in.scenes <- []ports.Scene{{Output: "A", Seq: 1}}
	110	+	in.scenes <- []ports.Scene{{Output: "A", Seq: 2}}
	111	+	in.cursor <- ports.CursorChange{Shape: "text"}
	112	+	in.cursor <- ports.CursorChange{Shape: "pointer"}
	113	+	in.scenes <- nil // the router took both cursors
	114	+	r.noteEventSend(1500 * time.Microsecond)
	115	+	time.Sleep(routerStatsEvery)
	116	+	synctest.Wait()
	117	+
	118	+	lines := strings.Split(strings.TrimSpace(out.String()), "\n")
	119	+	if len(lines) != 1 {
	120	+	t.Fatalf("log lines: %q", lines)
	121	+	}
	122	+	var entry map[string]any
	123	+	if err := json.Unmarshal([]byte(lines[0]), &entry); err != nil {
	124	+	t.Fatal(err)
	125	+	}
	126	+	want := map[string]any{"message": "routing stats", "component": "router", "scenes": 2.0, "scenes_replaced": 1.0, "cursor": 2.0, "cursor_replaced": 1.0, "captures_full": 0.0, "max_event_send_ms": 1.5}
	127	+	for k, v := range want {
	128	+	if entry[k] != v {
	129	+	t.Errorf("%s = %v, want %v (%s)", k, entry[k], v, lines[0])
	130	+	}
	131	+	}
	132	+
	133	+	time.Sleep(routerStatsEvery)
	134	+	synctest.Wait()
	135	+	if n := strings.Count(out.String(), "routing stats"); n != 1 {
	136	+	t.Fatalf("counters were not reset: %d entries", n)
	137	+	}
	138	+	cancel()
	139	+	<-r.done
	140	+	})
	141	+	}

M internal/app/session.go +2 -0


@@ -342,6 +342,8 @@ func (s *session) run(ctx context.Context, inject func(chan<- ports.InputEvent))
342	342		}()
343	343		go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()
344	344		go func() { defer workers.Done(); done <- c.Run(ctx) }()
	345	+	workers.Add(1)
	346	+	go func() { defer workers.Done(); logCommandStats(ctx, c, logging.For(ctx, "core")) }()
345	347		if xdisplay != nil {
346	348		workers.Add(1)
347	349		// X11 failing leaves the Wayland session running; Run logs why.


@@ -342,6 +342,8 @@ func (s *session) run(ctx context.Context, inject func(chan<- ports.InputEvent))
342	}()	342		}()
343	go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()	343		go func() { defer workers.Done(); done <- child.Run(ctx, p.spawn) }()
344	go func() { defer workers.Done(); done <- c.Run(ctx) }()	344		go func() { defer workers.Done(); done <- c.Run(ctx) }()
		345	+	workers.Add(1)
		346	+	go func() { defer workers.Done(); logCommandStats(ctx, c, logging.For(ctx, "core")) }()
345	if xdisplay != nil {	347		if xdisplay != nil {
346	workers.Add(1)	348		workers.Add(1)
347	// X11 failing leaves the Wayland session running; Run logs why.	349		// X11 failing leaves the Wayland session running; Run logs why.

M internal/core/core.go +4 -0


@@ -1103,6 +1103,10 @@ func (c *Core) Ping(ctx context.Context) {
1103	1103		}
1104	1104		}
1105	1105
	1106	+	// CommandStats returns the command outbox counters since the last call.
	1107	+	// Any goroutine may call it.
	1108	+	func (c *Core) CommandStats() CommandStats { return c.out.takeStats() }
	1109	+
1106	1110		func (c *Core) Run(ctx context.Context) error {
1107	1111		// The sender stops with Run, not only with ctx: Run also returns on
1108	1112		// quit and on errors. Run waits for it so nothing sends after Run.


@@ -1103,6 +1103,10 @@ func (c *Core) Ping(ctx context.Context) {
1103	}	1103		}
1104	}	1104		}
1105		1105
		1106	+	// CommandStats returns the command outbox counters since the last call.
		1107	+	// Any goroutine may call it.
		1108	+	func (c *Core) CommandStats() CommandStats { return c.out.takeStats() }
		1109	+
1106	func (c *Core) Run(ctx context.Context) error {	1110		func (c *Core) Run(ctx context.Context) error {
1107	// The sender stops with Run, not only with ctx: Run also returns on	1111		// The sender stops with Run, not only with ctx: Run also returns on
1108	// quit and on errors. Run waits for it so nothing sends after Run.	1112		// quit and on errors. Run waits for it so nothing sends after Run.

M internal/core/outbox.go +59 -0


@@ -3,6 +3,7 @@ package core
3	3		import (
4	4		"context"
5	5		"sync"
	6	+	"time"
6	7
7	8		"github.com/bnema/neferwl/internal/ports"
8	9		)

@@ -34,6 +35,34 @@ type outbox struct {
34	35		// sendMu is held by whoever takes and delivers a batch, so batches
35	36		// never interleave (the sender, or flush without one).
36	37		sendMu sync.Mutex
	38	+	// stats are guarded by mu, reset by takeStats. Their times come from
	39	+	// the system clock, not core's: they measure real waits, and the sender
	40	+	// must not step a test clock.
	41	+	stats CommandStats
	42	+	}
	43	+
	44	+	// CommandStats tell whether the display holds up core's commands: how many
	45	+	// queued instead of going straight out, how long they waited, and how long
	46	+	// core itself waited (flush). They are reset when read.
	47	+	type CommandStats struct {
	48	+	// Direct and Queued count the commands sent at once and queued; Merged
	49	+	// counts motions folded into a queued one.
	50	+	Direct, Queued, Merged int
	51	+	// MaxBacklog is the longest queue seen.
	52	+	MaxBacklog uint64
	53	+	// MaxQueueWait is the longest time a queued command took to reach the
	54	+	// channel; MaxFlushWait the longest time core waited in flush (a spawn's
	55	+	// SlotsPending, or a full outbox).
	56	+	MaxQueueWait, MaxFlushWait time.Duration
	57	+	}
	58	+
	59	+	// takeStats returns the counters and starts over.
	60	+	func (o *outbox) takeStats() CommandStats {
	61	+	o.mu.Lock()
	62	+	defer o.mu.Unlock()
	63	+	s := o.stats
	64	+	o.stats = CommandStats{}
	65	+	return s
37	66		}
38	67
39	68		// outCmd is a queued command. A pointer motion is kept unboxed so that a

@@ -48,6 +77,8 @@ type outCmd struct {
48	77		motion ports.PointerMotionTo
49	78		isMotion bool
50	79		merged bool
	80	+	// at is when it queued; a merge keeps the first motion's.
	81	+	at time.Time
51	82		}
52	83
53	84		// outboxLimit bounds the commands waiting for the display. Motion merges, so

@@ -101,12 +132,15 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
101	132		case out <- e.command():
102	133		o.pushed++
103	134		o.sent = o.pushed
	135	+	o.stats.Direct++
104	136		seq := o.pushed
105	137		o.mu.Unlock()
106	138		return seq
107	139		default:
108	140		}
109	141		}
	142	+	// Only a queued command pays for the clock read.
	143	+	e.at = time.Now()
110	144		if n := len(o.queue); n > 0 && e.isMotion {
111	145		last := &o.queue[n-1]
112	146		if last.isMotion && last.motion.ID == e.motion.ID && last.secured == secured && last.state == state {

@@ -115,7 +149,9 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
115	149		e.motion.UnaccelDX += last.motion.UnaccelDX
116	150		e.motion.UnaccelDY += last.motion.UnaccelDY
117	151		e.merged = true
	152	+	e.at = last.at
118	153		*last = e
	154	+	o.stats.Merged++
119	155		seq := o.pushed
120	156		o.mu.Unlock()
121	157		return seq

@@ -123,6 +159,8 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
123	159		}
124	160		o.queue = append(o.queue, e)
125	161		o.pushed++
	162	+	o.stats.Queued++
	163	+	o.stats.MaxBacklog = max(o.stats.MaxBacklog, o.pushed-o.sent)
126	164		seq := o.pushed
127	165		o.mu.Unlock()
128	166		select {

@@ -137,6 +175,15 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
137	175		// everything else never waits. Without a running sender (core driven
138	176		// without Run), it delivers the queue itself, in order.
139	177		func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
	178	+	var start time.Time
	179	+	defer func() {
	180	+	if !start.IsZero() {
	181	+	w := time.Since(start)
	182	+	o.mu.Lock()
	183	+	o.stats.MaxFlushWait = max(o.stats.MaxFlushWait, w)
	184	+	o.mu.Unlock()
	185	+	}
	186	+	}()
140	187		for {
141	188		o.mu.Lock()
142	189		done, running := o.sent >= seq, o.running

@@ -144,6 +191,9 @@ func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq
144	191		if done {
145	192		return nil
146	193		}
	194	+	if start.IsZero() {
	195	+	start = time.Now()
	196	+	}
147	197		if !running {
148	198		o.sendMu.Lock()
149	199		ok := o.deliver(ctx, out, o.take())

@@ -240,6 +290,15 @@ func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, ba
240	290		return false
241	291		}
242	292		}
	293	+	// The batch's first entry waited longest: they queued in order, and a
	294	+	// merge keeps the first time.
	295	+	var wait time.Duration
	296	+	if len(batch) > 0 {
	297	+	wait = time.Since(batch[0].at)
	298	+	}
	299	+	o.mu.Lock()
	300	+	o.stats.MaxQueueWait = max(o.stats.MaxQueueWait, wait)
	301	+	o.mu.Unlock()
243	302		o.recycle(batch)
244	303		return true
245	304		}


@@ -3,6 +3,7 @@ package core
3	import (	3		import (
4	"context"	4		"context"
5	"sync"	5		"sync"
		6	+	"time"
6		7
7	"github.com/bnema/neferwl/internal/ports"	8		"github.com/bnema/neferwl/internal/ports"
8	)	9		)

@@ -34,6 +35,34 @@ type outbox struct {
34	// sendMu is held by whoever takes and delivers a batch, so batches	35		// sendMu is held by whoever takes and delivers a batch, so batches
35	// never interleave (the sender, or flush without one).	36		// never interleave (the sender, or flush without one).
36	sendMu sync.Mutex	37		sendMu sync.Mutex
		38	+	// stats are guarded by mu, reset by takeStats. Their times come from
		39	+	// the system clock, not core's: they measure real waits, and the sender
		40	+	// must not step a test clock.
		41	+	stats CommandStats
		42	+	}
		43	+
		44	+	// CommandStats tell whether the display holds up core's commands: how many
		45	+	// queued instead of going straight out, how long they waited, and how long
		46	+	// core itself waited (flush). They are reset when read.
		47	+	type CommandStats struct {
		48	+	// Direct and Queued count the commands sent at once and queued; Merged
		49	+	// counts motions folded into a queued one.
		50	+	Direct, Queued, Merged int
		51	+	// MaxBacklog is the longest queue seen.
		52	+	MaxBacklog uint64
		53	+	// MaxQueueWait is the longest time a queued command took to reach the
		54	+	// channel; MaxFlushWait the longest time core waited in flush (a spawn's
		55	+	// SlotsPending, or a full outbox).
		56	+	MaxQueueWait, MaxFlushWait time.Duration
		57	+	}
		58	+
		59	+	// takeStats returns the counters and starts over.
		60	+	func (o *outbox) takeStats() CommandStats {
		61	+	o.mu.Lock()
		62	+	defer o.mu.Unlock()
		63	+	s := o.stats
		64	+	o.stats = CommandStats{}
		65	+	return s
37	}	66		}
38		67
39	// outCmd is a queued command. A pointer motion is kept unboxed so that a	68		// outCmd is a queued command. A pointer motion is kept unboxed so that a

@@ -48,6 +77,8 @@ type outCmd struct {
48	motion ports.PointerMotionTo	77		motion ports.PointerMotionTo
49	isMotion bool	78		isMotion bool
50	merged bool	79		merged bool
		80	+	// at is when it queued; a merge keeps the first motion's.
		81	+	at time.Time
51	}	82		}
52		83
53	// outboxLimit bounds the commands waiting for the display. Motion merges, so	84		// outboxLimit bounds the commands waiting for the display. Motion merges, so

@@ -101,12 +132,15 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
101	case out <- e.command():	132		case out <- e.command():
102	o.pushed++	133		o.pushed++
103	o.sent = o.pushed	134		o.sent = o.pushed
		135	+	o.stats.Direct++
104	seq := o.pushed	136		seq := o.pushed
105	o.mu.Unlock()	137		o.mu.Unlock()
106	return seq	138		return seq
107	default:	139		default:
108	}	140		}
109	}	141		}
		142	+	// Only a queued command pays for the clock read.
		143	+	e.at = time.Now()
110	if n := len(o.queue); n > 0 && e.isMotion {	144		if n := len(o.queue); n > 0 && e.isMotion {
111	last := &o.queue[n-1]	145		last := &o.queue[n-1]
112	if last.isMotion && last.motion.ID == e.motion.ID && last.secured == secured && last.state == state {	146		if last.isMotion && last.motion.ID == e.motion.ID && last.secured == secured && last.state == state {

@@ -115,7 +149,9 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
115	e.motion.UnaccelDX += last.motion.UnaccelDX	149		e.motion.UnaccelDX += last.motion.UnaccelDX
116	e.motion.UnaccelDY += last.motion.UnaccelDY	150		e.motion.UnaccelDY += last.motion.UnaccelDY
117	e.merged = true	151		e.merged = true
		152	+	e.at = last.at
118	*last = e	153		*last = e
		154	+	o.stats.Merged++
119	seq := o.pushed	155		seq := o.pushed
120	o.mu.Unlock()	156		o.mu.Unlock()
121	return seq	157		return seq

@@ -123,6 +159,8 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
123	}	159		}
124	o.queue = append(o.queue, e)	160		o.queue = append(o.queue, e)
125	o.pushed++	161		o.pushed++
		162	+	o.stats.Queued++
		163	+	o.stats.MaxBacklog = max(o.stats.MaxBacklog, o.pushed-o.sent)
126	seq := o.pushed	164		seq := o.pushed
127	o.mu.Unlock()	165		o.mu.Unlock()
128	select {	166		select {

@@ -137,6 +175,15 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
137	// everything else never waits. Without a running sender (core driven	175		// everything else never waits. Without a running sender (core driven
138	// without Run), it delivers the queue itself, in order.	176		// without Run), it delivers the queue itself, in order.
139	func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {	177		func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
		178	+	var start time.Time
		179	+	defer func() {
		180	+	if !start.IsZero() {
		181	+	w := time.Since(start)
		182	+	o.mu.Lock()
		183	+	o.stats.MaxFlushWait = max(o.stats.MaxFlushWait, w)
		184	+	o.mu.Unlock()
		185	+	}
		186	+	}()
140	for {	187		for {
141	o.mu.Lock()	188		o.mu.Lock()
142	done, running := o.sent >= seq, o.running	189		done, running := o.sent >= seq, o.running

@@ -144,6 +191,9 @@ func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq
144	if done {	191		if done {
145	return nil	192		return nil
146	}	193		}
		194	+	if start.IsZero() {
		195	+	start = time.Now()
		196	+	}
147	if !running {	197		if !running {
148	o.sendMu.Lock()	198		o.sendMu.Lock()
149	ok := o.deliver(ctx, out, o.take())	199		ok := o.deliver(ctx, out, o.take())

@@ -240,6 +290,15 @@ func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, ba
240	return false	290		return false
241	}	291		}
242	}	292		}
		293	+	// The batch's first entry waited longest: they queued in order, and a
		294	+	// merge keeps the first time.
		295	+	var wait time.Duration
		296	+	if len(batch) > 0 {
		297	+	wait = time.Since(batch[0].at)
		298	+	}
		299	+	o.mu.Lock()
		300	+	o.stats.MaxQueueWait = max(o.stats.MaxQueueWait, wait)
		301	+	o.mu.Unlock()
243	o.recycle(batch)	302		o.recycle(batch)
244	return true	303		return true
245	}	304		}

M internal/core/outbox_test.go +32 -0


@@ -5,6 +5,7 @@ import (
5	5		"slices"
6	6		"testing"
7	7		"testing/synctest"
	8	+	"time"
8	9
9	10		"github.com/bnema/neferwl/internal/ports"
10	11		)

@@ -153,3 +154,34 @@ func TestOutboxQueuedDeliveryAllocations(t *testing.T) {
153	154		t.Fatalf("%v allocations per queued delivery, want at most 1", n)
154	155		}
155	156		}
	157	+
	158	+	// The stats tell direct sends from queued ones, count merges, and time how
	159	+	// long the first queued command waited for the channel and core in flush.
	160	+	func TestOutboxStats(t *testing.T) {
	161	+	synctest.Test(t, func(t *testing.T) {
	162	+	o := newOutbox()
	163	+	out := make(chan ports.ClientCommand, 1)
	164	+	ctx := context.Background()
	165	+	var none ports.SecurityState
	166	+	o.push(out, ports.FocusWindow{ID: 1}, none, false) // direct: room
	167	+	o.push(out, ports.PointerMotionTo{ID: 1, DX: 1}, none, false)
	168	+	time.Sleep(5 * time.Millisecond)
	169	+	seq := o.push(out, ports.PointerMotionTo{ID: 1, DX: 2}, none, false) // merges
	170	+	time.Sleep(5 * time.Millisecond)
	171	+	go func() {
	172	+	time.Sleep(20 * time.Millisecond)
	173	+	<-out
	174	+	}()
	175	+	if err := o.flush(ctx, out, seq); err != nil {
	176	+	t.Fatal(err)
	177	+	}
	178	+	s := o.takeStats()
	179	+	want := CommandStats{Direct: 1, Queued: 1, Merged: 1, MaxBacklog: 1, MaxQueueWait: 30 * time.Millisecond, MaxFlushWait: 20 * time.Millisecond}
	180	+	if s != want {
	181	+	t.Fatalf("stats %+v, want %+v", s, want)
	182	+	}
	183	+	if s := o.takeStats(); s != (CommandStats{}) {
	184	+	t.Fatalf("not reset: %+v", s)
	185	+	}
	186	+	})
	187	+	}


@@ -5,6 +5,7 @@ import (
5	"slices"	5		"slices"
6	"testing"	6		"testing"
7	"testing/synctest"	7		"testing/synctest"
		8	+	"time"
8		9
9	"github.com/bnema/neferwl/internal/ports"	10		"github.com/bnema/neferwl/internal/ports"
10	)	11		)

@@ -153,3 +154,34 @@ func TestOutboxQueuedDeliveryAllocations(t *testing.T) {
153	t.Fatalf("%v allocations per queued delivery, want at most 1", n)	154		t.Fatalf("%v allocations per queued delivery, want at most 1", n)
154	}	155		}
155	}	156		}
		157	+
		158	+	// The stats tell direct sends from queued ones, count merges, and time how
		159	+	// long the first queued command waited for the channel and core in flush.
		160	+	func TestOutboxStats(t *testing.T) {
		161	+	synctest.Test(t, func(t *testing.T) {
		162	+	o := newOutbox()
		163	+	out := make(chan ports.ClientCommand, 1)
		164	+	ctx := context.Background()
		165	+	var none ports.SecurityState
		166	+	o.push(out, ports.FocusWindow{ID: 1}, none, false) // direct: room
		167	+	o.push(out, ports.PointerMotionTo{ID: 1, DX: 1}, none, false)
		168	+	time.Sleep(5 * time.Millisecond)
		169	+	seq := o.push(out, ports.PointerMotionTo{ID: 1, DX: 2}, none, false) // merges
		170	+	time.Sleep(5 * time.Millisecond)
		171	+	go func() {
		172	+	time.Sleep(20 * time.Millisecond)
		173	+	<-out
		174	+	}()
		175	+	if err := o.flush(ctx, out, seq); err != nil {
		176	+	t.Fatal(err)
		177	+	}
		178	+	s := o.takeStats()
		179	+	want := CommandStats{Direct: 1, Queued: 1, Merged: 1, MaxBacklog: 1, MaxQueueWait: 30 * time.Millisecond, MaxFlushWait: 20 * time.Millisecond}
		180	+	if s != want {
		181	+	t.Fatalf("stats %+v, want %+v", s, want)
		182	+	}
		183	+	if s := o.takeStats(); s != (CommandStats{}) {
		184	+	t.Fatalf("not reset: %+v", s)
		185	+	}
		186	+	})
		187	+	}