refactor(core): queue client commands in an outbox instead of blocking

core sent each client command on the wayland channel and waited when it
was full, so a busy display goroutine stalled keyboard handling, scene
publishing and frame wakes. command() now queues in an outbox and a
sender goroutine owned by Run delivers it in order. Adjacent pointer
motions of one window under the same envelope merge with their deltas
summed; a command sends at once when nothing is queued and the channel
has room.

commit: d7dc4d5c8991b8e9be6ae4dddff64cbe79fe614c
author: brice · 4d ago · 2026-10-18 05:21 UTC
parent: b5629dc
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +496 -68

M internal/core/core.go +34 -12


@@ -142,7 +142,9 @@ type Core struct {
142	142		// (clientgesture.go).
143	143		gesture clientGesture
144	144		// ping carries watchdog probes (Ping); Run answers each one.
145		-	ping chan chan struct{}
	145	+	ping chan chan struct{}
	146	+	// out queues the commands for c.ch.Commands; Run's sender drains it.
	147	+	out outbox
146	148		frameC <-chan time.Time
147	149		frameTimer ports.Timer
148	150		// shots is the snapshot of what the screens show before an action

@@ -405,7 +407,7 @@ func New(cfg ports.Config, ch Channels, opts Options) (*Core, error) {
405	407		return nil, fmt.Errorf("scenes, layouts, constraints, state and workspaces must have capacity 1")
406	408		}
407	409		// A placeholder screen holds windows until the first output arrives.
408		-	c := &Core{inputKeys: map[string]bool{}, slots: map[slotKey]slotState{}, placement: newSpawnPlacement(), windows: newWindowRegistry(), popups: map[WindowID]popupState{}, ch: ch, opts: opts, pressed: map[string]bool{}, buttons: map[uint32]bool{}, swallow: map[uint32]bool{}, configures: newConfigures(), ping: make(chan chan struct{})}
	410	+	c := &Core{inputKeys: map[string]bool{}, slots: map[slotKey]slotState{}, placement: newSpawnPlacement(), windows: newWindowRegistry(), popups: map[WindowID]popupState{}, ch: ch, opts: opts, pressed: map[string]bool{}, buttons: map[uint32]bool{}, swallow: map[uint32]bool{}, configures: newConfigures(), ping: make(chan chan struct{}), out: newOutbox()}
409	411		c.screens = []*screen{{mon: newMonitorWithIDs("", "", &c.nextWorkspaceID), scale: 1, cfgScale: 1}}
410	412		if err := c.apply(cfg); err != nil {
411	413		return nil, err

@@ -442,22 +444,28 @@ func (c *Core) now() time.Time {
442	444		return time.Now()
443	445		}
444	446
	447	+	// command queues v for the wayland goroutine and never blocks: a busy
	448	+	// display must not stall keyboard, scenes or frame wakes. The sender
	449	+	// goroutine of Run delivers the queue in order.
445	450		func (c *Core) command(ctx context.Context, v ports.ClientCommand) error {
	451	+	if err := ctx.Err(); err != nil {
	452	+	return err
	453	+	}
446	454		if c.inputEpochChanged() {
447	455		return errSecurityChanged
448	456		}
449		-	if c.opts.Security != nil {
450		-	v = ports.SecurityCommand{State: c.security, Command: v}
451		-	}
452		-	select {
453		-	case <-ctx.Done():
454		-	return ctx.Err()
455		-	case c.ch.Commands <- v:
456		-	if c.inputEpochChanged() {
457		-	return errSecurityChanged
458		-	}
	457	+	if c.ch.Commands == nil {
	458	+	// Nobody receives: queueing would only grow without bound.
459	459		return nil
460	460		}
	461	+	c.out.push(c.ch.Commands, v, c.security, c.opts.Security != nil)
	462	+	// The queued command keeps the epoch it was made under, so a display
	463	+	// that has moved on drops it. Report the change all the same: the rest
	464	+	// of the event was decided under the old epoch and must not go on.
	465	+	if c.inputEpochChanged() {
	466	+	return errSecurityChanged
	467	+	}
	468	+	return nil
461	469		}
462	470
463	471		// clientRect is the placement minus the border on its inset sides.

@@ -1059,6 +1067,20 @@ func (c *Core) Ping(ctx context.Context) {
1059	1067		}
1060	1068
1061	1069		func (c *Core) Run(ctx context.Context) error {
	1070	+	// The sender stops with Run, not only with ctx: Run also returns on
	1071	+	// quit and on errors. Run waits for it so nothing sends after Run.
	1072	+	sendCtx, stopSender := context.WithCancel(ctx)
	1073	+	senderDone := make(chan struct{})
	1074	+	go func() {
	1075	+	defer close(senderDone)
	1076	+	if c.ch.Commands != nil {
	1077	+	c.out.run(sendCtx, c.ch.Commands)
	1078	+	}
	1079	+	}()
	1080	+	defer func() {
	1081	+	stopSender()
	1082	+	<-senderDone
	1083	+	}()
1062	1084		c.syncSecurity()
1063	1085		defer c.stopFrame()
1064	1086		defer c.stopCaptureTimer()

 	// (clientgesture.go).
 	gesture clientGesture
 	// ping carries watchdog probes (Ping); Run answers each one.
-	ping       chan chan struct{}
+	ping chan chan struct{}
+	// out queues the commands for c.ch.Commands; Run's sender drains it.
+	out        outbox
 	frameC     <-chan time.Time
 	frameTimer ports.Timer
 	// shots is the snapshot of what the screens show before an action
 		return nil, fmt.Errorf("scenes, layouts, constraints, state and workspaces must have capacity 1")
 	}
 	// A placeholder screen holds windows until the first output arrives.
-	c := &Core{inputKeys: map[string]bool{}, slots: map[slotKey]*slotState{}, placement: newSpawnPlacement(), windows: newWindowRegistry(), popups: map[WindowID]*popupState{}, ch: ch, opts: opts, pressed: map[string]bool{}, buttons: map[uint32]bool{}, swallow: map[uint32]bool{}, configures: newConfigures(), ping: make(chan chan struct{})}
+	c := &Core{inputKeys: map[string]bool{}, slots: map[slotKey]*slotState{}, placement: newSpawnPlacement(), windows: newWindowRegistry(), popups: map[WindowID]*popupState{}, ch: ch, opts: opts, pressed: map[string]bool{}, buttons: map[uint32]bool{}, swallow: map[uint32]bool{}, configures: newConfigures(), ping: make(chan chan struct{}), out: newOutbox()}
 	c.screens = []*screen{{mon: newMonitorWithIDs("", "", &c.nextWorkspaceID), scale: 1, cfgScale: 1}}
 	if err := c.apply(cfg); err != nil {
 		return nil, err
 	return time.Now()
 }
+// command queues v for the wayland goroutine and never blocks: a busy
+// display must not stall keyboard, scenes or frame wakes. The sender
+// goroutine of Run delivers the queue in order.
 func (c *Core) command(ctx context.Context, v ports.ClientCommand) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
 	if c.inputEpochChanged() {
 		return errSecurityChanged
 	}
-	if c.opts.Security != nil {
+	if c.ch.Commands == nil {
-		v = ports.SecurityCommand{State: c.security, Command: v}
+		// Nobody receives: queueing would only grow without bound.
-	}
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	case c.ch.Commands <- v:
-		if c.inputEpochChanged() {
-			return errSecurityChanged
-		}
 		return nil
 	}
+	c.out.push(c.ch.Commands, v, c.security, c.opts.Security != nil)
+	// The queued command keeps the epoch it was made under, so a display
+	// that has moved on drops it. Report the change all the same: the rest
+	// of the event was decided under the old epoch and must not go on.
+	if c.inputEpochChanged() {
+		return errSecurityChanged
+	}
+	return nil
 }
 // clientRect is the placement minus the border on its inset sides.
 }
 func (c *Core) Run(ctx context.Context) error {
+	// The sender stops with Run, not only with ctx: Run also returns on
+	// quit and on errors. Run waits for it so nothing sends after Run.
+	sendCtx, stopSender := context.WithCancel(ctx)
+	senderDone := make(chan struct{})
+	go func() {
+		defer close(senderDone)
+		if c.ch.Commands != nil {
+			c.out.run(sendCtx, c.ch.Commands)
+		}
+	}()
+	defer func() {
+		stopSender()
+		<-senderDone
+	}()
 	c.syncSecurity()
 	defer c.stopFrame()
 	defer c.stopCaptureTimer()

A internal/core/outbox.go +142 -0

@@ -0,0 +1,142 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"context"
	5	+	"sync"
	6	+
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	// outbox carries commands from the owner goroutine to the sender goroutine
	11	+	// without ever blocking the owner. It is a transport queue, not window
	12	+	// state: the mutex guards only the slices. Commands leave in the order they
	13	+	// were queued; the only merge is of adjacent pointer motions (see push).
	14	+	//
	15	+	// The wayland display goroutine can be busy for a while. Sending from the
	16	+	// owner would stall keyboard, scenes and frame wakes behind it, so the owner
	17	+	// queues and the sender goroutine does the blocking send.
	18	+	type outbox struct {
	19	+	// wake is signalled (never blocking) after each push.
	20	+	wake chan struct{}
	21	+
	22	+	mu sync.Mutex
	23	+	queue, spare []outCmd
	24	+	// busy is set while the sender holds a batch it has not finished
	25	+	// sending: a command pushed then must queue behind it.
	26	+	busy bool
	27	+	}
	28	+
	29	+	// outCmd is a queued command. A pointer motion is kept unboxed so that a
	30	+	// merge boxes the sum once, when it leaves, not at every merge.
	31	+	type outCmd struct {
	32	+	cmd ports.ClientCommand
	33	+	// state is the owner envelope; secured is false without a security gate.
	34	+	state ports.SecurityState
	35	+	secured bool
	36	+	// motion is cmd when it is a pointer motion. merged means cmd is stale
	37	+	// and motion carries the sum.
	38	+	motion ports.PointerMotionTo
	39	+	isMotion bool
	40	+	merged bool
	41	+	}
	42	+
	43	+	func newOutbox() outbox { return outbox{wake: make(chan struct{}, 1)} }
	44	+
	45	+	// command returns what the sender delivers.
	46	+	func (o *outCmd) command() ports.ClientCommand {
	47	+	v := o.cmd
	48	+	if o.merged {
	49	+	v = o.motion
	50	+	}
	51	+	if o.secured {
	52	+	return ports.SecurityCommand{State: o.state, Command: v}
	53	+	}
	54	+	return v
	55	+	}
	56	+
	57	+	// push delivers v, wrapped in an envelope for state when secured. With
	58	+	// nothing queued or in flight and room in out it sends at once, which keeps
	59	+	// the display's latency when it keeps up; otherwise it queues v. A pointer
	60	+	// motion for the same window and envelope as the one queued last replaces
	61	+	// it: the position is the latest and the relative deltas add up, so locked
	62	+	// pointers (games) lose no movement. Nothing else is merged or reordered:
	63	+	// buttons, focus, keys and gestures depend on their order.
	64	+	func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, state ports.SecurityState, secured bool) {
	65	+	e := outCmd{cmd: v, state: state, secured: secured}
	66	+	e.motion, e.isMotion = v.(ports.PointerMotionTo)
	67	+	o.mu.Lock()
	68	+	if !o.busy && len(o.queue) == 0 {
	69	+	// A send that cannot block is safe under the lock.
	70	+	select {
	71	+	case out <- e.command():
	72	+	o.mu.Unlock()
	73	+	return
	74	+	default:
	75	+	}
	76	+	}
	77	+	if n := len(o.queue); n > 0 && e.isMotion {
	78	+	last := &o.queue[n-1]
	79	+	if last.isMotion && last.motion.ID == e.motion.ID && last.secured == secured && last.state == state {
	80	+	e.motion.DX += last.motion.DX
	81	+	e.motion.DY += last.motion.DY
	82	+	e.motion.UnaccelDX += last.motion.UnaccelDX
	83	+	e.motion.UnaccelDY += last.motion.UnaccelDY
	84	+	e.merged = true
	85	+	*last = e
	86	+	o.mu.Unlock()
	87	+	return
	88	+	}
	89	+	}
	90	+	o.queue = append(o.queue, e)
	91	+	o.mu.Unlock()
	92	+	select {
	93	+	case o.wake <- struct{}{}:
	94	+	default:
	95	+	}
	96	+	}
	97	+
	98	+	// take returns the queued commands and leaves an empty queue. The caller
	99	+	// hands the batch back with recycle once it has sent it.
	100	+	func (o *outbox) take() []outCmd {
	101	+	o.mu.Lock()
	102	+	batch := o.queue
	103	+	o.queue, o.spare = o.spare[:0], nil
	104	+	o.busy = len(batch) > 0
	105	+	o.mu.Unlock()
	106	+	return batch
	107	+	}
	108	+
	109	+	// recycle keeps the batch's storage for the next take.
	110	+	func (o *outbox) recycle(batch []outCmd) {
	111	+	// Do not retain what the commands reference in the spare batch.
	112	+	clear(batch)
	113	+	o.mu.Lock()
	114	+	o.spare = batch[:0]
	115	+	o.busy = false
	116	+	o.mu.Unlock()
	117	+	}
	118	+
	119	+	// run delivers the queue to out in order until ctx ends. A command whose
	120	+	// send is waiting is not merged any more, but what queues behind it is.
	121	+	func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
	122	+	for {
	123	+	batch := o.take()
	124	+	if len(batch) == 0 {
	125	+	o.recycle(batch)
	126	+	select {
	127	+	case <-o.wake:
	128	+	case <-ctx.Done():
	129	+	return
	130	+	}
	131	+	continue
	132	+	}
	133	+	for i := range batch {
	134	+	select {
	135	+	case out <- batch[i].command():
	136	+	case <-ctx.Done():
	137	+	return
	138	+	}
	139	+	}
	140	+	o.recycle(batch)
	141	+	}
	142	+	}

