1package core
2
3import (
4 "context"
5 "slices"
6 "testing"
7 "testing/synctest"
8 "time"
9
10 "github.com/bnema/neferwl/internal/ports"
11)
12
13func wantOutbox(t *testing.T, got []ports.ClientCommand, want ...ports.ClientCommand) {
14 t.Helper()
15 if !slices.Equal(got, want) {
16 t.Fatalf("commands %#v, want %#v", got, want)
17 }
18}
19
20// takeAll returns what the sender would deliver, in order.
21func takeAll(o *outbox) []ports.ClientCommand {
22 var got []ports.ClientCommand
23 for _, e := range o.take() {
24 got = append(got, e.command())
25 }
26 return got
27}
28
29// Only adjacent motions for the same window merge: the position is the
30// latest, the deltas add up. Everything else keeps its place.
31func TestOutboxMergesAdjacentMotionOnly(t *testing.T) {
32 o := newOutbox()
33 // Nobody receives, so every command queues.
34 out := make(chan ports.ClientCommand)
35 push := func(v ports.ClientCommand) { o.push(out, v, ports.SecurityState{}, false) }
36 push(ports.PointerMotionTo{ID: 1, X: 1, Y: 1, DX: 1, DY: 2, UnaccelDX: 3, UnaccelDY: 4})
37 push(ports.PointerMotionTo{ID: 1, X: 2, Y: 2, DX: 10, DY: 20, UnaccelDX: 30, UnaccelDY: 40})
38 push(ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true})
39 push(ports.PointerMotionTo{ID: 1, X: 3, Y: 3, DX: 5})
40 push(ports.PointerMotionTo{ID: 2, X: 4, Y: 4, DX: 7})
41 push(ports.FocusWindow{ID: 2})
42 push(ports.PointerMotionTo{ID: 2, X: 5, Y: 5, DX: 1})
43 push(ports.PointerMotionTo{ID: 2, X: 6, Y: 6, DX: 1})
44 want := []ports.ClientCommand{
45 ports.PointerMotionTo{ID: 1, X: 2, Y: 2, DX: 11, DY: 22, UnaccelDX: 33, UnaccelDY: 44},
46 ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true},
47 ports.PointerMotionTo{ID: 1, X: 3, Y: 3, DX: 5},
48 ports.PointerMotionTo{ID: 2, X: 4, Y: 4, DX: 7},
49 ports.FocusWindow{ID: 2},
50 ports.PointerMotionTo{ID: 2, X: 6, Y: 6, DX: 2},
51 }
52 wantOutbox(t, takeAll(&o), want...)
53}
54
55// Wrapped motions merge only under an identical owner envelope, and keep it.
56func TestOutboxMergesSecuredMotionPerEnvelope(t *testing.T) {
57 o := newOutbox()
58 out := make(chan ports.ClientCommand)
59 a := ports.SecurityState{Generation: 1}
60 b := ports.SecurityState{Generation: 2}
61 o.push(out, ports.PointerMotionTo{ID: 1, X: 1, DX: 1}, a, true)
62 o.push(out, ports.PointerMotionTo{ID: 1, X: 2, DX: 2}, a, true)
63 o.push(out, ports.PointerMotionTo{ID: 1, X: 3, DX: 4}, b, true)
64 o.push(out, ports.PointerMotionTo{ID: 1, X: 4, DX: 8}, ports.SecurityState{}, false)
65 wantOutbox(t, takeAll(&o),
66 ports.SecurityCommand{State: a, Command: ports.PointerMotionTo{ID: 1, X: 2, DX: 3}},
67 ports.SecurityCommand{State: b, Command: ports.PointerMotionTo{ID: 1, X: 3, DX: 4}},
68 ports.PointerMotionTo{ID: 1, X: 4, DX: 8},
69 )
70}
71
72// With room in the channel and nothing queued, a command goes straight out.
73func TestOutboxSendsDirectlyWhileDisplayKeepsUp(t *testing.T) {
74 o := newOutbox()
75 out := make(chan ports.ClientCommand, 4)
76 o.push(out, ports.PointerMotionTo{ID: 1, DX: 1}, ports.SecurityState{}, false)
77 o.push(out, ports.PointerMotionTo{ID: 1, DX: 2}, ports.SecurityState{}, false)
78 if len(out) != 2 {
79 t.Fatalf("%d commands sent, want 2 unmerged", len(out))
80 }
81 if got := o.take(); len(got) != 0 {
82 t.Fatalf("queued %d", len(got))
83 }
84}
85
86// A display that reads nothing for long fills the outbox to its limit, then
87// core waits for room: the queue never grows past it, nothing is dropped.
88func TestOutboxIsBounded(t *testing.T) {
89 synctest.Test(t, func(t *testing.T) {
90 out := make(chan ports.ClientCommand)
91 c := &Core{out: newOutbox(), ch: Channels{Commands: out}}
92 c.out.start()
93 ctx, cancel := context.WithCancel(context.Background())
94 defer cancel()
95 go c.out.run(ctx, out)
96 done := make(chan error, 1)
97 go func() {
98 for i := range outboxLimit + 10 {
99 // Not mergeable: each takes a place.
100 if err := c.command(ctx, ports.FocusWindow{ID: WindowID(i)}); err != nil {
101 done <- err
102 return
103 }
104 }
105 done <- nil
106 }()
107 synctest.Wait()
108 if n := c.out.backlog(); n != outboxLimit+1 {
109 t.Fatalf("backlog %d, want the limit and the one waiting", n)
110 }
111 select {
112 case err := <-done:
113 t.Fatalf("pushed past the limit: %v", err)
114 default:
115 }
116 for i := range outboxLimit + 10 {
117 if got := (<-out).(ports.FocusWindow); got.ID != WindowID(i) {
118 t.Fatalf("command %d is %+v", i, got)
119 }
120 }
121 if err := <-done; err != nil {
122 t.Fatal(err)
123 }
124 })
125}
126
127// Queuing a command and delivering it allocates nothing once warm: the
128// batches are recycled and flush's progress signal is reused.
129func TestOutboxQueuedDeliveryAllocations(t *testing.T) {
130 o := newOutbox()
131 full := make(chan ports.ClientCommand, 1)
132 full <- ports.FocusWindow{}
133 sink := make(chan ports.ClientCommand, 8)
134 var cmd ports.ClientCommand = ports.FocusWindow{ID: 1}
135 var state ports.SecurityState
136 ctx := context.Background()
137 cycle := func() {
138 // full has no room: push queues, even for a secured command.
139 o.push(full, cmd, state, false)
140 o.push(full, cmd, state, true)
141 o.deliver(ctx, sink, o.take())
142 <-sink
143 <-sink
144 select {
145 case <-o.progress:
146 default:
147 t.Fatal("delivery left no progress signal")
148 }
149 }
150 cycle()
151 // The secured delivery boxes its envelope once; the failed direct send
152 // must not box another.
153 if n := testing.AllocsPerRun(100, cycle); n > 1 {
154 t.Fatalf("%v allocations per queued delivery, want at most 1", n)
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.
160func 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}
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.
191func 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}