7c51167 fix(stats): time each queued command and keep stall periods in the log · neferwl

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


@@ -113,7 +113,8 @@ const commandStatsInterval = 10 * time.Second
113	113
114	114		// logCommandStats reports command path latency every interval until ctx or the
115	115		// display ends, so a live log shows whether the display goroutine delays input.
116		-	// A period without applied commands logs nothing.
	116	+	// Only a period with every counter at zero logs nothing: a stalled display
	117	+	// applies nothing but still fills the ring.
117	118		func (s *Server) logCommandStats(ctx context.Context) {
118	119		t := time.NewTicker(commandStatsInterval)
119	120		defer t.Stop()

@@ -132,7 +133,7 @@ func (s *Server) logCommandStats(ctx context.Context) {
132	133		func (s *Server) reportCommandStats() {
133	134		rs := s.commands.takeStats()
134	135		applied, latency := s.commandStats.take()
135		-	if applied == 0 {
	136	+	if applied == 0 && latency == 0 && rs == (ringStats{}) {
136	137		return
137	138		}
138	139		s.log.Info().

 // logCommandStats reports command path latency every interval until ctx or the
 // display ends, so a live log shows whether the display goroutine delays input.
-// A period without applied commands logs nothing.
+// Only a period with every counter at zero logs nothing: a stalled display
+// applies nothing but still fills the ring.
 func (s *Server) logCommandStats(ctx context.Context) {
 	t := time.NewTicker(commandStatsInterval)
 	defer t.Stop()
 func (s *Server) reportCommandStats() {
 	rs := s.commands.takeStats()
 	applied, latency := s.commandStats.take()
-	if applied == 0 {
+	if applied == 0 && latency == 0 && rs == (ringStats{}) {
 		return
 	}
 	s.log.Info().

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


@@ -290,3 +290,22 @@ func TestReportCommandStats(t *testing.T) {
290	290		t.Fatalf("stats not reset: %s", out.String())
291	291		}
292	292		}
	293	+
	294	+	// A stalled display applies nothing but fills its ring: the period still
	295	+	// logs, so a long stall is not hidden.
	296	+	func TestReportCommandStatsDuringStall(t *testing.T) {
	297	+	var out lockedBuffer
	298	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out}).WithField("component", "wayland")
	299	+	h := newAdmissionHarnessWith(t, 1, Options{}, log)
	300	+	s := h.s
	301	+	s.commands.mu.Lock()
	302	+	s.commands.stats.maxLen = 5
	303	+	s.commands.mu.Unlock()
	304	+	s.reportCommandStats()
	305	+	got := out.String()
	306	+	for _, want := range []string{`"message":"command stats"`, `"commands":0`, `"max_queued":5`} {
	307	+	if !strings.Contains(got, want) {
	308	+	t.Fatalf("log lacks %s: %s", want, got)
	309	+	}
	310	+	}
	311	+	}


@@ -290,3 +290,22 @@ func TestReportCommandStats(t *testing.T) {
290	t.Fatalf("stats not reset: %s", out.String())	290		t.Fatalf("stats not reset: %s", out.String())
291	}	291		}
292	}	292		}
		293	+
		294	+	// A stalled display applies nothing but fills its ring: the period still
		295	+	// logs, so a long stall is not hidden.
		296	+	func TestReportCommandStatsDuringStall(t *testing.T) {
		297	+	var out lockedBuffer
		298	+	log := zerowrap.New(zerowrap.Config{Level: "info", Format: "json", Output: &out}).WithField("component", "wayland")
		299	+	h := newAdmissionHarnessWith(t, 1, Options{}, log)
		300	+	s := h.s
		301	+	s.commands.mu.Lock()
		302	+	s.commands.stats.maxLen = 5
		303	+	s.commands.mu.Unlock()
		304	+	s.reportCommandStats()
		305	+	got := out.String()
		306	+	for _, want := range []string{`"message":"command stats"`, `"commands":0`, `"max_queued":5`} {
		307	+	if !strings.Contains(got, want) {
		308	+	t.Fatalf("log lacks %s: %s", want, got)
		309	+	}
		310	+	}
		311	+	}

M internal/adapters/wayland/postring_test.go +13 -5


@@ -166,9 +166,10 @@ func TestPostRingPushStopsWithDisplay(t *testing.T) {
166	166		func TestPostRingStatsPushBlock(t *testing.T) {
167	167		s, _, _, _, _, _ := presentServer(t)
168	168		r := newPostRing(s.display, 1, func(int) {})
169		-	release := make(chan struct{})
	169	+	// A failed assertion still frees the display before server shutdown.
	170	+	release := releaseOnce(t)
170	171		stalled := make(chan struct{})
171		-	s.display.Post(func() { close(stalled); <-release })
	172	+	s.display.Post(func() { close(stalled); <-release.ch })
172	173		<-stalled
173	174		r.Push(1)
174	175		if st := r.takeStats(); st.maxLen != 1 \|\| st.maxPushBlock != 0 {

@@ -177,10 +178,13 @@ func TestPostRingStatsPushBlock(t *testing.T) {
177	178		done := make(chan struct{})
178	179		go func() { defer close(done); r.Push(2) }() // full: waits for the drain
179	180		waitPushWaiting(t)
180		-	close(release)
	181	+	seen := time.Now()
	182	+	time.Sleep(2 * time.Millisecond) // the block lasts at least this
	183	+	held := time.Since(seen)
	184	+	release.do()
181	185		<-done
182		-	if st := r.takeStats(); st.maxPushBlock <= 0 {
183		-	t.Fatalf("stats = %+v, want a recorded push block", st)
	186	+	if st := r.takeStats(); st.maxPushBlock < held {
	187	+	t.Fatalf("stats = %+v, want a push block of at least %v", st, held)
184	188		}
185	189		}
186	190

@@ -189,7 +193,11 @@ func TestPostRingStatsPushBlock(t *testing.T) {
189	193		func waitPushWaiting(t *testing.T) {
190	194		t.Helper()
191	195		buf := make([]byte, 1<<20)
	196	+	deadline := time.Now().Add(5 * time.Second)
192	197		for {
	198	+	if time.Now().After(deadline) {
	199	+	t.Fatalf("no goroutine waits in postRing.Push:\n%s", buf[:runtime.Stack(buf, true)])
	200	+	}
193	201		dump := string(buf[:runtime.Stack(buf, true)])
194	202		for _, g := range strings.Split(dump, "\n\n") {
195	203		if strings.Contains(g, "[select") && strings.Contains(g, "postRing") && strings.Contains(g, ").Push(") {

 func TestPostRingStatsPushBlock(t *testing.T) {
 	s, _, _, _, _, _ := presentServer(t)
 	r := newPostRing(s.display, 1, func(int) {})
-	release := make(chan struct{})
+	// A failed assertion still frees the display before server shutdown.
+	release := releaseOnce(t)
 	stalled := make(chan struct{})
-	s.display.Post(func() { close(stalled); <-release })
+	s.display.Post(func() { close(stalled); <-release.ch })
 	<-stalled
 	r.Push(1)
 	if st := r.takeStats(); st.maxLen != 1 || st.maxPushBlock != 0 {
 	done := make(chan struct{})
 	go func() { defer close(done); r.Push(2) }() // full: waits for the drain
 	waitPushWaiting(t)
-	close(release)
+	seen := time.Now()
+	time.Sleep(2 * time.Millisecond) // the block lasts at least this
+	held := time.Since(seen)
+	release.do()
 	<-done
-	if st := r.takeStats(); st.maxPushBlock <= 0 {
+	if st := r.takeStats(); st.maxPushBlock < held {
-		t.Fatalf("stats = %+v, want a recorded push block", st)
+		t.Fatalf("stats = %+v, want a push block of at least %v", st, held)
 	}
 }
 func waitPushWaiting(t *testing.T) {
 	t.Helper()
 	buf := make([]byte, 1<<20)
+	deadline := time.Now().Add(5 * time.Second)
 	for {
+		if time.Now().After(deadline) {
+			t.Fatalf("no goroutine waits in postRing.Push:\n%s", buf[:runtime.Stack(buf, true)])
+		}
 		dump := string(buf[:runtime.Stack(buf, true)])
 		for _, g := range strings.Split(dump, "\n\n") {
 			if strings.Contains(g, "[select") && strings.Contains(g, "postRing") && strings.Contains(g, ").Push(") {

M internal/app/corestats.go +3 -2


@@ -14,7 +14,8 @@ const commandStatsEvery = 10 * time.Second
14	14
15	15		// logCommandStats logs core's command counters every commandStatsEvery
16	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.
	17	+	// held core's commands up; a period with no counter set logs nothing (a wait
	18	+	// that ends after the tick still logs, with no new command).
18	19		func logCommandStats(ctx context.Context, c *core.Core, log zerowrap.Logger) {
19	20		t := time.NewTicker(commandStatsEvery)
20	21		defer t.Stop()

@@ -25,7 +26,7 @@ func logCommandStats(ctx context.Context, c *core.Core, log zerowrap.Logger) {
25	26		case <-t.C:
26	27		}
27	28		s := c.CommandStats()
28		-	if s.Direct == 0 && s.Queued == 0 {
	29	+	if s == (core.CommandStats{}) {
29	30		continue
30	31		}
31	32		log.Info().Int("direct", s.Direct).Int("queued", s.Queued).Int("merged", s.Merged).

 // logCommandStats logs core's command counters every commandStatsEvery
 // until ctx ends. A queued command or a flush wait means the display goroutine
-// held core's commands up; an idle period logs nothing.
+// held core's commands up; a period with no counter set logs nothing (a wait
+// that ends after the tick still logs, with no new command).
 func logCommandStats(ctx context.Context, c *core.Core, log zerowrap.Logger) {
 	t := time.NewTicker(commandStatsEvery)
 	defer t.Stop()
 		case <-t.C:
 		}
 		s := c.CommandStats()
-		if s.Direct == 0 && s.Queued == 0 {
+		if s == (core.CommandStats{}) {
 			continue
 		}
 		log.Info().Int("direct", s.Direct).Int("queued", s.Queued).Int("merged", s.Merged).

M internal/core/outbox.go +4 -6


@@ -283,18 +283,16 @@ func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
283	283		// deliver sends batch to out in order and recycles it; false when ctx ended
284	284		// first (the rest of the batch is dropped with the session).
285	285		func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, batch []outCmd) bool {
	286	+	// Each entry's wait ends with its own send: a send that blocks midway
	287	+	// must not count against the entries already out.
	288	+	var wait time.Duration
286	289		for i := range batch {
287	290		select {
288	291		case out <- batch[i].command():
289	292		case <-ctx.Done():
290	293		return false
291	294		}
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)
	295	+	wait = max(wait, time.Since(batch[i].at))
298	296		}
299	297		o.mu.Lock()
300	298		o.stats.MaxQueueWait = max(o.stats.MaxQueueWait, wait)

 // deliver sends batch to out in order and recycles it; false when ctx ended
 // first (the rest of the batch is dropped with the session).
 func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, batch []outCmd) bool {
+	// Each entry's wait ends with its own send: a send that blocks midway
+	// must not count against the entries already out.
+	var wait time.Duration
 	for i := range batch {
 		select {
 		case out <- batch[i].command():
 		case <-ctx.Done():
 			return false
 		}
-	}
+		wait = max(wait, time.Since(batch[i].at))
-	// The batch's first entry waited longest: they queued in order, and a
-	// merge keeps the first time.
-	var wait time.Duration
-	if len(batch) > 0 {
-		wait = time.Since(batch[0].at)
 	}
 	o.mu.Lock()
 	o.stats.MaxQueueWait = max(o.stats.MaxQueueWait, wait)

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


@@ -185,3 +185,28 @@ func TestOutboxStats(t *testing.T) {
185	185		}
186	186		})
187	187		}
	188	+
	189	+	// A send that blocks midway through a batch counts against the entries still
	190	+	// waiting, not the ones already out: the wait is per entry.
	191	+	func TestOutboxQueueWaitIsPerEntry(t *testing.T) {
	192	+	synctest.Test(t, func(t *testing.T) {
	193	+	o := newOutbox()
	194	+	out := make(chan ports.ClientCommand)
	195	+	var none ports.SecurityState
	196	+	o.push(out, ports.FocusWindow{ID: 1}, none, false) // queued at 0
	197	+	time.Sleep(9 * time.Millisecond)
	198	+	o.push(out, ports.FocusWindow{ID: 2}, none, false) // queued at 9 ms
	199	+	go func() {
	200	+	time.Sleep(time.Millisecond)
	201	+	<-out // first out at 10 ms
	202	+	time.Sleep(10 * time.Millisecond)
	203	+	<-out // second out at 20 ms
	204	+	}()
	205	+	if !o.deliver(context.Background(), out, o.take()) {
	206	+	t.Fatal("deliver stopped")
	207	+	}
	208	+	if got := o.takeStats().MaxQueueWait; got != 11*time.Millisecond {
	209	+	t.Fatalf("max queue wait %v, want 11ms (the second entry's)", got)
	210	+	}
	211	+	})
	212	+	}


@@ -185,3 +185,28 @@ func TestOutboxStats(t *testing.T) {
185	}	185		}
186	})	186		})
187	}	187		}
		188	+
		189	+	// A send that blocks midway through a batch counts against the entries still
		190	+	// waiting, not the ones already out: the wait is per entry.
		191	+	func TestOutboxQueueWaitIsPerEntry(t *testing.T) {
		192	+	synctest.Test(t, func(t *testing.T) {
		193	+	o := newOutbox()
		194	+	out := make(chan ports.ClientCommand)
		195	+	var none ports.SecurityState
		196	+	o.push(out, ports.FocusWindow{ID: 1}, none, false) // queued at 0
		197	+	time.Sleep(9 * time.Millisecond)
		198	+	o.push(out, ports.FocusWindow{ID: 2}, none, false) // queued at 9 ms
		199	+	go func() {
		200	+	time.Sleep(time.Millisecond)
		201	+	<-out // first out at 10 ms
		202	+	time.Sleep(10 * time.Millisecond)
		203	+	<-out // second out at 20 ms
		204	+	}()
		205	+	if !o.deliver(context.Background(), out, o.take()) {
		206	+	t.Fatal("deliver stopped")
		207	+	}
		208	+	if got := o.takeStats().MaxQueueWait; got != 11*time.Millisecond {
		209	+	t.Fatalf("max queue wait %v, want 11ms (the second entry's)", got)
		210	+	}
		211	+	})
		212	+	}