@@ -0,0 +1,142 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"context"
	5	+	"sync"
	6	+
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	// outbox carries commands from the owner goroutine to the sender goroutine
	11	+	// without ever blocking the owner. It is a transport queue, not window
	12	+	// state: the mutex guards only the slices. Commands leave in the order they
	13	+	// were queued; the only merge is of adjacent pointer motions (see push).
	14	+	//
	15	+	// The wayland display goroutine can be busy for a while. Sending from the
	16	+	// owner would stall keyboard, scenes and frame wakes behind it, so the owner
	17	+	// queues and the sender goroutine does the blocking send.
	18	+	type outbox struct {
	19	+	// wake is signalled (never blocking) after each push.
	20	+	wake chan struct{}
	21	+
	22	+	mu sync.Mutex
	23	+	queue, spare []outCmd
	24	+	// busy is set while the sender holds a batch it has not finished
	25	+	// sending: a command pushed then must queue behind it.
	26	+	busy bool
	27	+	}
	28	+
	29	+	// outCmd is a queued command. A pointer motion is kept unboxed so that a
	30	+	// merge boxes the sum once, when it leaves, not at every merge.
	31	+	type outCmd struct {
	32	+	cmd ports.ClientCommand
	33	+	// state is the owner envelope; secured is false without a security gate.
	34	+	state ports.SecurityState
	35	+	secured bool
	36	+	// motion is cmd when it is a pointer motion. merged means cmd is stale
	37	+	// and motion carries the sum.
	38	+	motion ports.PointerMotionTo
	39	+	isMotion bool
	40	+	merged bool
	41	+	}
	42	+
	43	+	func newOutbox() outbox { return outbox{wake: make(chan struct{}, 1)} }
	44	+
	45	+	// command returns what the sender delivers.
	46	+	func (o *outCmd) command() ports.ClientCommand {
	47	+	v := o.cmd
	48	+	if o.merged {
	49	+	v = o.motion
	50	+	}
	51	+	if o.secured {
	52	+	return ports.SecurityCommand{State: o.state, Command: v}
	53	+	}
	54	+	return v
	55	+	}
	56	+
	57	+	// push delivers v, wrapped in an envelope for state when secured. With
	58	+	// nothing queued or in flight and room in out it sends at once, which keeps
	59	+	// the display's latency when it keeps up; otherwise it queues v. A pointer
	60	+	// motion for the same window and envelope as the one queued last replaces
	61	+	// it: the position is the latest and the relative deltas add up, so locked
	62	+	// pointers (games) lose no movement. Nothing else is merged or reordered:
	63	+	// buttons, focus, keys and gestures depend on their order.
	64	+	func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, state ports.SecurityState, secured bool) {
	65	+	e := outCmd{cmd: v, state: state, secured: secured}
	66	+	e.motion, e.isMotion = v.(ports.PointerMotionTo)
	67	+	o.mu.Lock()
	68	+	if !o.busy && len(o.queue) == 0 {
	69	+	// A send that cannot block is safe under the lock.
	70	+	select {
	71	+	case out <- e.command():
	72	+	o.mu.Unlock()
	73	+	return
	74	+	default:
	75	+	}
	76	+	}
	77	+	if n := len(o.queue); n > 0 && e.isMotion {
	78	+	last := &o.queue[n-1]
	79	+	if last.isMotion && last.motion.ID == e.motion.ID && last.secured == secured && last.state == state {
	80	+	e.motion.DX += last.motion.DX
	81	+	e.motion.DY += last.motion.DY
	82	+	e.motion.UnaccelDX += last.motion.UnaccelDX
	83	+	e.motion.UnaccelDY += last.motion.UnaccelDY
	84	+	e.merged = true
	85	+	*last = e
	86	+	o.mu.Unlock()
	87	+	return
	88	+	}
	89	+	}
	90	+	o.queue = append(o.queue, e)
	91	+	o.mu.Unlock()
	92	+	select {
	93	+	case o.wake <- struct{}{}:
	94	+	default:
	95	+	}
	96	+	}
	97	+
	98	+	// take returns the queued commands and leaves an empty queue. The caller
	99	+	// hands the batch back with recycle once it has sent it.
	100	+	func (o *outbox) take() []outCmd {
	101	+	o.mu.Lock()
	102	+	batch := o.queue
	103	+	o.queue, o.spare = o.spare[:0], nil
	104	+	o.busy = len(batch) > 0
	105	+	o.mu.Unlock()
	106	+	return batch
	107	+	}
	108	+
	109	+	// recycle keeps the batch's storage for the next take.
	110	+	func (o *outbox) recycle(batch []outCmd) {
	111	+	// Do not retain what the commands reference in the spare batch.
	112	+	clear(batch)
	113	+	o.mu.Lock()
	114	+	o.spare = batch[:0]
	115	+	o.busy = false
	116	+	o.mu.Unlock()
	117	+	}
	118	+
	119	+	// run delivers the queue to out in order until ctx ends. A command whose
	120	+	// send is waiting is not merged any more, but what queues behind it is.
	121	+	func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
	122	+	for {
	123	+	batch := o.take()
	124	+	if len(batch) == 0 {
	125	+	o.recycle(batch)
	126	+	select {
	127	+	case <-o.wake:
	128	+	case <-ctx.Done():
	129	+	return
	130	+	}
	131	+	continue
	132	+	}
	133	+	for i := range batch {
	134	+	select {
	135	+	case out <- batch[i].command():
	136	+	case <-ctx.Done():
	137	+	return
	138	+	}
	139	+	}
	140	+	o.recycle(batch)
	141	+	}
	142	+	}

