test(wayland): cover the content handshake and a large replay

Stress the pending/flushPosted handshake with Returns racing emitContent,
check a 100-window replay lands in one batch without the output reading,
and replace the timed negative waits with a display round trip.

commit: 76b5fd77eebb4e1f7f7d088928aaa88803f555fe
author: brice · 9d ago · 2026-10-13 17:58 UTC
committer: brice · 8d ago · 2026-10-14 04:34 UTC
parent: 897ff5b
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

1 file · +124 -5

M internal/adapters/wayland/contentlink_test.go +124 -5


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

@@ -50,12 +51,16 @@ func nextBatch(t testing.T, in <-chan ports.ContentBatch) *ports.ContentBatch
50	51		}
51	52		}
52	53
53		-	func noBatch(t testing.T, in <-chan ports.ContentBatch) {
	54	+	// noBatch asserts that no batch is on the channel once the display goroutine
	55	+	// went through everything posted so far (a Return's flush included): sends
	56	+	// happen only there, so there is nothing left to wait for.
	57	+	func noBatch(t testing.T, s Server, in <-chan *ports.ContentBatch) {
54	58		t.Helper()
	59	+	s.display.Do(func() {})
55	60		select {
56	61		case b := <-in:
57	62		t.Fatalf("unexpected batch %+v", b.Items)
58		-	case <-time.After(30 * time.Millisecond):
	63	+	default:
59	64		}
60	65		}
61	66

@@ -73,7 +78,7 @@ func TestContentBatchKeepsLatestPerWindow(t *testing.T) {
73	78		if len(first.Items) != 1 \|\| first.Items[0].ID != 1 \|\| first.Items[0].Seq != 1 {
74	79		t.Fatalf("first batch %+v", first.Items)
75	80		}
76		-	noBatch(t, in) // nothing moves before the output returns a batch...
	81	+	noBatch(t, s, in) // nothing moves before the output returns a batch...
77	82		first.Reset()
78	83		ret.Return(first) // ...which sends the waiting contents
79	84		second := nextBatch(t, in)

@@ -113,7 +118,7 @@ func TestContentBatchAttachSeedsLiveWindows(t *testing.T) {
113	118		// An attach with nothing to show sends nothing.
114	119		s.display.Do(func() { clear(s.latest) })
115	120		in2, _ := s.AttachOutput("B")
116		-	noBatch(t, in2)
	121	+	noBatch(t, s, in2)
117	122		}
118	123
119	124		// An empty content reaches the attached outputs (it releases the window's

@@ -165,7 +170,7 @@ func TestContentBatchDetach(t *testing.T) {
165	170		a.Reset()
166	171		retA.Return(a)
167	172		s.publish(t, solid(1))
168		-	noBatch(t, inA)
	173	+	noBatch(t, s, inA)
169	174		if got := nextBatch(t, inB); len(got.Items) != 1 {
170	175		t.Fatalf("B batch %+v", got.Items)
171	176		}

@@ -215,3 +220,117 @@ func TestContentBatchCycleAllocations(t *testing.T) {
215	220		t.Errorf("content batch cycle: %.2f allocs, want 0", allocs)
216	221		}
217	222		}
	223	+
	224	+	// A late-attached output gets all the live windows in one batch, however
	225	+	// many there are, and AttachOutput returns without the output reading.
	226	+	func TestContentBatchAttachReplaysLargeSet(t *testing.T) {
	227	+	s := linkServer(t)
	228	+	const windows = 100
	229	+	all := make([]ports.SurfaceContent, 0, windows)
	230	+	for id := 1; id <= windows; id++ {
	231	+	all = append(all, solid(ports.WindowID(id)))
	232	+	}
	233	+	s.publish(t, all...)
	234	+	attached := make(chan (<-chan *ports.ContentBatch), 1)
	235	+	go func() { in, _ := s.AttachOutput("A"); attached <- in }()
	236	+	var in <-chan *ports.ContentBatch
	237	+	select {
	238	+	case in = <-attached:
	239	+	case <-time.After(2 * time.Second):
	240	+	t.Fatal("AttachOutput waits for the output to read")
	241	+	}
	242	+	b := nextBatch(t, in)
	243	+	ids := map[ports.WindowID]struct{}{}
	244	+	for _, c := range b.Items {
	245	+	ids[c.ID] = struct{}{}
	246	+	}
	247	+	if len(b.Items) != windows \|\| len(ids) != windows {
	248	+	t.Fatalf("replay batch holds %d items, %d distinct windows, want %d", len(b.Items), len(ids), windows)
	249	+	}
	250	+	}
	251	+
	252	+	// The output returns batches in bursts and holds some while content keeps
	253	+	// coming, so the channel is full and no batch is free when emitContent runs;
	254	+	// Returns race with it from another goroutine. Whatever the interleaving, the
	255	+	// output ends up with the newest Seq of every window (the pending/flushPosted
	256	+	// handshake never strands content). Run with -race -count=20.
	257	+	func TestContentBatchHandshakeNeverStrandsContent(t *testing.T) {
	258	+	s := linkServer(t)
	259	+	in, ret := s.AttachOutput("A")
	260	+	const windows, rounds = 4, 3000
	261	+	var last [windows + 1]atomic.Uint64
	262	+	stop, ended := make(chan struct{}), make(chan struct{})
	263	+	t.Cleanup(func() { close(stop); <-ended })
	264	+	go func() {
	265	+	defer close(ended)
	266	+	var held []*ports.ContentBatch
	267	+	giveBack := func(n int) {
	268	+	for ; n > 0 && len(held) > 0; n-- {
	269	+	b := held[0]
	270	+	held = held[1:]
	271	+	b.Reset()
	272	+	ret.Return(b)
	273	+	}
	274	+	}
	275	+	for n := 0; ; n++ {
	276	+	var b *ports.ContentBatch
	277	+	select {
	278	+	case b = <-in:
	279	+	default:
	280	+	// Nothing waiting: hand everything back, as an idle
	281	+	// output does, then block.
	282	+	giveBack(len(held))
	283	+	select {
	284	+	case b = <-in:
	285	+	case <-stop:
	286	+	return
	287	+	}
	288	+	}
	289	+	for _, c := range b.Items {
	290	+	if c.Seq > last[c.ID].Load() {
	291	+	last[c.ID].Store(c.Seq)
	292	+	}
	293	+	}
	294	+	held = append(held, b)
	295	+	if len(held) >= 2 {
	296	+	giveBack(1 + n%2) // a burst of one or two
	297	+	}
	298	+	}
	299	+	}()
	300	+	for range rounds {
	301	+	s.publishRound(windows)
	302	+	}
	303	+	var want [windows + 1]uint64
	304	+	s.display.Do(func() {
	305	+	for id := 1; id <= windows; id++ {
	306	+	want[id] = s.contentSeq[ports.WindowID(id)]
	307	+	}
	308	+	})
	309	+	for deadline := time.Now().Add(5 * time.Second); ; {
	310	+	missing := 0
	311	+	for id := 1; id <= windows; id++ {
	312	+	if last[id].Load() != want[id] {
	313	+	missing++
	314	+	}
	315	+	}
	316	+	if missing == 0 {
	317	+	return
	318	+	}
	319	+	if time.Now().After(deadline) {
	320	+	for id := 1; id <= windows; id++ {
	321	+	t.Errorf("window %d: output saw seq %d, newest is %d", id, last[id].Load(), want[id])
	322	+	}
	323	+	t.Fatal("content stranded")
	324	+	}
	325	+	time.Sleep(time.Millisecond)
	326	+	}
	327	+	}
	328	+
	329	+	// publishRound publishes one content per window in one display turn.
	330	+	func (s *Server) publishRound(windows int) {
	331	+	s.display.Do(func() {
	332	+	for id := 1; id <= windows; id++ {
	333	+	s.emitContent(solid(ports.WindowID(id)), damage{full: true})
	334	+	}
	335	+	})
	336	+	}


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

@@ -50,12 +51,16 @@ func nextBatch(t testing.T, in <-chan ports.ContentBatch) *ports.ContentBatch
50		}	51		}
51		}	52		}
52			53
53	-	func noBatch(t testing.T, in <-chan ports.ContentBatch) {	54	+	// noBatch asserts that no batch is on the channel once the display goroutine
			55	+	// went through everything posted so far (a Return's flush included): sends
			56	+	// happen only there, so there is nothing left to wait for.
			57	+	func noBatch(t testing.T, s Server, in <-chan *ports.ContentBatch) {
54		t.Helper()	58		t.Helper()
			59	+	s.display.Do(func() {})
55		select {	60		select {
56		case b := <-in:	61		case b := <-in:
57		t.Fatalf("unexpected batch %+v", b.Items)	62		t.Fatalf("unexpected batch %+v", b.Items)
58	-	case <-time.After(30 * time.Millisecond):	63	+	default:
59		}	64		}
60		}	65		}
61			66

