fix(core): deliver SlotsPending before its spawn and bound the outbox

The outbox let a spawn reach the launcher while SlotsPending{true} still
waited behind a stalled display: wayland ignored the slot token and the
window lost its placement. Raising the flag now waits until it is in the
command channel, and core drives delivery itself when no sender runs. The
outbox also holds at most outboxLimit commands; past that, core waits for
room instead of growing without bound or dropping input.

commit: 011648f8a8329b0ada45599fcc6ca192c2984c73
author: brice · 4d ago · 2026-10-18 05:42 UTC
parent: d7dc4d5
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +229 -38

M internal/core/core.go +46 -6


@@ -352,7 +352,12 @@ func (c *Core) apply(cfg ports.Config) error {
352	352		// their windows can map.
353	353		func (c *Core) publishPending(ctx context.Context) error {
354	354		if p := c.placement.anyPending(); p != c.sent.pending {
355		-	if err := c.command(ctx, ports.SlotsPending{Pending: p}); err != nil {
	355	+	// Pending must be in wayland's queue before the spawn that follows.
	356	+	send := c.command
	357	+	if p {
	358	+	send = c.commandFlushed
	359	+	}
	360	+	if err := send(ctx, ports.SlotsPending{Pending: p}); err != nil {
356	361		return err
357	362		}
358	363		c.sent.pending = p

@@ -448,24 +453,56 @@ func (c *Core) now() time.Time {
448	453		// display must not stall keyboard, scenes or frame wakes. The sender
449	454		// goroutine of Run delivers the queue in order.
450	455		func (c *Core) command(ctx context.Context, v ports.ClientCommand) error {
451		-	if err := ctx.Err(); err != nil {
	456	+	_, err := c.queue(ctx, v)
	457	+	return err
	458	+	}
	459	+
	460	+	// commandFlushed is command for a command that another channel depends on:
	461	+	// it returns once v is in the command channel. SlotsPending must reach
	462	+	// wayland before the spawn whose window it lets wayland claim; the spawn
	463	+	// goes on its own channel, so queue order alone does not keep them in order.
	464	+	func (c *Core) commandFlushed(ctx context.Context, v ports.ClientCommand) error {
	465	+	seq, err := c.queue(ctx, v)
	466	+	if err != nil \|\| seq == 0 {
452	467		return err
453	468		}
	469	+	if err := c.out.flush(ctx, c.ch.Commands, seq); err != nil {
	470	+	return err
	471	+	}
	472	+	// As a blocked send did: the rest of the event must not go on under a
	473	+	// new epoch.
454	474		if c.inputEpochChanged() {
455	475		return errSecurityChanged
456	476		}
	477	+	return nil
	478	+	}
	479	+
	480	+	// queue pushes v to the outbox and returns its sequence number (0: not
	481	+	// queued).
	482	+	func (c *Core) queue(ctx context.Context, v ports.ClientCommand) (uint64, error) {
	483	+	if err := ctx.Err(); err != nil {
	484	+	return 0, err
	485	+	}
	486	+	if c.inputEpochChanged() {
	487	+	return 0, errSecurityChanged
	488	+	}
457	489		if c.ch.Commands == nil {
458	490		// Nobody receives: queueing would only grow without bound.
459		-	return nil
	491	+	return 0, nil
	492	+	}
	493	+	seq := c.out.push(c.ch.Commands, v, c.security, c.opts.Security != nil)
	494	+	if c.out.backlog() > outboxLimit {
	495	+	if err := c.out.flush(ctx, c.ch.Commands, seq-outboxLimit); err != nil {
	496	+	return seq, err
	497	+	}
460	498		}
461		-	c.out.push(c.ch.Commands, v, c.security, c.opts.Security != nil)
462	499		// The queued command keeps the epoch it was made under, so a display
463	500		// that has moved on drops it. Report the change all the same: the rest
464	501		// of the event was decided under the old epoch and must not go on.
465	502		if c.inputEpochChanged() {
466		-	return errSecurityChanged
	503	+	return seq, errSecurityChanged
467	504		}
468		-	return nil
	505	+	return seq, nil
469	506		}
470	507
471	508		// clientRect is the placement minus the border on its inset sides.

@@ -1071,6 +1108,9 @@ func (c *Core) Run(ctx context.Context) error {
1071	1108		// quit and on errors. Run waits for it so nothing sends after Run.
1072	1109		sendCtx, stopSender := context.WithCancel(ctx)
1073	1110		senderDone := make(chan struct{})
	1111	+	if c.ch.Commands != nil {
	1112	+	c.out.start()
	1113	+	}
1074	1114		go func() {
1075	1115		defer close(senderDone)
1076	1116		if c.ch.Commands != nil {


@@ -352,7 +352,12 @@ func (c *Core) apply(cfg ports.Config) error {
352		// their windows can map.	352		// their windows can map.
353		func (c *Core) publishPending(ctx context.Context) error {	353		func (c *Core) publishPending(ctx context.Context) error {
354		if p := c.placement.anyPending(); p != c.sent.pending {	354		if p := c.placement.anyPending(); p != c.sent.pending {
355	-	if err := c.command(ctx, ports.SlotsPending{Pending: p}); err != nil {	355	+	// Pending must be in wayland's queue before the spawn that follows.
			356	+	send := c.command
			357	+	if p {
			358	+	send = c.commandFlushed
			359	+	}
			360	+	if err := send(ctx, ports.SlotsPending{Pending: p}); err != nil {
356		return err	361		return err
357		}	362		}
358		c.sent.pending = p	363		c.sent.pending = p

@@ -448,24 +453,56 @@ func (c *Core) now() time.Time {
448		// display must not stall keyboard, scenes or frame wakes. The sender	453		// display must not stall keyboard, scenes or frame wakes. The sender
449		// goroutine of Run delivers the queue in order.	454		// goroutine of Run delivers the queue in order.
450		func (c *Core) command(ctx context.Context, v ports.ClientCommand) error {	455		func (c *Core) command(ctx context.Context, v ports.ClientCommand) error {
451	-	if err := ctx.Err(); err != nil {	456	+	_, err := c.queue(ctx, v)
			457	+	return err
			458	+	}
			459	+
			460	+	// commandFlushed is command for a command that another channel depends on:
			461	+	// it returns once v is in the command channel. SlotsPending must reach
			462	+	// wayland before the spawn whose window it lets wayland claim; the spawn
			463	+	// goes on its own channel, so queue order alone does not keep them in order.
			464	+	func (c *Core) commandFlushed(ctx context.Context, v ports.ClientCommand) error {
			465	+	seq, err := c.queue(ctx, v)
			466	+	if err != nil \|\| seq == 0 {
452		return err	467		return err
453		}	468		}
			469	+	if err := c.out.flush(ctx, c.ch.Commands, seq); err != nil {
			470	+	return err
			471	+	}
			472	+	// As a blocked send did: the rest of the event must not go on under a
			473	+	// new epoch.
454		if c.inputEpochChanged() {	474		if c.inputEpochChanged() {
455		return errSecurityChanged	475		return errSecurityChanged
456		}	476		}
			477	+	return nil
			478	+	}
			479	+
			480	+	// queue pushes v to the outbox and returns its sequence number (0: not
			481	+	// queued).
			482	+	func (c *Core) queue(ctx context.Context, v ports.ClientCommand) (uint64, error) {
			483	+	if err := ctx.Err(); err != nil {
			484	+	return 0, err
			485	+	}
			486	+	if c.inputEpochChanged() {
			487	+	return 0, errSecurityChanged
			488	+	}
457		if c.ch.Commands == nil {	489		if c.ch.Commands == nil {
458		// Nobody receives: queueing would only grow without bound.	490		// Nobody receives: queueing would only grow without bound.
459	-	return nil	491	+	return 0, nil
			492	+	}
			493	+	seq := c.out.push(c.ch.Commands, v, c.security, c.opts.Security != nil)
			494	+	if c.out.backlog() > outboxLimit {
			495	+	if err := c.out.flush(ctx, c.ch.Commands, seq-outboxLimit); err != nil {
			496	+	return seq, err
			497	+	}
460		}	498		}
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	499		// 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	500		// 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.	501		// of the event was decided under the old epoch and must not go on.
465		if c.inputEpochChanged() {	502		if c.inputEpochChanged() {
466	-	return errSecurityChanged	503	+	return seq, errSecurityChanged
467		}	504		}
468	-	return nil	505	+	return seq, nil
469		}	506		}
470			507
471		// clientRect is the placement minus the border on its inset sides.	508		// clientRect is the placement minus the border on its inset sides.

@@ -1071,6 +1108,9 @@ func (c *Core) Run(ctx context.Context) error {
1071		// quit and on errors. Run waits for it so nothing sends after Run.	1108		// quit and on errors. Run waits for it so nothing sends after Run.
1072		sendCtx, stopSender := context.WithCancel(ctx)	1109		sendCtx, stopSender := context.WithCancel(ctx)
1073		senderDone := make(chan struct{})	1110		senderDone := make(chan struct{})
			1111	+	if c.ch.Commands != nil {
			1112	+	c.out.start()
			1113	+	}
1074		go func() {	1114		go func() {
1075		defer close(senderDone)	1115		defer close(senderDone)
1076		if c.ch.Commands != nil {	1116		if c.ch.Commands != nil {

M internal/core/outbox.go +105 -11


@@ -24,6 +24,15 @@ type outbox struct {
24	24		// busy is set while the sender holds a batch it has not finished
25	25		// sending: a command pushed then must queue behind it.
26	26		busy bool
	27	+	// pushed and sent count the entries queued and delivered (a merge adds
	28	+	// none); progress is closed and replaced at each delivery, for flush.
	29	+	pushed, sent uint64
	30	+	progress chan struct{}
	31	+	// running is set while a sender (run) owns delivery.
	32	+	running bool
	33	+	// sendMu is held by whoever takes and delivers a batch, so batches
	34	+	// never interleave (the sender, or flush without one).
	35	+	sendMu sync.Mutex
27	36		}
28	37
29	38		// outCmd is a queued command. A pointer motion is kept unboxed so that a

@@ -40,7 +49,22 @@ type outCmd struct {
40	49		merged bool
41	50		}
42	51
43		-	func newOutbox() outbox { return outbox{wake: make(chan struct{}, 1)} }
	52	+	// outboxLimit bounds the commands waiting for the display. Motion merges, so
	53	+	// only a display stalled for a long while reaches it; then core waits for
	54	+	// room as a full channel made it wait before, rather than grow without
	55	+	// bound or drop order-sensitive input.
	56	+	const outboxLimit = 4096
	57	+
	58	+	// backlog is the number of queued commands not yet delivered.
	59	+	func (o *outbox) backlog() uint64 {
	60	+	o.mu.Lock()
	61	+	defer o.mu.Unlock()
	62	+	return o.pushed - o.sent
	63	+	}
	64	+
	65	+	func newOutbox() outbox {
	66	+	return outbox{wake: make(chan struct{}, 1), progress: make(chan struct{})}
	67	+	}
44	68
45	69		// command returns what the sender delivers.
46	70		func (o *outCmd) command() ports.ClientCommand {

@@ -61,7 +85,9 @@ func (o *outCmd) command() ports.ClientCommand {
61	85		// it: the position is the latest and the relative deltas add up, so locked
62	86		// pointers (games) lose no movement. Nothing else is merged or reordered:
63	87		// 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) {
	88	+	//
	89	+	// It returns the entry's sequence number, for flush.
	90	+	func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, state ports.SecurityState, secured bool) uint64 {
65	91		e := outCmd{cmd: v, state: state, secured: secured}
66	92		e.motion, e.isMotion = v.(ports.PointerMotionTo)
67	93		o.mu.Lock()

@@ -69,8 +95,11 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
69	95		// A send that cannot block is safe under the lock.
70	96		select {
71	97		case out <- e.command():
	98	+	o.pushed++
	99	+	o.sent = o.pushed
	100	+	seq := o.pushed
72	101		o.mu.Unlock()
73		-	return
	102	+	return seq
74	103		default:
75	104		}
76	105		}

@@ -83,16 +112,49 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
83	112		e.motion.UnaccelDY += last.motion.UnaccelDY
84	113		e.merged = true
85	114		*last = e
	115	+	seq := o.pushed
86	116		o.mu.Unlock()
87		-	return
	117	+	return seq
88	118		}
89	119		}
90	120		o.queue = append(o.queue, e)
	121	+	o.pushed++
	122	+	seq := o.pushed
91	123		o.mu.Unlock()
92	124		select {
93	125		case o.wake <- struct{}{}:
94	126		default:
95	127		}
	128	+	return seq
	129	+	}
	130	+
	131	+	// flush waits until the entry seq is in out, or ctx ends. It is for the
	132	+	// rare command another channel depends on (SlotsPending before a spawn);
	133	+	// everything else never waits. Without a running sender (core driven
	134	+	// without Run), it delivers the queue itself, in order.
	135	+	func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
	136	+	for {
	137	+	o.mu.Lock()
	138	+	done, progress, running := o.sent >= seq, o.progress, o.running
	139	+	o.mu.Unlock()
	140	+	if done {
	141	+	return nil
	142	+	}
	143	+	if !running {
	144	+	o.sendMu.Lock()
	145	+	ok := o.deliver(ctx, out, o.take())
	146	+	o.sendMu.Unlock()
	147	+	if !ok {
	148	+	return ctx.Err()
	149	+	}
	150	+	continue
	151	+	}
	152	+	select {
	153	+	case <-progress:
	154	+	case <-ctx.Done():
	155	+	return ctx.Err()
	156	+	}
	157	+	}
96	158		}
97	159
98	160		// take returns the queued commands and leaves an empty queue. The caller

@@ -109,20 +171,41 @@ func (o *outbox) take() []outCmd {
109	171		// recycle keeps the batch's storage for the next take.
110	172		func (o *outbox) recycle(batch []outCmd) {
111	173		// Do not retain what the commands reference in the spare batch.
	174	+	n := uint64(len(batch))
112	175		clear(batch)
113	176		o.mu.Lock()
114	177		o.spare = batch[:0]
115	178		o.busy = false
	179	+	if n > 0 {
	180	+	o.sent += n
	181	+	close(o.progress)
	182	+	o.progress = make(chan struct{})
	183	+	}
	184	+	o.mu.Unlock()
	185	+	}
	186	+
	187	+	// start marks a sender as running; call it before starting run, so that a
	188	+	// flush from the owner never delivers alongside it.
	189	+	func (o *outbox) start() {
	190	+	o.mu.Lock()
	191	+	o.running = true
116	192		o.mu.Unlock()
117	193		}
118	194
119	195		// run delivers the queue to out in order until ctx ends. A command whose
120	196		// send is waiting is not merged any more, but what queues behind it is.
121	197		func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
	198	+	defer func() {
	199	+	o.mu.Lock()
	200	+	o.running = false
	201	+	o.mu.Unlock()
	202	+	}()
122	203		for {
	204	+	o.sendMu.Lock()
123	205		batch := o.take()
124	206		if len(batch) == 0 {
125	207		o.recycle(batch)
	208	+	o.sendMu.Unlock()
126	209		select {
127	210		case <-o.wake:
128	211		case <-ctx.Done():

@@ -130,13 +213,24 @@ func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
130	213		}
131	214		continue
132	215		}
133		-	for i := range batch {
134		-	select {
135		-	case out <- batch[i].command():
136		-	case <-ctx.Done():
137		-	return
138		-	}
	216	+	ok := o.deliver(ctx, out, batch)
	217	+	o.sendMu.Unlock()
	218	+	if !ok {
	219	+	return
	220	+	}
	221	+	}
	222	+	}
	223	+
	224	+	// deliver sends batch to out in order and recycles it; false when ctx ended
	225	+	// first (the rest of the batch is dropped with the session).
	226	+	func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, batch []outCmd) bool {
	227	+	for i := range batch {
	228	+	select {
	229	+	case out <- batch[i].command():
	230	+	case <-ctx.Done():
	231	+	return false
139	232		}
140		-	o.recycle(batch)
141	233		}
	234	+	o.recycle(batch)
	235	+	return true
142	236		}


@@ -24,6 +24,15 @@ type outbox struct {
24		// busy is set while the sender holds a batch it has not finished	24		// busy is set while the sender holds a batch it has not finished
25		// sending: a command pushed then must queue behind it.	25		// sending: a command pushed then must queue behind it.
26		busy bool	26		busy bool
			27	+	// pushed and sent count the entries queued and delivered (a merge adds
			28	+	// none); progress is closed and replaced at each delivery, for flush.
			29	+	pushed, sent uint64
			30	+	progress chan struct{}
			31	+	// running is set while a sender (run) owns delivery.
			32	+	running bool
			33	+	// sendMu is held by whoever takes and delivers a batch, so batches
			34	+	// never interleave (the sender, or flush without one).
			35	+	sendMu sync.Mutex
27		}	36		}
28			37
29		// outCmd is a queued command. A pointer motion is kept unboxed so that a	38		// outCmd is a queued command. A pointer motion is kept unboxed so that a

@@ -40,7 +49,22 @@ type outCmd struct {
40		merged bool	49		merged bool
41		}	50		}
42			51
43	-	func newOutbox() outbox { return outbox{wake: make(chan struct{}, 1)} }	52	+	// outboxLimit bounds the commands waiting for the display. Motion merges, so
			53	+	// only a display stalled for a long while reaches it; then core waits for
			54	+	// room as a full channel made it wait before, rather than grow without
			55	+	// bound or drop order-sensitive input.
			56	+	const outboxLimit = 4096
			57	+
			58	+	// backlog is the number of queued commands not yet delivered.
			59	+	func (o *outbox) backlog() uint64 {
			60	+	o.mu.Lock()
			61	+	defer o.mu.Unlock()
			62	+	return o.pushed - o.sent
			63	+	}
			64	+
			65	+	func newOutbox() outbox {
			66	+	return outbox{wake: make(chan struct{}, 1), progress: make(chan struct{})}
			67	+	}
44			68
45		// command returns what the sender delivers.	69		// command returns what the sender delivers.
46		func (o *outCmd) command() ports.ClientCommand {	70		func (o *outCmd) command() ports.ClientCommand {

@@ -61,7 +85,9 @@ func (o *outCmd) command() ports.ClientCommand {
61		// it: the position is the latest and the relative deltas add up, so locked	85		// 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:	86		// pointers (games) lose no movement. Nothing else is merged or reordered:
63		// buttons, focus, keys and gestures depend on their order.	87		// 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) {	88	+	//
			89	+	// It returns the entry's sequence number, for flush.
			90	+	func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, state ports.SecurityState, secured bool) uint64 {
65		e := outCmd{cmd: v, state: state, secured: secured}	91		e := outCmd{cmd: v, state: state, secured: secured}
66		e.motion, e.isMotion = v.(ports.PointerMotionTo)	92		e.motion, e.isMotion = v.(ports.PointerMotionTo)
67		o.mu.Lock()	93		o.mu.Lock()

@@ -69,8 +95,11 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
69		// A send that cannot block is safe under the lock.	95		// A send that cannot block is safe under the lock.
70		select {	96		select {
71		case out <- e.command():	97		case out <- e.command():
			98	+	o.pushed++
			99	+	o.sent = o.pushed
			100	+	seq := o.pushed
72		o.mu.Unlock()	101		o.mu.Unlock()
73	-	return	102	+	return seq
74		default:	103		default:
75		}	104		}
76		}	105		}

@@ -83,16 +112,49 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta
83		e.motion.UnaccelDY += last.motion.UnaccelDY	112		e.motion.UnaccelDY += last.motion.UnaccelDY
84		e.merged = true	113		e.merged = true
85		*last = e	114		*last = e
			115	+	seq := o.pushed
86		o.mu.Unlock()	116		o.mu.Unlock()
87	-	return	117	+	return seq
88		}	118		}
89		}	119		}
90		o.queue = append(o.queue, e)	120		o.queue = append(o.queue, e)
			121	+	o.pushed++
			122	+	seq := o.pushed
91		o.mu.Unlock()	123		o.mu.Unlock()
92		select {	124		select {
93		case o.wake <- struct{}{}:	125		case o.wake <- struct{}{}:
94		default:	126		default:
95		}	127		}
			128	+	return seq
			129	+	}
			130	+
			131	+	// flush waits until the entry seq is in out, or ctx ends. It is for the
			132	+	// rare command another channel depends on (SlotsPending before a spawn);
			133	+	// everything else never waits. Without a running sender (core driven
			134	+	// without Run), it delivers the queue itself, in order.
			135	+	func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error {
			136	+	for {
			137	+	o.mu.Lock()
			138	+	done, progress, running := o.sent >= seq, o.progress, o.running
			139	+	o.mu.Unlock()
			140	+	if done {
			141	+	return nil
			142	+	}
			143	+	if !running {
			144	+	o.sendMu.Lock()
			145	+	ok := o.deliver(ctx, out, o.take())
			146	+	o.sendMu.Unlock()
			147	+	if !ok {
			148	+	return ctx.Err()
			149	+	}
			150	+	continue
			151	+	}
			152	+	select {
			153	+	case <-progress:
			154	+	case <-ctx.Done():
			155	+	return ctx.Err()
			156	+	}
			157	+	}
96		}	158		}
97			159
98		// take returns the queued commands and leaves an empty queue. The caller	160		// take returns the queued commands and leaves an empty queue. The caller

@@ -109,20 +171,41 @@ func (o *outbox) take() []outCmd {
109		// recycle keeps the batch's storage for the next take.	171		// recycle keeps the batch's storage for the next take.
110		func (o *outbox) recycle(batch []outCmd) {	172		func (o *outbox) recycle(batch []outCmd) {
111		// Do not retain what the commands reference in the spare batch.	173		// Do not retain what the commands reference in the spare batch.
			174	+	n := uint64(len(batch))
112		clear(batch)	175		clear(batch)
113		o.mu.Lock()	176		o.mu.Lock()
114		o.spare = batch[:0]	177		o.spare = batch[:0]
115		o.busy = false	178		o.busy = false
			179	+	if n > 0 {
			180	+	o.sent += n
			181	+	close(o.progress)
			182	+	o.progress = make(chan struct{})
			183	+	}
			184	+	o.mu.Unlock()
			185	+	}
			186	+
			187	+	// start marks a sender as running; call it before starting run, so that a
			188	+	// flush from the owner never delivers alongside it.
			189	+	func (o *outbox) start() {
			190	+	o.mu.Lock()
			191	+	o.running = true
116		o.mu.Unlock()	192		o.mu.Unlock()
117		}	193		}
118			194
119		// run delivers the queue to out in order until ctx ends. A command whose	195		// 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.	196		// 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) {	197		func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
			198	+	defer func() {
			199	+	o.mu.Lock()
			200	+	o.running = false
			201	+	o.mu.Unlock()
			202	+	}()
122		for {	203		for {
			204	+	o.sendMu.Lock()
123		batch := o.take()	205		batch := o.take()
124		if len(batch) == 0 {	206		if len(batch) == 0 {
125		o.recycle(batch)	207		o.recycle(batch)
			208	+	o.sendMu.Unlock()
126		select {	209		select {
127		case <-o.wake:	210		case <-o.wake:
128		case <-ctx.Done():	211		case <-ctx.Done():

@@ -130,13 +213,24 @@ func (o *outbox) run(ctx context.Context, out chan<- ports.ClientCommand) {
130		}	213		}
131		continue	214		continue
132		}	215		}
133	-	for i := range batch {	216	+	ok := o.deliver(ctx, out, batch)
134	-	select {	217	+	o.sendMu.Unlock()
135	-	case out <- batch[i].command():	218	+	if !ok {
136	-	case <-ctx.Done():	219	+	return
137	-	return	220	+	}
138	-	}	221	+	}
			222	+	}
			223	+
			224	+	// deliver sends batch to out in order and recycles it; false when ctx ended
			225	+	// first (the rest of the batch is dropped with the session).
			226	+	func (o *outbox) deliver(ctx context.Context, out chan<- ports.ClientCommand, batch []outCmd) bool {
			227	+	for i := range batch {
			228	+	select {
			229	+	case out <- batch[i].command():
			230	+	case <-ctx.Done():
			231	+	return false
139		}	232		}
140	-	o.recycle(batch)
141		}	233		}
			234	+	o.recycle(batch)
			235	+	return true
142		}	236		}

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

@@ -1,8 +1,10 @@
1	1		package core
2	2
3	3		import (
	4	+	"context"
4	5		"slices"
5	6		"testing"
	7	+	"testing/synctest"
6	8
7	9		"github.com/bnema/neferwl/internal/ports"
8	10		)

@@ -79,3 +81,44 @@ func TestOutboxSendsDirectlyWhileDisplayKeepsUp(t *testing.T) {
79	81		t.Fatalf("queued %d", len(got))
80	82		}
81	83		}
	84	+
	85	+	// A display that reads nothing for long fills the outbox to its limit, then
	86	+	// core waits for room: the queue never grows past it, nothing is dropped.
	87	+	func TestOutboxIsBounded(t *testing.T) {
	88	+	synctest.Test(t, func(t *testing.T) {
	89	+	out := make(chan ports.ClientCommand)
	90	+	c := &Core{out: newOutbox(), ch: Channels{Commands: out}}
	91	+	c.out.start()
	92	+	ctx, cancel := context.WithCancel(context.Background())
	93	+	defer cancel()
	94	+	go c.out.run(ctx, out)
	95	+	done := make(chan error, 1)
	96	+	go func() {
	97	+	for i := range outboxLimit + 10 {
	98	+	// Not mergeable: each takes a place.
	99	+	if err := c.command(ctx, ports.FocusWindow{ID: WindowID(i)}); err != nil {
	100	+	done <- err
	101	+	return
	102	+	}
	103	+	}
	104	+	done <- nil
	105	+	}()
	106	+	synctest.Wait()
	107	+	if n := c.out.backlog(); n != outboxLimit+1 {
	108	+	t.Fatalf("backlog %d, want the limit and the one waiting", n)
	109	+	}
	110	+	select {
	111	+	case err := <-done:
	112	+	t.Fatalf("pushed past the limit: %v", err)
	113	+	default:
	114	+	}
	115	+	for i := range outboxLimit + 10 {
	116	+	if got := (<-out).(ports.FocusWindow); got.ID != WindowID(i) {
	117	+	t.Fatalf("command %d is %+v", i, got)
	118	+	}
	119	+	}
	120	+	if err := <-done; err != nil {
	121	+	t.Fatal(err)
	122	+	}
	123	+	})
	124	+	}

@@ -1,8 +1,10 @@
1	package core	1		package core
2		2
3	import (	3		import (
		4	+	"context"
4	"slices"	5		"slices"
5	"testing"	6		"testing"
		7	+	"testing/synctest"
6		8
7	"github.com/bnema/neferwl/internal/ports"	9		"github.com/bnema/neferwl/internal/ports"
8	)	10		)

@@ -79,3 +81,44 @@ func TestOutboxSendsDirectlyWhileDisplayKeepsUp(t *testing.T) {
79	t.Fatalf("queued %d", len(got))	81		t.Fatalf("queued %d", len(got))
80	}	82		}
81	}	83		}
		84	+
		85	+	// A display that reads nothing for long fills the outbox to its limit, then
		86	+	// core waits for room: the queue never grows past it, nothing is dropped.
		87	+	func TestOutboxIsBounded(t *testing.T) {
		88	+	synctest.Test(t, func(t *testing.T) {
		89	+	out := make(chan ports.ClientCommand)
		90	+	c := &Core{out: newOutbox(), ch: Channels{Commands: out}}
		91	+	c.out.start()
		92	+	ctx, cancel := context.WithCancel(context.Background())
		93	+	defer cancel()
		94	+	go c.out.run(ctx, out)
		95	+	done := make(chan error, 1)
		96	+	go func() {
		97	+	for i := range outboxLimit + 10 {
		98	+	// Not mergeable: each takes a place.
		99	+	if err := c.command(ctx, ports.FocusWindow{ID: WindowID(i)}); err != nil {
		100	+	done <- err
		101	+	return
		102	+	}
		103	+	}
		104	+	done <- nil
		105	+	}()
		106	+	synctest.Wait()
		107	+	if n := c.out.backlog(); n != outboxLimit+1 {
		108	+	t.Fatalf("backlog %d, want the limit and the one waiting", n)
		109	+	}
		110	+	select {
		111	+	case err := <-done:
		112	+	t.Fatalf("pushed past the limit: %v", err)
		113	+	default:
		114	+	}
		115	+	for i := range outboxLimit + 10 {
		116	+	if got := (<-out).(ports.FocusWindow); got.ID != WindowID(i) {
		117	+	t.Fatalf("command %d is %+v", i, got)
		118	+	}
		119	+	}
		120	+	if err := <-done; err != nil {
		121	+	t.Fatal(err)
		122	+	}
		123	+	})
		124	+	}

M internal/core/stash_spawn_internal_test.go +35 -21


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

@@ -39,29 +40,42 @@ func claimToken(req ports.SpawnRequest) string {
39	40		return strings.TrimPrefix(req.Env[0], ports.SlotEnv+"=")
40	41		}
41	42
42		-	// Wayland reads tokens only while slots are pending, so the flag is sent
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.
	43	+	// Wayland reads tokens only while slots are pending, so the flag is in the
	44	+	// command channel before the spawn can reach the launcher, also when the
	45	+	// display reads nothing: in a synctest bubble the sender is waited on until
	46	+	// it blocks on the unread channel, and the spawn channel must still be empty.
45	47		func TestSpawnStashPublishesPendingBeforeSpawn(t *testing.T) {
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")
	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	+	c.out.start()
	56	+	go c.out.run(ctx, commands)
	57	+	done := make(chan error, 1)
	58	+	go func() { done <- c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"firefox"}}, false) }()
	59	+	synctest.Wait()
	60	+	if len(spawn) != 0 {
	61	+	t.Fatal("spawned before the pending flag was delivered")
55	62		}
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		-	}
	63	+	select {
	64	+	case err := <-done:
	65	+	t.Fatalf("spawnStash returned before the flag was read: %v", err)
	66	+	default:
	67	+	}
	68	+	if p, ok := (<-commands).(ports.SlotsPending); !ok \|\| !p.Pending {
	69	+	t.Fatal("first command is not SlotsPending{true}")
	70	+	}
	71	+	synctest.Wait()
	72	+	if err := <-done; err != nil {
	73	+	t.Fatal(err)
	74	+	}
	75	+	if len(spawn) != 1 {
	76	+	t.Fatal("no spawn after the flag")
	77	+	}
	78	+	})
65	79		}
66	80
67	81		// A send the launcher queue suppresses leaves no claim behind, and the


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

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