A internal/core/outbox_run_test.go +204 -0

@@ -0,0 +1,204 @@
	1	+	package core_test
	2	+
	3	+	import (
	4	+	"context"
	5	+	"testing"
	6	+	"testing/synctest"
	7	+
	8	+	"github.com/bnema/neferwl/internal/core"
	9	+	"github.com/bnema/neferwl/internal/ports"
	10	+	)
	11	+
	12	+	// outboxRig runs a core whose command channel nobody reads until the test
	13	+	// does. Run and the sender live in a synctest bubble, so synctest.Wait
	14	+	// returns once both are idle.
	15	+	type outboxRig struct {
	16	+	input chan ports.InputEvent
	17	+	client chan ports.ClientEvent
	18	+	output chan ports.OutputEvent
	19	+	commands chan ports.ClientCommand
	20	+	scenes chan []ports.Scene
	21	+	}
	22	+
	23	+	func startOutboxRig(t testing.T, commandCap int) outboxRig {
	24	+	t.Helper()
	25	+	cfg := scrollDefaults()
	26	+	cfg.Border.Width = 0
	27	+	cfg.Outputs = []ports.OutputConfig{{Name: "DP-2", Scale: 2}}
	28	+	cfg.Binds["Cmd+minus"] = "scale-down"
	29	+	r := &outboxRig{
	30	+	input: make(chan ports.InputEvent),
	31	+	client: make(chan ports.ClientEvent),
	32	+	output: make(chan ports.OutputEvent),
	33	+	commands: make(chan ports.ClientCommand, commandCap),
	34	+	scenes: make(chan []ports.Scene, 1),
	35	+	}
	36	+	c, err := core.New(cfg, core.Channels{Input: r.input, Client: r.client, Output: r.output, Commands: r.commands, Scenes: r.scenes}, core.Options{})
	37	+	if err != nil {
	38	+	t.Fatal(err)
	39	+	}
	40	+	ctx, cancel := context.WithCancel(context.Background())
	41	+	done := make(chan error, 1)
	42	+	go func() { done <- c.Run(ctx) }()
	43	+	t.Cleanup(func() {
	44	+	cancel()
	45	+	// Run waits for its sender, so a sender stuck on the full channel
	46	+	// would keep this from returning.
	47	+	if err := receive(t, done); err != nil {
	48	+	t.Error(err)
	49	+	}
	50	+	})
	51	+	r.output <- ports.OutputAdded{Info: ports.OutputInfo{Name: "DP-2", Width: 5120, Height: 2160}}
	52	+	r.client <- ports.WindowMapped{ID: 1}
	53	+	synctest.Wait()
	54	+	// Settle: drop the scenes and the commands of the map.
	55	+	r.drain()
	56	+	return r
	57	+	}
	58	+
	59	+	// drain empties the scene and command channels and lets the sender refill
	60	+	// them from its queue until nothing moves.
	61	+	func (r *outboxRig) drain() []ports.ClientCommand {
	62	+	var got []ports.ClientCommand
	63	+	for {
	64	+	synctest.Wait()
	65	+	select {
	66	+	case <-r.scenes:
	67	+	default:
	68	+	}
	69	+	select {
	70	+	case v := <-r.commands:
	71	+	got = append(got, v)
	72	+	default:
	73	+	synctest.Wait()
	74	+	if len(r.commands) == 0 {
	75	+	return got
	76	+	}
	77	+	}
	78	+	}
	79	+	}
	80	+
	81	+	// A display that reads nothing must not stall core: motion keeps coming, and
	82	+	// a bind after it still runs and publishes its scene.
	83	+	func TestCommandsNeverBlockCore(t *testing.T) {
	84	+	synctest.Test(t, func(t *testing.T) {
	85	+	r := startOutboxRig(t, 32)
	86	+	const n = 10000
	87	+	for i := range n {
	88	+	// Each send returns only once Run is back in its select.
	89	+	r.input <- ports.PointerMotion{X: 500 + float64(i%100), Y: 200, DX: 1, DY: 2, UnaccelDX: 3, UnaccelDY: 4}
	90	+	}
	91	+	r.input <- ports.KeyEvent{Keysym: "minus", Keycode: 12, Mods: ports.ModSuper, Pressed: true}
	92	+	synctest.Wait()
	93	+	select {
	94	+	case set := <-r.scenes:
	95	+	if set[0].Scale != 5.0/3 {
	96	+	t.Fatalf("scale %v, want 5/3", set[0].Scale)
	97	+	}
	98	+	default:
	99	+	t.Fatal("no scene: core is blocked on commands")
	100	+	}
	101	+	// Once the display reads again, it gets everything in order, with
	102	+	// the motion merged and no movement lost.
	103	+	var dx, dy, ux, uy float64
	104	+	var motions int
	105	+	var scaled bool
	106	+	for _, v := range r.drain() {
	107	+	switch v := v.(type) {
	108	+	case ports.PointerMotionTo:
	109	+	motions++
	110	+	dx, dy, ux, uy = dx+v.DX, dy+v.DY, ux+v.UnaccelDX, uy+v.UnaccelDY
	111	+	case ports.SetOutputs:
	112	+	scaled = v.Outputs[0].Scale == 5.0/3
	113	+	}
	114	+	}
	115	+	if dx != n \|\| dy != 2n \|\| ux != 3n \|\| uy != 4*n {
	116	+	t.Fatalf("lost movement: %v %v %v %v", dx, dy, ux, uy)
	117	+	}
	118	+	if motions > 40 {
	119	+	t.Fatalf("%d motions reached the display, want them merged while it was away", motions)
	120	+	}
	121	+	if !scaled {
	122	+	t.Fatal("scale change not sent after the motions")
	123	+	}
	124	+	})
	125	+	}
	126	+
	127	+	// Commands reach the display in the order core made them. Only adjacent
	128	+	// motions of one window merge.
	129	+	func TestCommandsKeepOrderAndMergeAdjacentMotion(t *testing.T) {
	130	+	synctest.Test(t, func(t *testing.T) {
	131	+	r := startOutboxRig(t, 0)
	132	+	// No reader: the sender holds the first command, which is in flight
	133	+	// and no longer merges, and the rest queues behind it.
	134	+	motion := func(x, dx float64) {
	135	+	r.input <- ports.PointerMotion{X: x, Y: 200, DX: dx, UnaccelDX: dx}
	136	+	synctest.Wait()
	137	+	}
	138	+	// How the first commands split between the sender's batch and the
	139	+	// queue depends on when it wakes. The first key is a barrier: what
	140	+	// is pushed after the owner handled it never merges with what
	141	+	// came before, so the rest is deterministic.
	142	+	motion(500, 1)
	143	+	r.input <- ports.KeyEvent{Keysym: "y", Keycode: 21, Pressed: true}
	144	+	synctest.Wait()
	145	+	motion(501, 2)
	146	+	motion(502, 4)
	147	+	r.input <- ports.PointerButton{Button: 0x110, Pressed: true}
	148	+	synctest.Wait()
	149	+	motion(503, 8)
	150	+	r.input <- ports.KeyEvent{Keysym: "x", Keycode: 45, Pressed: true}
	151	+	synctest.Wait()
	152	+	motion(504, 16)
	153	+	motion(505, 32)
	154	+	var got []ports.ClientCommand
	155	+	behind := false
	156	+	for _, v := range r.drainUnbuffered() {
	157	+	switch v.(type) {
	158	+	case ports.UserActivity, ports.SetOutputs:
	159	+	continue
	160	+	}
	161	+	if k, ok := v.(ports.ForwardKey); ok && k.Key.Keysym == "y" {
	162	+	behind = true
	163	+	continue
	164	+	}
	165	+	if behind {
	166	+	got = append(got, v)
	167	+	}
	168	+	}
	169	+	want := []ports.ClientCommand{
	170	+	ports.PointerMotionTo{ID: 1, X: 502, Y: 200, DX: 6, UnaccelDX: 6},
	171	+	ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true, Grab: true},
	172	+	ports.PointerMotionTo{ID: 1, X: 503, Y: 200, DX: 8, UnaccelDX: 8},
	173	+	ports.ForwardKey{ID: 1, Key: ports.KeyEvent{Keysym: "x", Keycode: 45, Pressed: true}},
	174	+	ports.PointerMotionTo{ID: 1, X: 505, Y: 200, DX: 48, UnaccelDX: 48},
	175	+	}
	176	+	wantOrder(t, got, want)
	177	+	})
	178	+	}
	179	+
	180	+	// drainUnbuffered receives what the sender hands over one by one.
	181	+	func (r *outboxRig) drainUnbuffered() []ports.ClientCommand {
	182	+	var got []ports.ClientCommand
	183	+	for {
	184	+	synctest.Wait()
	185	+	select {
	186	+	case v := <-r.commands:
	187	+	got = append(got, v)
	188	+	default:
	189	+	return got
	190	+	}
	191	+	}
	192	+	}
	193	+
	194	+	func wantOrder(t *testing.T, got, want []ports.ClientCommand) {
	195	+	t.Helper()
	196	+	if len(got) != len(want) {
	197	+	t.Fatalf("got %d commands, want %d:\n%#v", len(got), len(want), got)
	198	+	}
	199	+	for i := range want {
	200	+	if got[i] != want[i] {
	201	+	t.Errorf("command %d: got %#v, want %#v", i, got[i], want[i])
	202	+	}
	203	+	}
	204	+	}