@@ -73,7 +78,7 @@ func TestContentBatchKeepsLatestPerWindow(t *testing.T) {
73		if len(first.Items) != 1 \|\| first.Items[0].ID != 1 \|\| first.Items[0].Seq != 1 {	78		if len(first.Items) != 1 \|\| first.Items[0].ID != 1 \|\| first.Items[0].Seq != 1 {
74		t.Fatalf("first batch %+v", first.Items)	79		t.Fatalf("first batch %+v", first.Items)
75		}	80		}
76	-	noBatch(t, in) // nothing moves before the output returns a batch...	81	+	noBatch(t, s, in) // nothing moves before the output returns a batch...
77		first.Reset()	82		first.Reset()
78		ret.Return(first) // ...which sends the waiting contents	83		ret.Return(first) // ...which sends the waiting contents
79		second := nextBatch(t, in)	84		second := nextBatch(t, in)

@@ -113,7 +118,7 @@ func TestContentBatchAttachSeedsLiveWindows(t *testing.T) {
113		// An attach with nothing to show sends nothing.	118		// An attach with nothing to show sends nothing.
114		s.display.Do(func() { clear(s.latest) })	119		s.display.Do(func() { clear(s.latest) })
115		in2, _ := s.AttachOutput("B")	120		in2, _ := s.AttachOutput("B")
116	-	noBatch(t, in2)	121	+	noBatch(t, s, in2)
117		}	122		}
118			123
119		// An empty content reaches the attached outputs (it releases the window's	124		// An empty content reaches the attached outputs (it releases the window's

