fix: stop scheduler for closed windows

commit: 8f5354becd8bf7636ce7cb7b024aa40d154921f1
author: brice · 3mo ago · 2026-07-02 15:43 UTC
parent: 9fab834
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +98 -3

M internal/usecase/daemon/daemon.go +19 -0


@@ -128,6 +128,8 @@ type window struct {
128	128		screen *vt.Screen
129	129		dirty chan struct{}
130	130		size domain.Size
	131	+	ctx context.Context
	132	+	cancel context.CancelFunc
131	133		}
132	134
133	135		// attachedClient is a client currently attached to a session's window. rend is

@@ -506,6 +508,7 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, sz domain.Size
506	508		size: winSize,
507	509		}
508	510		sctx, cancel := context.WithCancel(d.serveCtx)
	511	+	win.ctx, win.cancel = context.WithCancel(sctx)
509	512		sess := &session{
510	513		id: id,
511	514		name: name,

@@ -531,6 +534,7 @@ func (d Daemon) createWindow(sess session, sz domain.Size) error {
531	534		dirty: make(chan struct{}, 1),
532	535		size: winSize,
533	536		}
	537	+	win.ctx, win.cancel = context.WithCancel(sess.ctx)
534	538		sess.mu.Lock()
535	539		sess.windows = append(sess.windows, win)
536	540		sess.active = len(sess.windows) - 1

@@ -780,6 +784,9 @@ func (d Daemon) closeWindow(sess session, win *window, repaint bool) {
780	784		ac := sess.client
781	785		sess.mu.Unlock()
782	786
	787	+	if win.cancel != nil {
	788	+	win.cancel()
	789	+	}
783	790		_ = win.pty.Close()
784	791		if repaint && ac != nil {
785	792		d.paint(sess, ac, true)

@@ -818,6 +825,8 @@ func (d Daemon) scheduler(sess session, win *window) {
818	825		select {
819	826		case <-sess.ctx.Done():
820	827		return
	828	+	case <-windowDone(win):
	829	+	return
821	830		case <-win.dirty:
822	831		}
823	832

@@ -828,6 +837,9 @@ func (d Daemon) scheduler(sess session, win *window) {
828	837		case <-sess.ctx.Done():
829	838		timer.Stop()
830	839		return
	840	+	case <-windowDone(win):
	841	+	timer.Stop()
	842	+	return
831	843		case <-win.dirty:
832	844		// Coalesced into the pending render.
833	845		case <-timer.C():

@@ -838,6 +850,13 @@ func (d Daemon) scheduler(sess session, win *window) {
838	850		}
839	851		}
840	852
	853	+	func windowDone(win *window) <-chan struct{} {
	854	+	if win.ctx == nil {
	855	+	return nil
	856	+	}
	857	+	return win.ctx.Done()
	858	+	}
	859	+
841	860		// render paints the current client, or (when detached) just clears accumulated
842	861		// damage so it never grows unbounded while headless.
843	862		func (d Daemon) render(sess session, win *window) {


@@ -128,6 +128,8 @@ type window struct {
128	screen *vt.Screen	128		screen *vt.Screen
129	dirty chan struct{}	129		dirty chan struct{}
130	size domain.Size	130		size domain.Size
		131	+	ctx context.Context
		132	+	cancel context.CancelFunc
131	}	133		}
132		134
133	// attachedClient is a client currently attached to a session's window. rend is	135		// attachedClient is a client currently attached to a session's window. rend is

@@ -506,6 +508,7 @@ func (d *Daemon) createSessionLocked(name string, ephemeral bool, sz domain.Size
506	size: winSize,	508		size: winSize,
507	}	509		}
508	sctx, cancel := context.WithCancel(d.serveCtx)	510		sctx, cancel := context.WithCancel(d.serveCtx)
		511	+	win.ctx, win.cancel = context.WithCancel(sctx)
509	sess := &session{	512		sess := &session{
510	id: id,	513		id: id,
511	name: name,	514		name: name,

@@ -531,6 +534,7 @@ func (d Daemon) createWindow(sess session, sz domain.Size) error {
531	dirty: make(chan struct{}, 1),	534		dirty: make(chan struct{}, 1),
532	size: winSize,	535		size: winSize,
533	}	536		}
		537	+	win.ctx, win.cancel = context.WithCancel(sess.ctx)
534	sess.mu.Lock()	538		sess.mu.Lock()
535	sess.windows = append(sess.windows, win)	539		sess.windows = append(sess.windows, win)
536	sess.active = len(sess.windows) - 1	540		sess.active = len(sess.windows) - 1

@@ -780,6 +784,9 @@ func (d Daemon) closeWindow(sess session, win *window, repaint bool) {
780	ac := sess.client	784		ac := sess.client
781	sess.mu.Unlock()	785		sess.mu.Unlock()
782		786
		787	+	if win.cancel != nil {
		788	+	win.cancel()
		789	+	}
783	_ = win.pty.Close()	790		_ = win.pty.Close()
784	if repaint && ac != nil {	791		if repaint && ac != nil {
785	d.paint(sess, ac, true)	792		d.paint(sess, ac, true)

@@ -818,6 +825,8 @@ func (d Daemon) scheduler(sess session, win *window) {
818	select {	825		select {
819	case <-sess.ctx.Done():	826		case <-sess.ctx.Done():
820	return	827		return
		828	+	case <-windowDone(win):
		829	+	return
821	case <-win.dirty:	830		case <-win.dirty:
822	}	831		}
823		832

@@ -828,6 +837,9 @@ func (d Daemon) scheduler(sess session, win *window) {
828	case <-sess.ctx.Done():	837		case <-sess.ctx.Done():
829	timer.Stop()	838		timer.Stop()
830	return	839		return
		840	+	case <-windowDone(win):
		841	+	timer.Stop()
		842	+	return
831	case <-win.dirty:	843		case <-win.dirty:
832	// Coalesced into the pending render.	844		// Coalesced into the pending render.
833	case <-timer.C():	845		case <-timer.C():

@@ -838,6 +850,13 @@ func (d Daemon) scheduler(sess session, win *window) {
838	}	850		}
839	}	851		}
840		852
		853	+	func windowDone(win *window) <-chan struct{} {
		854	+	if win.ctx == nil {
		855	+	return nil
		856	+	}
		857	+	return win.ctx.Done()
		858	+	}
		859	+
841	// render paints the current client, or (when detached) just clears accumulated	860		// render paints the current client, or (when detached) just clears accumulated
842	// damage so it never grows unbounded while headless.	861		// damage so it never grows unbounded while headless.
843	func (d Daemon) render(sess session, win *window) {	862		func (d Daemon) render(sess session, win *window) {

M internal/usecase/daemon/daemon_test.go +79 -3


@@ -35,6 +35,17 @@ func (stubTimer) C() <-chan time.Time { return nil }
35	35		func (stubTimer) Reset(time.Duration) bool { return false }
36	36		func (stubTimer) Stop() bool { return true }
37	37
	38	+	type signalClock struct {
	39	+	called chan struct{}
	40	+	once sync.Once
	41	+	}
	42	+
	43	+	func (c *signalClock) Now() time.Time { return time.Time{} }
	44	+	func (c *signalClock) NewTimer(time.Duration) ports.Timer {
	45	+	c.once.Do(func() { close(c.called) })
	46	+	return stubTimer{}
	47	+	}
	48	+
38	49		func newTestDaemon(t testing.T, ptys ports.PTYFactory, clk ports.Clock) Daemon {
39	50		t.Helper()
40	51		d := New(ptys, clk, slog.New(slog.NewTextHandler(io.Discard, nil)))

@@ -207,7 +218,8 @@ func newManualSessionWithPTYs(t testing.T, ptys ...ports.PTY) (Daemon, *sessio
207	218		sctx, cancel := context.WithCancel(d.serveCtx)
208	219		windows := make([]*window, 0, len(ptys))
209	220		for _, p := range ptys {
210		-	windows = append(windows, &window{pty: p, screen: vt.NewScreen(80, 23), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 80, Rows: 23}})
	221	+	wctx, wcancel := context.WithCancel(sctx)
	222	+	windows = append(windows, &window{pty: p, screen: vt.NewScreen(80, 23), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 80, Rows: 23}, ctx: wctx, cancel: wcancel})
211	223		}
212	224		sess := &session{id: "manual", name: "work", ctx: sctx, cancel: cancel, windows: windows, client: ac}
213	225		d.sessions[sess.id] = sess

@@ -234,6 +246,12 @@ func activeWindowIndex(sess *session) int {
234	246		return sess.active
235	247		}
236	248
	249	+	func windowCount(sess *session) int {
	250	+	sess.mu.Lock()
	251	+	defer sess.mu.Unlock()
	252	+	return len(sess.windows)
	253	+	}
	254	+
237	255		// --- handshake --------------------------------------------------------------
238	256
239	257		func TestHandshakeEphemeralHappy(t *testing.T) {

@@ -461,6 +479,64 @@ func TestAltXClosesActiveWindowAndSelectsRemaining(t *testing.T) {
461	479		releasePTY3()
462	480		}
463	481
	482	+	func waitGroupDone(wg *sync.WaitGroup) <-chan struct{} {
	483	+	done := make(chan struct{})
	484	+	go func() {
	485	+	wg.Wait()
	486	+	close(done)
	487	+	}()
	488	+	return done
	489	+	}
	490	+
	491	+	func TestAltXClosesNonFinalWindowScheduler(t *testing.T) {
	492	+	p1, releasePTY1 := newBlockingPTY(t)
	493	+	p2, releasePTY2 := newBlockingPTY(t)
	494	+	clk := &signalClock{called: make(chan struct{})}
	495	+	d, sess, ac, _ := newManualSessionWithPTYs(t, p1, p2)
	496	+	d.clock = clk
	497	+	defer releasePTY1()
	498	+	defer releasePTY2()
	499	+
	500	+	d.sessWg.Add(1)
	501	+	go d.scheduler(sess, sess.windows[1])
	502	+	sess.windows[1].dirty <- struct{}{}
	503	+	<-clk.called
	504	+
	505	+	d.handleInput(sess, ac, []byte("\x1b2"))
	506	+	d.handleInput(sess, ac, []byte("\x1bx"))
	507	+
	508	+	select {
	509	+	case <-waitGroupDone(&d.sessWg):
	510	+	case <-time.After(2 * time.Second):
	511	+	t.Fatal("scheduler for removed window did not exit")
	512	+	}
	513	+	require.Equal(t, 1, sessionCount(d))
	514	+	}
	515	+
	516	+	func TestPTYEOFClosesNonFinalWindowScheduler(t *testing.T) {
	517	+	p1, releasePTY1 := newBlockingPTY(t)
	518	+	p2, releasePTY2 := newBlockingPTY(t)
	519	+	clk := &signalClock{called: make(chan struct{})}
	520	+	d, sess, _, _ := newManualSessionWithPTYs(t, p1, p2)
	521	+	d.clock = clk
	522	+	defer releasePTY1()
	523	+	defer releasePTY2()
	524	+
	525	+	d.sessWg.Add(2)
	526	+	go d.scheduler(sess, sess.windows[1])
	527	+	go d.ptyReader(sess, sess.windows[1])
	528	+	sess.windows[1].dirty <- struct{}{}
	529	+	<-clk.called
	530	+	releasePTY2()
	531	+
	532	+	select {
	533	+	case <-waitGroupDone(&d.sessWg):
	534	+	case <-time.After(2 * time.Second):
	535	+	t.Fatal("scheduler for EOF-removed window did not exit")
	536	+	}
	537	+	require.Equal(t, 1, sessionCount(d))
	538	+	}
	539	+
464	540		func TestAltXClosesFinalWindowAndDetaches(t *testing.T) {
465	541		d, sess, ac, sends, releases := newManualWindowSession(t, 1)
466	542		defer releases[0]()

@@ -486,7 +562,7 @@ func TestPTYEOFClosesActiveNonFinalWindowAndRepaintsRemaining(t *testing.T) {
486	562		go d.ptyReader(sess, sess.windows[0])
487	563		releasePTY1()
488	564
489		-	require.Eventually(t, func() bool { return len(sess.windows) == 1 }, 2time.Second, 5time.Millisecond)
	565	+	require.Eventually(t, func() bool { return windowCount(sess) == 1 }, 2time.Second, 5time.Millisecond)
490	566		require.Equal(t, 1, sessionCount(d))
491	567		require.Equal(t, 0, activeWindowIndex(sess))
492	568		f := awaitFrame(t, sends, ports.MsgOutput)

@@ -512,7 +588,7 @@ func TestPTYEOFClosesInactiveNonFinalWindowAndRepaintsStatus(t *testing.T) {
512	588		go d.ptyReader(sess, sess.windows[1])
513	589		releasePTY2()
514	590
515		-	require.Eventually(t, func() bool { return len(sess.windows) == 1 }, 2time.Second, 5time.Millisecond)
	591	+	require.Eventually(t, func() bool { return windowCount(sess) == 1 }, 2time.Second, 5time.Millisecond)
516	592		require.Equal(t, 1, sessionCount(d))
517	593		require.Equal(t, 0, activeWindowIndex(sess))
518	594		f := awaitFrame(t, sends, ports.MsgOutput)


@@ -35,6 +35,17 @@ func (stubTimer) C() <-chan time.Time { return nil }
35		func (stubTimer) Reset(time.Duration) bool { return false }	35		func (stubTimer) Reset(time.Duration) bool { return false }
36		func (stubTimer) Stop() bool { return true }	36		func (stubTimer) Stop() bool { return true }
37			37
			38	+	type signalClock struct {
			39	+	called chan struct{}
			40	+	once sync.Once
			41	+	}
			42	+
			43	+	func (c *signalClock) Now() time.Time { return time.Time{} }
			44	+	func (c *signalClock) NewTimer(time.Duration) ports.Timer {
			45	+	c.once.Do(func() { close(c.called) })
			46	+	return stubTimer{}
			47	+	}
			48	+
38		func newTestDaemon(t testing.T, ptys ports.PTYFactory, clk ports.Clock) Daemon {	49		func newTestDaemon(t testing.T, ptys ports.PTYFactory, clk ports.Clock) Daemon {
39		t.Helper()	50		t.Helper()
40		d := New(ptys, clk, slog.New(slog.NewTextHandler(io.Discard, nil)))	51		d := New(ptys, clk, slog.New(slog.NewTextHandler(io.Discard, nil)))

@@ -207,7 +218,8 @@ func newManualSessionWithPTYs(t testing.T, ptys ...ports.PTY) (Daemon, *sessio
207		sctx, cancel := context.WithCancel(d.serveCtx)	218		sctx, cancel := context.WithCancel(d.serveCtx)
208		windows := make([]*window, 0, len(ptys))	219		windows := make([]*window, 0, len(ptys))
209		for _, p := range ptys {	220		for _, p := range ptys {
210	-	windows = append(windows, &window{pty: p, screen: vt.NewScreen(80, 23), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 80, Rows: 23}})	221	+	wctx, wcancel := context.WithCancel(sctx)
			222	+	windows = append(windows, &window{pty: p, screen: vt.NewScreen(80, 23), dirty: make(chan struct{}, 1), size: domain.Size{Cols: 80, Rows: 23}, ctx: wctx, cancel: wcancel})
211		}	223		}
212		sess := &session{id: "manual", name: "work", ctx: sctx, cancel: cancel, windows: windows, client: ac}	224		sess := &session{id: "manual", name: "work", ctx: sctx, cancel: cancel, windows: windows, client: ac}
213		d.sessions[sess.id] = sess	225		d.sessions[sess.id] = sess

@@ -234,6 +246,12 @@ func activeWindowIndex(sess *session) int {
234		return sess.active	246		return sess.active
235		}	247		}
236			248
			249	+	func windowCount(sess *session) int {
			250	+	sess.mu.Lock()
			251	+	defer sess.mu.Unlock()
			252	+	return len(sess.windows)
			253	+	}
			254	+
237		// --- handshake --------------------------------------------------------------	255		// --- handshake --------------------------------------------------------------
238			256
239		func TestHandshakeEphemeralHappy(t *testing.T) {	257		func TestHandshakeEphemeralHappy(t *testing.T) {

@@ -461,6 +479,64 @@ func TestAltXClosesActiveWindowAndSelectsRemaining(t *testing.T) {
461		releasePTY3()	479		releasePTY3()
462		}	480		}
463			481
			482	+	func waitGroupDone(wg *sync.WaitGroup) <-chan struct{} {
			483	+	done := make(chan struct{})
			484	+	go func() {
			485	+	wg.Wait()
			486	+	close(done)
			487	+	}()
			488	+	return done
			489	+	}
			490	+
			491	+	func TestAltXClosesNonFinalWindowScheduler(t *testing.T) {
			492	+	p1, releasePTY1 := newBlockingPTY(t)
			493	+	p2, releasePTY2 := newBlockingPTY(t)
			494	+	clk := &signalClock{called: make(chan struct{})}
			495	+	d, sess, ac, _ := newManualSessionWithPTYs(t, p1, p2)
			496	+	d.clock = clk
			497	+	defer releasePTY1()
			498	+	defer releasePTY2()
			499	+
			500	+	d.sessWg.Add(1)
			501	+	go d.scheduler(sess, sess.windows[1])
			502	+	sess.windows[1].dirty <- struct{}{}
			503	+	<-clk.called
			504	+
			505	+	d.handleInput(sess, ac, []byte("\x1b2"))
			506	+	d.handleInput(sess, ac, []byte("\x1bx"))
			507	+
			508	+	select {
			509	+	case <-waitGroupDone(&d.sessWg):
			510	+	case <-time.After(2 * time.Second):
			511	+	t.Fatal("scheduler for removed window did not exit")
			512	+	}
			513	+	require.Equal(t, 1, sessionCount(d))
			514	+	}
			515	+
			516	+	func TestPTYEOFClosesNonFinalWindowScheduler(t *testing.T) {
			517	+	p1, releasePTY1 := newBlockingPTY(t)
			518	+	p2, releasePTY2 := newBlockingPTY(t)
			519	+	clk := &signalClock{called: make(chan struct{})}
			520	+	d, sess, _, _ := newManualSessionWithPTYs(t, p1, p2)
			521	+	d.clock = clk
			522	+	defer releasePTY1()
			523	+	defer releasePTY2()
			524	+
			525	+	d.sessWg.Add(2)
			526	+	go d.scheduler(sess, sess.windows[1])
			527	+	go d.ptyReader(sess, sess.windows[1])
			528	+	sess.windows[1].dirty <- struct{}{}
			529	+	<-clk.called
			530	+	releasePTY2()
			531	+
			532	+	select {
			533	+	case <-waitGroupDone(&d.sessWg):
			534	+	case <-time.After(2 * time.Second):
			535	+	t.Fatal("scheduler for EOF-removed window did not exit")
			536	+	}
			537	+	require.Equal(t, 1, sessionCount(d))
			538	+	}
			539	+
464		func TestAltXClosesFinalWindowAndDetaches(t *testing.T) {	540		func TestAltXClosesFinalWindowAndDetaches(t *testing.T) {
465		d, sess, ac, sends, releases := newManualWindowSession(t, 1)	541		d, sess, ac, sends, releases := newManualWindowSession(t, 1)
466		defer releases[0]()	542		defer releases[0]()

@@ -486,7 +562,7 @@ func TestPTYEOFClosesActiveNonFinalWindowAndRepaintsRemaining(t *testing.T) {
486		go d.ptyReader(sess, sess.windows[0])	562		go d.ptyReader(sess, sess.windows[0])
487		releasePTY1()	563		releasePTY1()
488			564
489	-	require.Eventually(t, func() bool { return len(sess.windows) == 1 }, 2time.Second, 5time.Millisecond)	565	+	require.Eventually(t, func() bool { return windowCount(sess) == 1 }, 2time.Second, 5time.Millisecond)
490		require.Equal(t, 1, sessionCount(d))	566		require.Equal(t, 1, sessionCount(d))
491		require.Equal(t, 0, activeWindowIndex(sess))	567		require.Equal(t, 0, activeWindowIndex(sess))
492		f := awaitFrame(t, sends, ports.MsgOutput)	568		f := awaitFrame(t, sends, ports.MsgOutput)

@@ -512,7 +588,7 @@ func TestPTYEOFClosesInactiveNonFinalWindowAndRepaintsStatus(t *testing.T) {
512		go d.ptyReader(sess, sess.windows[1])	588		go d.ptyReader(sess, sess.windows[1])
513		releasePTY2()	589		releasePTY2()
514			590
515	-	require.Eventually(t, func() bool { return len(sess.windows) == 1 }, 2time.Second, 5time.Millisecond)	591	+	require.Eventually(t, func() bool { return windowCount(sess) == 1 }, 2time.Second, 5time.Millisecond)
516		require.Equal(t, 1, sessionCount(d))	592		require.Equal(t, 1, sessionCount(d))
517		require.Equal(t, 0, activeWindowIndex(sess))	593		require.Equal(t, 0, activeWindowIndex(sess))
518		f := awaitFrame(t, sends, ports.MsgOutput)	594		f := awaitFrame(t, sends, ports.MsgOutput)