@@ -0,0 +1,204 @@
	1	+	package core_test
	2	+
	3	+	import (
	4	+	"context"
	5	+	"testing"
	6	+	"testing/synctest"
	7	+
	8	+	"github.com/bnema/neferwl/internal/core"
	9	+	"github.com/bnema/neferwl/internal/ports"
	10	+	)
	11	+
	12	+	// outboxRig runs a core whose command channel nobody reads until the test
	13	+	// does. Run and the sender live in a synctest bubble, so synctest.Wait
	14	+	// returns once both are idle.
	15	+	type outboxRig struct {
	16	+	input chan ports.InputEvent
	17	+	client chan ports.ClientEvent
	18	+	output chan ports.OutputEvent
	19	+	commands chan ports.ClientCommand
	20	+	scenes chan []ports.Scene
	21	+	}
	22	+
	23	+	func startOutboxRig(t testing.T, commandCap int) outboxRig {
	24	+	t.Helper()
	25	+	cfg := scrollDefaults()
	26	+	cfg.Border.Width = 0
	27	+	cfg.Outputs = []ports.OutputConfig{{Name: "DP-2", Scale: 2}}
	28	+	cfg.Binds["Cmd+minus"] = "scale-down"
	29	+	r := &outboxRig{
	30	+	input: make(chan ports.InputEvent),
	31	+	client: make(chan ports.ClientEvent),
	32	+	output: make(chan ports.OutputEvent),
	33	+	commands: make(chan ports.ClientCommand, commandCap),
	34	+	scenes: make(chan []ports.Scene, 1),
	35	+	}
	36	+	c, err := core.New(cfg, core.Channels{Input: r.input, Client: r.client, Output: r.output, Commands: r.commands, Scenes: r.scenes}, core.Options{})
	37	+	if err != nil {
	38	+	t.Fatal(err)
	39	+	}
	40	+	ctx, cancel := context.WithCancel(context.Background())
	41	+	done := make(chan error, 1)
	42	+	go func() { done <- c.Run(ctx) }()
	43	+	t.Cleanup(func() {
	44	+	cancel()
	45	+	// Run waits for its sender, so a sender stuck on the full channel
	46	+	// would keep this from returning.
	47	+	if err := receive(t, done); err != nil {
	48	+	t.Error(err)
	49	+	}
	50	+	})
	51	+	r.output <- ports.OutputAdded{Info: ports.OutputInfo{Name: "DP-2", Width: 5120, Height: 2160}}
	52	+	r.client <- ports.WindowMapped{ID: 1}
	53	+	synctest.Wait()
	54	+	// Settle: drop the scenes and the commands of the map.
	55	+	r.drain()
	56	+	return r
	57	+	}
	58	+
	59	+	// drain empties the scene and command channels and lets the sender refill
	60	+	// them from its queue until nothing moves.
	61	+	func (r *outboxRig) drain() []ports.ClientCommand {
	62	+	var got []ports.ClientCommand
	63	+	for {
	64	+	synctest.Wait()
	65	+	select {
	66	+	case <-r.scenes:
	67	+	default:
	68	+	}
	69	+	select {
	70	+	case v := <-r.commands:
	71	+	got = append(got, v)
	72	+	default:
	73	+	synctest.Wait()
	74	+	if len(r.commands) == 0 {
	75	+	return got
	76	+	}
	77	+	}
	78	+	}
	79	+	}
	80	+
	81	+	// A display that reads nothing must not stall core: motion keeps coming, and
	82	+	// a bind after it still runs and publishes its scene.
	83	+	func TestCommandsNeverBlockCore(t *testing.T) {
	84	+	synctest.Test(t, func(t *testing.T) {
	85	+	r := startOutboxRig(t, 32)
	86	+	const n = 10000
	87	+	for i := range n {
	88	+	// Each send returns only once Run is back in its select.
	89	+	r.input <- ports.PointerMotion{X: 500 + float64(i%100), Y: 200, DX: 1, DY: 2, UnaccelDX: 3, UnaccelDY: 4}
	90	+	}
	91	+	r.input <- ports.KeyEvent{Keysym: "minus", Keycode: 12, Mods: ports.ModSuper, Pressed: true}
	92	+	synctest.Wait()
	93	+	select {
	94	+	case set := <-r.scenes:
	95	+	if set[0].Scale != 5.0/3 {
	96	+	t.Fatalf("scale %v, want 5/3", set[0].Scale)
	97	+	}
	98	+	default:
	99	+	t.Fatal("no scene: core is blocked on commands")
	100	+	}
	101	+	// Once the display reads again, it gets everything in order, with
	102	+	// the motion merged and no movement lost.
	103	+	var dx, dy, ux, uy float64
	104	+	var motions int
	105	+	var scaled bool
	106	+	for _, v := range r.drain() {
	107	+	switch v := v.(type) {
	108	+	case ports.PointerMotionTo:
	109	+	motions++
	110	+	dx, dy, ux, uy = dx+v.DX, dy+v.DY, ux+v.UnaccelDX, uy+v.UnaccelDY
	111	+	case ports.SetOutputs:
	112	+	scaled = v.Outputs[0].Scale == 5.0/3
	113	+	}
	114	+	}
	115	+	if dx != n \|\| dy != 2n \|\| ux != 3n \|\| uy != 4*n {
	116	+	t.Fatalf("lost movement: %v %v %v %v", dx, dy, ux, uy)
	117	+	}
	118	+	if motions > 40 {
	119	+	t.Fatalf("%d motions reached the display, want them merged while it was away", motions)
	120	+	}
	121	+	if !scaled {
	122	+	t.Fatal("scale change not sent after the motions")
	123	+	}
	124	+	})
	125	+	}
	126	+
	127	+	// Commands reach the display in the order core made them. Only adjacent
	128	+	// motions of one window merge.
	129	+	func TestCommandsKeepOrderAndMergeAdjacentMotion(t *testing.T) {
	130	+	synctest.Test(t, func(t *testing.T) {
	131	+	r := startOutboxRig(t, 0)
	132	+	// No reader: the sender holds the first command, which is in flight
	133	+	// and no longer merges, and the rest queues behind it.
	134	+	motion := func(x, dx float64) {
	135	+	r.input <- ports.PointerMotion{X: x, Y: 200, DX: dx, UnaccelDX: dx}
	136	+	synctest.Wait()
	137	+	}
	138	+	// How the first commands split between the sender's batch and the
	139	+	// queue depends on when it wakes. The first key is a barrier: what
	140	+	// is pushed after the owner handled it never merges with what
	141	+	// came before, so the rest is deterministic.
	142	+	motion(500, 1)
	143	+	r.input <- ports.KeyEvent{Keysym: "y", Keycode: 21, Pressed: true}
	144	+	synctest.Wait()
	145	+	motion(501, 2)
	146	+	motion(502, 4)
	147	+	r.input <- ports.PointerButton{Button: 0x110, Pressed: true}
	148	+	synctest.Wait()
	149	+	motion(503, 8)
	150	+	r.input <- ports.KeyEvent{Keysym: "x", Keycode: 45, Pressed: true}
	151	+	synctest.Wait()
	152	+	motion(504, 16)
	153	+	motion(505, 32)
	154	+	var got []ports.ClientCommand
	155	+	behind := false
	156	+	for _, v := range r.drainUnbuffered() {
	157	+	switch v.(type) {
	158	+	case ports.UserActivity, ports.SetOutputs:
	159	+	continue
	160	+	}
	161	+	if k, ok := v.(ports.ForwardKey); ok && k.Key.Keysym == "y" {
	162	+	behind = true
	163	+	continue
	164	+	}
	165	+	if behind {
	166	+	got = append(got, v)
	167	+	}
	168	+	}
	169	+	want := []ports.ClientCommand{
	170	+	ports.PointerMotionTo{ID: 1, X: 502, Y: 200, DX: 6, UnaccelDX: 6},
	171	+	ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true, Grab: true},
	172	+	ports.PointerMotionTo{ID: 1, X: 503, Y: 200, DX: 8, UnaccelDX: 8},
	173	+	ports.ForwardKey{ID: 1, Key: ports.KeyEvent{Keysym: "x", Keycode: 45, Pressed: true}},
	174	+	ports.PointerMotionTo{ID: 1, X: 505, Y: 200, DX: 48, UnaccelDX: 48},
	175	+	}
	176	+	wantOrder(t, got, want)
	177	+	})
	178	+	}
	179	+
	180	+	// drainUnbuffered receives what the sender hands over one by one.
	181	+	func (r *outboxRig) drainUnbuffered() []ports.ClientCommand {
	182	+	var got []ports.ClientCommand
	183	+	for {
	184	+	synctest.Wait()
	185	+	select {
	186	+	case v := <-r.commands:
	187	+	got = append(got, v)
	188	+	default:
	189	+	return got
	190	+	}
	191	+	}
	192	+	}
	193	+
	194	+	func wantOrder(t *testing.T, got, want []ports.ClientCommand) {
	195	+	t.Helper()
	196	+	if len(got) != len(want) {
	197	+	t.Fatalf("got %d commands, want %d:\n%#v", len(got), len(want), got)
	198	+	}
	199	+	for i := range want {
	200	+	if got[i] != want[i] {
	201	+	t.Errorf("command %d: got %#v, want %#v", i, got[i], want[i])
	202	+	}
	203	+	}
	204	+	}