@@ -165,7 +170,7 @@ func TestContentBatchDetach(t *testing.T) {
165		a.Reset()	170		a.Reset()
166		retA.Return(a)	171		retA.Return(a)
167		s.publish(t, solid(1))	172		s.publish(t, solid(1))
168	-	noBatch(t, inA)	173	+	noBatch(t, s, inA)
169		if got := nextBatch(t, inB); len(got.Items) != 1 {	174		if got := nextBatch(t, inB); len(got.Items) != 1 {
170		t.Fatalf("B batch %+v", got.Items)	175		t.Fatalf("B batch %+v", got.Items)
171		}	176		}

@@ -215,3 +220,117 @@ func TestContentBatchCycleAllocations(t *testing.T) {
215		t.Errorf("content batch cycle: %.2f allocs, want 0", allocs)	220		t.Errorf("content batch cycle: %.2f allocs, want 0", allocs)
216		}	221		}
217		}	222		}
			223	+
			224	+	// A late-attached output gets all the live windows in one batch, however
			225	+	// many there are, and AttachOutput returns without the output reading.
			226	+	func TestContentBatchAttachReplaysLargeSet(t *testing.T) {
			227	+	s := linkServer(t)
			228	+	const windows = 100
			229	+	all := make([]ports.SurfaceContent, 0, windows)
			230	+	for id := 1; id <= windows; id++ {
			231	+	all = append(all, solid(ports.WindowID(id)))
			232	+	}
			233	+	s.publish(t, all...)
			234	+	attached := make(chan (<-chan *ports.ContentBatch), 1)
			235	+	go func() { in, _ := s.AttachOutput("A"); attached <- in }()
			236	+	var in <-chan *ports.ContentBatch
			237	+	select {
			238	+	case in = <-attached:
			239	+	case <-time.After(2 * time.Second):
			240	+	t.Fatal("AttachOutput waits for the output to read")
			241	+	}
			242	+	b := nextBatch(t, in)
			243	+	ids := map[ports.WindowID]struct{}{}
			244	+	for _, c := range b.Items {
			245	+	ids[c.ID] = struct{}{}
			246	+	}
			247	+	if len(b.Items) != windows \|\| len(ids) != windows {
			248	+	t.Fatalf("replay batch holds %d items, %d distinct windows, want %d", len(b.Items), len(ids), windows)
			249	+	}
			250	+	}
			251	+
			252	+	// The output returns batches in bursts and holds some while content keeps
			253	+	// coming, so the channel is full and no batch is free when emitContent runs;
			254	+	// Returns race with it from another goroutine. Whatever the interleaving, the
			255	+	// output ends up with the newest Seq of every window (the pending/flushPosted
			256	+	// handshake never strands content). Run with -race -count=20.
			257	+	func TestContentBatchHandshakeNeverStrandsContent(t *testing.T) {
			258	+	s := linkServer(t)
			259	+	in, ret := s.AttachOutput("A")
			260	+	const windows, rounds = 4, 3000
			261	+	var last [windows + 1]atomic.Uint64
			262	+	stop, ended := make(chan struct{}), make(chan struct{})
			263	+	t.Cleanup(func() { close(stop); <-ended })
			264	+	go func() {
			265	+	defer close(ended)
			266	+	var held []*ports.ContentBatch
			267	+	giveBack := func(n int) {
			268	+	for ; n > 0 && len(held) > 0; n-- {
			269	+	b := held[0]
			270	+	held = held[1:]
			271	+	b.Reset()
			272	+	ret.Return(b)
			273	+	}
			274	+	}
			275	+	for n := 0; ; n++ {
			276	+	var b *ports.ContentBatch
			277	+	select {
			278	+	case b = <-in:
			279	+	default:
			280	+	// Nothing waiting: hand everything back, as an idle
			281	+	// output does, then block.
			282	+	giveBack(len(held))
			283	+	select {
			284	+	case b = <-in:
			285	+	case <-stop:
			286	+	return
			287	+	}
			288	+	}
			289	+	for _, c := range b.Items {
			290	+	if c.Seq > last[c.ID].Load() {
			291	+	last[c.ID].Store(c.Seq)
			292	+	}
			293	+	}
			294	+	held = append(held, b)
			295	+	if len(held) >= 2 {
			296	+	giveBack(1 + n%2) // a burst of one or two
			297	+	}
			298	+	}
			299	+	}()
			300	+	for range rounds {
			301	+	s.publishRound(windows)
			302	+	}
			303	+	var want [windows + 1]uint64
			304	+	s.display.Do(func() {
			305	+	for id := 1; id <= windows; id++ {
			306	+	want[id] = s.contentSeq[ports.WindowID(id)]
			307	+	}
			308	+	})
			309	+	for deadline := time.Now().Add(5 * time.Second); ; {
			310	+	missing := 0
			311	+	for id := 1; id <= windows; id++ {
			312	+	if last[id].Load() != want[id] {
			313	+	missing++
			314	+	}
			315	+	}
			316	+	if missing == 0 {
			317	+	return
			318	+	}
			319	+	if time.Now().After(deadline) {
			320	+	for id := 1; id <= windows; id++ {
			321	+	t.Errorf("window %d: output saw seq %d, newest is %d", id, last[id].Load(), want[id])
			322	+	}
			323	+	t.Fatal("content stranded")
			324	+	}
			325	+	time.Sleep(time.Millisecond)
			326	+	}
			327	+	}
			328	+
			329	+	// publishRound publishes one content per window in one display turn.
			330	+	func (s *Server) publishRound(windows int) {
			331	+	s.display.Do(func() {
			332	+	for id := 1; id <= windows; id++ {
			333	+	s.emitContent(solid(ports.WindowID(id)), damage{full: true})
			334	+	}
			335	+	})
			336	+	}