A internal/core/outbox_test.go +81 -0

@@ -0,0 +1,81 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"slices"
	5	+	"testing"
	6	+
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	func wantOutbox(t *testing.T, got []ports.ClientCommand, want ...ports.ClientCommand) {
	11	+	t.Helper()
	12	+	if !slices.Equal(got, want) {
	13	+	t.Fatalf("commands %#v, want %#v", got, want)
	14	+	}
	15	+	}
	16	+
	17	+	// takeAll returns what the sender would deliver, in order.
	18	+	func takeAll(o *outbox) []ports.ClientCommand {
	19	+	var got []ports.ClientCommand
	20	+	for _, e := range o.take() {
	21	+	got = append(got, e.command())
	22	+	}
	23	+	return got
	24	+	}
	25	+
	26	+	// Only adjacent motions for the same window merge: the position is the
	27	+	// latest, the deltas add up. Everything else keeps its place.
	28	+	func TestOutboxMergesAdjacentMotionOnly(t *testing.T) {
	29	+	o := newOutbox()
	30	+	// Nobody receives, so every command queues.
	31	+	out := make(chan ports.ClientCommand)
	32	+	push := func(v ports.ClientCommand) { o.push(out, v, ports.SecurityState{}, false) }
	33	+	push(ports.PointerMotionTo{ID: 1, X: 1, Y: 1, DX: 1, DY: 2, UnaccelDX: 3, UnaccelDY: 4})
	34	+	push(ports.PointerMotionTo{ID: 1, X: 2, Y: 2, DX: 10, DY: 20, UnaccelDX: 30, UnaccelDY: 40})
	35	+	push(ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true})
	36	+	push(ports.PointerMotionTo{ID: 1, X: 3, Y: 3, DX: 5})
	37	+	push(ports.PointerMotionTo{ID: 2, X: 4, Y: 4, DX: 7})
	38	+	push(ports.FocusWindow{ID: 2})
	39	+	push(ports.PointerMotionTo{ID: 2, X: 5, Y: 5, DX: 1})
	40	+	push(ports.PointerMotionTo{ID: 2, X: 6, Y: 6, DX: 1})
	41	+	want := []ports.ClientCommand{
	42	+	ports.PointerMotionTo{ID: 1, X: 2, Y: 2, DX: 11, DY: 22, UnaccelDX: 33, UnaccelDY: 44},
	43	+	ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true},
	44	+	ports.PointerMotionTo{ID: 1, X: 3, Y: 3, DX: 5},
	45	+	ports.PointerMotionTo{ID: 2, X: 4, Y: 4, DX: 7},
	46	+	ports.FocusWindow{ID: 2},
	47	+	ports.PointerMotionTo{ID: 2, X: 6, Y: 6, DX: 2},
	48	+	}
	49	+	wantOutbox(t, takeAll(&o), want...)
	50	+	}
	51	+
	52	+	// Wrapped motions merge only under an identical owner envelope, and keep it.
	53	+	func TestOutboxMergesSecuredMotionPerEnvelope(t *testing.T) {
	54	+	o := newOutbox()
	55	+	out := make(chan ports.ClientCommand)
	56	+	a := ports.SecurityState{Generation: 1}
	57	+	b := ports.SecurityState{Generation: 2}
	58	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 1, DX: 1}, a, true)
	59	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 2, DX: 2}, a, true)
	60	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 3, DX: 4}, b, true)
	61	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 4, DX: 8}, ports.SecurityState{}, false)
	62	+	wantOutbox(t, takeAll(&o),
	63	+	ports.SecurityCommand{State: a, Command: ports.PointerMotionTo{ID: 1, X: 2, DX: 3}},
	64	+	ports.SecurityCommand{State: b, Command: ports.PointerMotionTo{ID: 1, X: 3, DX: 4}},
	65	+	ports.PointerMotionTo{ID: 1, X: 4, DX: 8},
	66	+	)
	67	+	}
	68	+
	69	+	// With room in the channel and nothing queued, a command goes straight out.
	70	+	func TestOutboxSendsDirectlyWhileDisplayKeepsUp(t *testing.T) {
	71	+	o := newOutbox()
	72	+	out := make(chan ports.ClientCommand, 4)
	73	+	o.push(out, ports.PointerMotionTo{ID: 1, DX: 1}, ports.SecurityState{}, false)
	74	+	o.push(out, ports.PointerMotionTo{ID: 1, DX: 2}, ports.SecurityState{}, false)
	75	+	if len(out) != 2 {
	76	+	t.Fatalf("%d commands sent, want 2 unmerged", len(out))
	77	+	}
	78	+	if got := o.take(); len(got) != 0 {
	79	+	t.Fatalf("queued %d", len(got))
	80	+	}
	81	+	}

@@ -0,0 +1,81 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"slices"
	5	+	"testing"
	6	+
	7	+	"github.com/bnema/neferwl/internal/ports"
	8	+	)
	9	+
	10	+	func wantOutbox(t *testing.T, got []ports.ClientCommand, want ...ports.ClientCommand) {
	11	+	t.Helper()
	12	+	if !slices.Equal(got, want) {
	13	+	t.Fatalf("commands %#v, want %#v", got, want)
	14	+	}
	15	+	}
	16	+
	17	+	// takeAll returns what the sender would deliver, in order.
	18	+	func takeAll(o *outbox) []ports.ClientCommand {
	19	+	var got []ports.ClientCommand
	20	+	for _, e := range o.take() {
	21	+	got = append(got, e.command())
	22	+	}
	23	+	return got
	24	+	}
	25	+
	26	+	// Only adjacent motions for the same window merge: the position is the
	27	+	// latest, the deltas add up. Everything else keeps its place.
	28	+	func TestOutboxMergesAdjacentMotionOnly(t *testing.T) {
	29	+	o := newOutbox()
	30	+	// Nobody receives, so every command queues.
	31	+	out := make(chan ports.ClientCommand)
	32	+	push := func(v ports.ClientCommand) { o.push(out, v, ports.SecurityState{}, false) }
	33	+	push(ports.PointerMotionTo{ID: 1, X: 1, Y: 1, DX: 1, DY: 2, UnaccelDX: 3, UnaccelDY: 4})
	34	+	push(ports.PointerMotionTo{ID: 1, X: 2, Y: 2, DX: 10, DY: 20, UnaccelDX: 30, UnaccelDY: 40})
	35	+	push(ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true})
	36	+	push(ports.PointerMotionTo{ID: 1, X: 3, Y: 3, DX: 5})
	37	+	push(ports.PointerMotionTo{ID: 2, X: 4, Y: 4, DX: 7})
	38	+	push(ports.FocusWindow{ID: 2})
	39	+	push(ports.PointerMotionTo{ID: 2, X: 5, Y: 5, DX: 1})
	40	+	push(ports.PointerMotionTo{ID: 2, X: 6, Y: 6, DX: 1})
	41	+	want := []ports.ClientCommand{
	42	+	ports.PointerMotionTo{ID: 1, X: 2, Y: 2, DX: 11, DY: 22, UnaccelDX: 33, UnaccelDY: 44},
	43	+	ports.PointerButtonTo{ID: 1, Button: 0x110, Pressed: true},
	44	+	ports.PointerMotionTo{ID: 1, X: 3, Y: 3, DX: 5},
	45	+	ports.PointerMotionTo{ID: 2, X: 4, Y: 4, DX: 7},
	46	+	ports.FocusWindow{ID: 2},
	47	+	ports.PointerMotionTo{ID: 2, X: 6, Y: 6, DX: 2},
	48	+	}
	49	+	wantOutbox(t, takeAll(&o), want...)
	50	+	}
	51	+
	52	+	// Wrapped motions merge only under an identical owner envelope, and keep it.
	53	+	func TestOutboxMergesSecuredMotionPerEnvelope(t *testing.T) {
	54	+	o := newOutbox()
	55	+	out := make(chan ports.ClientCommand)
	56	+	a := ports.SecurityState{Generation: 1}
	57	+	b := ports.SecurityState{Generation: 2}
	58	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 1, DX: 1}, a, true)
	59	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 2, DX: 2}, a, true)
	60	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 3, DX: 4}, b, true)
	61	+	o.push(out, ports.PointerMotionTo{ID: 1, X: 4, DX: 8}, ports.SecurityState{}, false)
	62	+	wantOutbox(t, takeAll(&o),
	63	+	ports.SecurityCommand{State: a, Command: ports.PointerMotionTo{ID: 1, X: 2, DX: 3}},
	64	+	ports.SecurityCommand{State: b, Command: ports.PointerMotionTo{ID: 1, X: 3, DX: 4}},
	65	+	ports.PointerMotionTo{ID: 1, X: 4, DX: 8},
	66	+	)
	67	+	}
	68	+
	69	+	// With room in the channel and nothing queued, a command goes straight out.
	70	+	func TestOutboxSendsDirectlyWhileDisplayKeepsUp(t *testing.T) {
	71	+	o := newOutbox()
	72	+	out := make(chan ports.ClientCommand, 4)
	73	+	o.push(out, ports.PointerMotionTo{ID: 1, DX: 1}, ports.SecurityState{}, false)
	74	+	o.push(out, ports.PointerMotionTo{ID: 1, DX: 2}, ports.SecurityState{}, false)
	75	+	if len(out) != 2 {
	76	+	t.Fatalf("%d commands sent, want 2 unmerged", len(out))
	77	+	}
	78	+	if got := o.take(); len(got) != 0 {
	79	+	t.Fatalf("queued %d", len(got))
	80	+	}
	81	+	}

M internal/core/security_review_test.go +15 -24


@@ -29,28 +29,25 @@ func TestInputOutputSwitchBackpressureKeepsAdmittedEpoch(t *testing.T) {
29	29		var state atomic.Value
30	30		state.Store(ports.SecurityState{})
31	31		gate := portsmocks.NewMockSessionSecurity(t)
32		-	entered := make(chan struct{}, 1)
33	32		var calls atomic.Int32
34	33		gate.EXPECT().Snapshot().RunAndReturn(func() ports.SecurityState {
35	34		v := state.Load().(ports.SecurityState)
36	35		if calls.Add(1) == 2 {
37		-	entered <- struct{}{}
	36	+	// The epoch changes right after the check that admits the command.
	37	+	state.Store(ports.SecurityState{Generation: 2})
38	38		}
39	39		return v
40	40		})
41	41		c.opts.Security = gate
42	42		c.sent.outputs = ports.SetOutputs{Outputs: c.layout(), Focused: c.cur().name(), Off: c.offOutputs()}
43		-	commands := make(chan ports.ClientCommand)
	43	+	commands := make(chan ports.ClientCommand, 8)
44	44		c.ch.Commands = commands
45		-	done := make(chan error, 1)
46		-	go func() { done <- c.handleInput(context.Background(), ports.PointerMotion{X: 350, Y: 30}) }()
47		-	reviewReceive(t, entered)
48		-	state.Store(ports.SecurityState{Generation: 2})
	45	+	err := c.handleInput(context.Background(), ports.PointerMotion{X: 350, Y: 30})
49	46		cmd := reviewReceive(t, commands).(ports.SecurityCommand)
50	47		if cmd.State != (ports.SecurityState{}) {
51	48		t.Fatalf("old handler adopted new epoch: %+v", cmd)
52	49		}
53		-	if err := reviewReceive(t, done); !errors.Is(err, errSecurityChanged) {
	50	+	if !errors.Is(err, errSecurityChanged) {
54	51		t.Fatalf("handler not aborted: %v", err)
55	52		}
56	53		if c.security != (ports.SecurityState{}) {

@@ -88,29 +85,26 @@ func TestSlotsPendingBackpressureSuppressesSpawnAndDefersSlots(t *testing.T) {
88	85		var state atomic.Value
89	86		state.Store(ports.SecurityState{})
90	87		gate := portsmocks.NewMockSessionSecurity(t)
91		-	checked := make(chan struct{}, 1)
92	88		var calls atomic.Int32
93	89		gate.EXPECT().Snapshot().RunAndReturn(func() ports.SecurityState {
94	90		v := state.Load().(ports.SecurityState)
95	91		if calls.Add(1) == 2 {
96		-	checked <- struct{}{}
	92	+	// The gate locks right after the check that admits the slot.
	93	+	state.Store(ports.SecurityState{Generation: 1, Protected: true})
97	94		}
98	95		return v
99	96		})
100	97		c.opts.Security = gate
101		-	commands := make(chan ports.ClientCommand)
	98	+	commands := make(chan ports.ClientCommand, 8)
102	99		c.ch.Commands = commands
103	100		spawn := make(chan ports.SpawnRequest, 2)
104	101		c.ch.Spawn = spawn
105		-	done := make(chan error, 1)
106		-	go func() { done <- c.spawnSlots(context.Background(), false) }()
107		-	reviewReceive(t, checked)
108		-	state.Store(ports.SecurityState{Generation: 1, Protected: true})
	102	+	err := c.spawnSlots(context.Background(), false)
109	103		cmd := reviewReceive(t, commands).(ports.SecurityCommand)
110	104		if _, ok := cmd.Command.(ports.SlotsPending); !ok {
111	105		t.Fatal(cmd)
112	106		}
113		-	if err := reviewReceive(t, done); err != nil {
	107	+	if err != nil {
114	108		t.Fatal(err)
115	109		}
116	110		if len(spawn) != 0 \|\| len(c.toSpawn) != 1 \|\| c.placement.slotPending(key) {

@@ -138,26 +132,23 @@ func TestTerminalPendingBackpressureSuppressesSpawnAndRestoresRetry(t *testing.T
138	132		var state atomic.Value
139	133		state.Store(ports.SecurityState{})
140	134		gate := portsmocks.NewMockSessionSecurity(t)
141		-	checked := make(chan struct{}, 1)
142	135		var calls atomic.Int32
143	136		gate.EXPECT().Snapshot().RunAndReturn(func() ports.SecurityState {
144	137		v := state.Load().(ports.SecurityState)
145	138		if calls.Add(1) == 1 {
146		-	checked <- struct{}{}
	139	+	// The gate locks right after the check that admits the spawn.
	140	+	state.Store(ports.SecurityState{Generation: 1, Protected: true})
147	141		}
148	142		return v
149	143		})
150	144		c.opts.Security = gate
151		-	commands := make(chan ports.ClientCommand)
	145	+	commands := make(chan ports.ClientCommand, 8)
152	146		c.ch.Commands = commands
153	147		spawn := make(chan ports.SpawnRequest, 1)
154	148		c.ch.Spawn = spawn
155		-	done := make(chan error, 1)
156		-	go func() { done <- c.spawnEmpty(context.Background()) }()
157		-	reviewReceive(t, checked)
158		-	state.Store(ports.SecurityState{Generation: 1, Protected: true})
	149	+	err := c.spawnEmpty(context.Background())
159	150		reviewReceive(t, commands)
160		-	if err := reviewReceive(t, done); err != nil {
	151	+	if err != nil {
161	152		t.Fatal(err)
162	153		}
163	154		if len(spawn) != 0 \|\| c.placement.anyPending() {

 	var state atomic.Value
 	state.Store(ports.SecurityState{})
 	gate := portsmocks.NewMockSessionSecurity(t)
-	entered := make(chan struct{}, 1)
 	var calls atomic.Int32
 	gate.EXPECT().Snapshot().RunAndReturn(func() ports.SecurityState {
 		v := state.Load().(ports.SecurityState)
 		if calls.Add(1) == 2 {
-			entered <- struct{}{}
+			// The epoch changes right after the check that admits the command.
+			state.Store(ports.SecurityState{Generation: 2})
 		}
 		return v
 	})
 	c.opts.Security = gate
 	c.sent.outputs = ports.SetOutputs{Outputs: c.layout(), Focused: c.cur().name(), Off: c.offOutputs()}
-	commands := make(chan ports.ClientCommand)
+	commands := make(chan ports.ClientCommand, 8)
 	c.ch.Commands = commands
-	done := make(chan error, 1)
+	err := c.handleInput(context.Background(), ports.PointerMotion{X: 350, Y: 30})
-	go func() { done <- c.handleInput(context.Background(), ports.PointerMotion{X: 350, Y: 30}) }()
-	reviewReceive(t, entered)
-	state.Store(ports.SecurityState{Generation: 2})
 	cmd := reviewReceive(t, commands).(ports.SecurityCommand)
 	if cmd.State != (ports.SecurityState{}) {
 		t.Fatalf("old handler adopted new epoch: %+v", cmd)
 	}
-	if err := reviewReceive(t, done); !errors.Is(err, errSecurityChanged) {
+	if !errors.Is(err, errSecurityChanged) {
 		t.Fatalf("handler not aborted: %v", err)
 	}
 	if c.security != (ports.SecurityState{}) {
 	var state atomic.Value
 	state.Store(ports.SecurityState{})
 	gate := portsmocks.NewMockSessionSecurity(t)
-	checked := make(chan struct{}, 1)
 	var calls atomic.Int32
 	gate.EXPECT().Snapshot().RunAndReturn(func() ports.SecurityState {
 		v := state.Load().(ports.SecurityState)
 		if calls.Add(1) == 2 {
-			checked <- struct{}{}
+			// The gate locks right after the check that admits the slot.
+			state.Store(ports.SecurityState{Generation: 1, Protected: true})
 		}
 		return v
 	})
 	c.opts.Security = gate
-	commands := make(chan ports.ClientCommand)
+	commands := make(chan ports.ClientCommand, 8)
 	c.ch.Commands = commands
 	spawn := make(chan ports.SpawnRequest, 2)
 	c.ch.Spawn = spawn
-	done := make(chan error, 1)
+	err := c.spawnSlots(context.Background(), false)
-	go func() { done <- c.spawnSlots(context.Background(), false) }()
-	reviewReceive(t, checked)
-	state.Store(ports.SecurityState{Generation: 1, Protected: true})
 	cmd := reviewReceive(t, commands).(ports.SecurityCommand)
 	if _, ok := cmd.Command.(ports.SlotsPending); !ok {
 		t.Fatal(cmd)
 	}
-	if err := reviewReceive(t, done); err != nil {
+	if err != nil {
 		t.Fatal(err)
 	}
 	if len(spawn) != 0 || len(c.toSpawn) != 1 || c.placement.slotPending(key) {
 	var state atomic.Value
 	state.Store(ports.SecurityState{})
 	gate := portsmocks.NewMockSessionSecurity(t)
-	checked := make(chan struct{}, 1)
 	var calls atomic.Int32
 	gate.EXPECT().Snapshot().RunAndReturn(func() ports.SecurityState {
 		v := state.Load().(ports.SecurityState)
 		if calls.Add(1) == 1 {
-			checked <- struct{}{}
+			// The gate locks right after the check that admits the spawn.
+			state.Store(ports.SecurityState{Generation: 1, Protected: true})
 		}
 		return v
 	})
 	c.opts.Security = gate
-	commands := make(chan ports.ClientCommand)
+	commands := make(chan ports.ClientCommand, 8)
 	c.ch.Commands = commands
 	spawn := make(chan ports.SpawnRequest, 1)
 	c.ch.Spawn = spawn
-	done := make(chan error, 1)
+	err := c.spawnEmpty(context.Background())
-	go func() { done <- c.spawnEmpty(context.Background()) }()
-	reviewReceive(t, checked)
-	state.Store(ports.SecurityState{Generation: 1, Protected: true})
 	reviewReceive(t, commands)
-	if err := reviewReceive(t, done); err != nil {
+	if err != nil {
 		t.Fatal(err)
 	}
 	if len(spawn) != 0 || c.placement.anyPending() {

M internal/core/stash_spawn_internal_test.go +20 -32


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

@@ -41,39 +40,28 @@ func claimToken(req ports.SpawnRequest) string {
41	40		}
42	41
43	42		// Wayland reads tokens only while slots are pending, so the flag is sent
44		-	// before the spawn can reach the launcher. In a synctest bubble the worker
45		-	// is waited on until it blocks sending the flag on the unread command
46		-	// channel: the spawn channel must still be empty then.
	43	+	// before the spawn can reach the launcher. While the display keeps up the
	44	+	// flag is in the command channel when the spawn is handed over.
47	45		func TestSpawnStashPublishesPendingBeforeSpawn(t *testing.T) {
48		-	synctest.Test(t, func(t *testing.T) {
49		-	now := claimBase
50		-	commands := make(chan ports.ClientCommand)
51		-	spawn := make(chan ports.SpawnRequest, 1)
52		-	c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now))
53		-	ctx, cancel := context.WithCancel(context.Background())
54		-	t.Cleanup(cancel)
55		-	done := make(chan error, 1)
56		-	go func() { done <- c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"firefox"}}, false) }()
57		-	synctest.Wait()
58		-	if len(spawn) != 0 {
59		-	t.Fatal("spawned before the pending flag was delivered")
60		-	}
61		-	select {
62		-	case err := <-done:
63		-	t.Fatalf("spawnStash returned before the flag was read: %v", err)
64		-	default:
65		-	}
66		-	if p, ok := (<-commands).(ports.SlotsPending); !ok \|\| !p.Pending {
67		-	t.Fatal("first command is not SlotsPending{true}")
68		-	}
69		-	synctest.Wait()
70		-	if err := <-done; err != nil {
71		-	t.Fatal(err)
72		-	}
73		-	if len(spawn) != 1 {
74		-	t.Fatal("no spawn after the flag")
	46	+	now := claimBase
	47	+	commands := make(chan ports.ClientCommand, 8)
	48	+	spawn := make(chan ports.SpawnRequest, 1)
	49	+	c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now))
	50	+	spawned := make(chan struct{})
	51	+	go func() {
	52	+	<-spawn
	53	+	if len(commands) != 1 {
	54	+	t.Error("spawned before the pending flag was delivered")
75	55		}
76		-	})
	56	+	close(spawned)
	57	+	}()
	58	+	if err := c.spawnStash(context.Background(), c.cur().mon.Current(), [][]string{{"firefox"}}, false); err != nil {
	59	+	t.Fatal(err)
	60	+	}
	61	+	<-spawned
	62	+	if p, ok := (<-commands).(ports.SlotsPending); !ok \|\| !p.Pending {
	63	+	t.Fatal("first command is not SlotsPending{true}")
	64	+	}
77	65		}
78	66
79	67		// A send the launcher queue suppresses leaves no claim behind, and the

 	"errors"
 	"strings"
 	"testing"
-	"testing/synctest"
 	"time"
 	portsmocks "github.com/bnema/neferwl/internal/mocks/ports"
 }
 // Wayland reads tokens only while slots are pending, so the flag is sent
-// before the spawn can reach the launcher. In a synctest bubble the worker
+// before the spawn can reach the launcher. While the display keeps up the
-// is waited on until it blocks sending the flag on the unread command
+// flag is in the command channel when the spawn is handed over.
-// channel: the spawn channel must still be empty then.
 func TestSpawnStashPublishesPendingBeforeSpawn(t *testing.T) {
-	synctest.Test(t, func(t *testing.T) {
+	now := claimBase
-		now := claimBase
+	commands := make(chan ports.ClientCommand, 8)
-		commands := make(chan ports.ClientCommand)
+	spawn := make(chan ports.SpawnRequest, 1)
-		spawn := make(chan ports.SpawnRequest, 1)
+	c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now))
-		c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now))
+	spawned := make(chan struct{})
-		ctx, cancel := context.WithCancel(context.Background())
+	go func() {
-		t.Cleanup(cancel)
+		<-spawn
-		done := make(chan error, 1)
+		if len(commands) != 1 {
-		go func() { done <- c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"firefox"}}, false) }()
+			t.Error("spawned before the pending flag was delivered")
-		synctest.Wait()
-		if len(spawn) != 0 {
-			t.Fatal("spawned before the pending flag was delivered")
-		}
-		select {
-		case err := <-done:
-			t.Fatalf("spawnStash returned before the flag was read: %v", err)
-		default:
-		}
-		if p, ok := (<-commands).(ports.SlotsPending); !ok || !p.Pending {
-			t.Fatal("first command is not SlotsPending{true}")
-		}
-		synctest.Wait()
-		if err := <-done; err != nil {
-			t.Fatal(err)
-		}
-		if len(spawn) != 1 {
-			t.Fatal("no spawn after the flag")
 		}
-	})
+		close(spawned)
+	}()
+	if err := c.spawnStash(context.Background(), c.cur().mon.Current(), [][]string{{"firefox"}}, false); err != nil {
+		t.Fatal(err)
+	}
+	<-spawned
+	if p, ok := (<-commands).(ports.SlotsPending); !ok || !p.Pending {
+		t.Fatal("first command is not SlotsPending{true}")
+	}
 }
 // A send the launcher queue suppresses leaves no claim behind, and the