fix(daemon): fence failed route cleanup

commit: 0bca6cde2b574747aa4da76e10207159b66f0d9f
author: brice · 59d ago · 2026-08-24 12:10 UTC
parent: 81f1e15
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +96 -11

M internal/usecase/daemon/handshake.go +1 -8


@@ -153,13 +153,6 @@ func (d Daemon) failHandshakeAttachment(sess session, ac *attachedClient, tr p
153	153		_ = tr.Close()
154	154		}
155	155		if ac.routeCreatedSession {
156		-	sess.mu.Lock()
157		-	empty := len(sess.attachments) == 0
158		-	sess.mu.Unlock()
159		-	if empty {
160		-	// killSession revalidates the attachment set under its publication
161		-	// locks before removing the session.
162		-	_ = d.killSession(sess, ports.ReasonSessionKilled, ac.routeSessionPurge)
163		-	}
	156	+	_ = d.killSessionIfEmpty(sess, ports.ReasonSessionKilled, ac.routeSessionPurge)
164	157		}
165	158		}


@@ -153,13 +153,6 @@ func (d Daemon) failHandshakeAttachment(sess session, ac *attachedClient, tr p
153		_ = tr.Close()	153		_ = tr.Close()
154		}	154		}
155		if ac.routeCreatedSession {	155		if ac.routeCreatedSession {
156	-	sess.mu.Lock()	156	+	_ = d.killSessionIfEmpty(sess, ports.ReasonSessionKilled, ac.routeSessionPurge)
157	-	empty := len(sess.attachments) == 0
158	-	sess.mu.Unlock()
159	-	if empty {
160	-	// killSession revalidates the attachment set under its publication
161	-	// locks before removing the session.
162	-	_ = d.killSession(sess, ports.ReasonSessionKilled, ac.routeSessionPurge)
163	-	}
164		}	157		}
165		}	158		}

M internal/usecase/daemon/remote_attach.go +1 -1


@@ -138,7 +138,7 @@ func (d *Daemon) sendCommittedRouteIdentityForAttachment(token attachmentConnect
138	138		func (d Daemon) finishRouteAttach(sess session, tr ports.Transport, sz domain.Size, h ports.Hello, routeCreated, purge bool) (*attachedClient, error) {
139	139		ac, err := d.finishAttach(sess, tr, sz, h)
140	140		if err != nil && routeCreated {
141		-	if cleanupErr := d.killSession(sess, ports.ReasonSessionKilled, purge); cleanupErr != nil {
	141	+	if cleanupErr := d.killSessionIfEmpty(sess, ports.ReasonSessionKilled, purge); cleanupErr != nil {
142	142		err = errors.Join(err, cleanupErr)
143	143		}
144	144		return nil, err


@@ -138,7 +138,7 @@ func (d *Daemon) sendCommittedRouteIdentityForAttachment(token attachmentConnect
138		func (d Daemon) finishRouteAttach(sess session, tr ports.Transport, sz domain.Size, h ports.Hello, routeCreated, purge bool) (*attachedClient, error) {	138		func (d Daemon) finishRouteAttach(sess session, tr ports.Transport, sz domain.Size, h ports.Hello, routeCreated, purge bool) (*attachedClient, error) {
139		ac, err := d.finishAttach(sess, tr, sz, h)	139		ac, err := d.finishAttach(sess, tr, sz, h)
140		if err != nil && routeCreated {	140		if err != nil && routeCreated {
141	-	if cleanupErr := d.killSession(sess, ports.ReasonSessionKilled, purge); cleanupErr != nil {	141	+	if cleanupErr := d.killSessionIfEmpty(sess, ports.ReasonSessionKilled, purge); cleanupErr != nil {
142		err = errors.Join(err, cleanupErr)	142		err = errors.Join(err, cleanupErr)
143		}	143		}
144		return nil, err	144		return nil, err

M internal/usecase/daemon/remote_route_test.go +75 -0


@@ -50,6 +50,81 @@ func TestFinishRouteAttachRollsBackCreatedSession(t *testing.T) {
50	50		require.False(t, retained, "failed route attach must remove its newly created session")
51	51		}
52	52
	53	+	func TestFinishRouteAttachPreservesConcurrentAttachment(t *testing.T) {
	54	+	pty, release := newBlockingPTY(t)
	55	+	defer release()
	56	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	57	+	sess, err := createSessionForTest(d, "work", false, "/tmp/work", defaultSize, terminalEnv{}, d.baseEnv)
	58	+	require.NoError(t, err)
	59	+	winner, winnerTransport := attachWhenRouteCleanupSnapshots(t, d, sess)
	60	+
	61	+	missing := domain.RemoteSessionTarget{LifecycleID: sess.incarnation, SessionName: "work", LiveTabID: "missing-tab"}
	62	+	d.mu.Lock()
	63	+	_, err = d.finishRouteAttach(sess, &closeTrackingTransport{}, defaultSize, ports.Hello{
	64	+	Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: defaultSize,
	65	+	RemoteTarget: &missing, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
	66	+	}, true, true)
	67	+
	68	+	var protocol *protoErr
	69	+	require.ErrorAs(t, err, &protocol)
	70	+	require.Equal(t, ports.ErrNoSuchTarget, protocol.code)
	71	+	require.Same(t, sess, winner().currentAttachmentSession())
	72	+	d.mu.Lock()
	73	+	require.Same(t, sess, d.sessions[sess.id])
	74	+	d.mu.Unlock()
	75	+	d.clientGone(sess, winner(), winnerTransport, false)
	76	+	}
	77	+
	78	+	func TestFailedHandshakeCleanupPreservesConcurrentAttachment(t *testing.T) {
	79	+	pty, release := newBlockingPTY(t)
	80	+	defer release()
	81	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	82	+	sess, err := createSessionForTest(d, "work", false, "/tmp/work", defaultSize, terminalEnv{}, d.baseEnv)
	83	+	require.NoError(t, err)
	84	+	initialTransport, _ := newCapturingTransport(t)
	85	+	target := domain.RemoteSessionTarget{Endpoint: "arch", DisplayOrigin: "arch", LifecycleID: sess.incarnation, SessionName: "work", LiveTabID: domain.TabStableID(sess.tabs[0].stableID)}
	86	+	_, initial, err := d.routeWithContext(context.Background(), ports.Hello{
	87	+	Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: defaultSize,
	88	+	RemoteTarget: &target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
	89	+	}, initialTransport)
	90	+	require.NoError(t, err)
	91	+	initial.routeCreatedSession = true
	92	+	initial.routeSessionPurge = true
	93	+	winner, winnerTransport := attachWhenRouteCleanupSnapshots(t, d, sess)
	94	+
	95	+	d.failHandshakeAttachment(sess, initial, initialTransport, false)
	96	+
	97	+	require.Same(t, sess, winner().currentAttachmentSession())
	98	+	d.mu.Lock()
	99	+	require.Same(t, sess, d.sessions[sess.id])
	100	+	d.mu.Unlock()
	101	+	d.clientGone(sess, winner(), winnerTransport, false)
	102	+	}
	103	+
	104	+	func attachWhenRouteCleanupSnapshots(t testing.T, d Daemon, sess session) (func() attachedClient, ports.Transport) {
	105	+	t.Helper()
	106	+	winnerTransport, _ := newCapturingTransport(t)
	107	+	var winner *attachedClient
	108	+	d.afterAttachmentEffectParticipantsSnapshotted = func(string, []*attachedClient) {
	109	+	d.afterAttachmentEffectParticipantsSnapshotted = nil
	110	+	target := domain.RemoteSessionTarget{
	111	+	Endpoint: "arch", DisplayOrigin: "arch", LifecycleID: sess.incarnation,
	112	+	SessionName: sess.name, LiveTabID: domain.TabStableID(sess.tabs[0].stableID),
	113	+	}
	114	+	var err error
	115	+	_, winner, err = d.routeWithContext(context.Background(), ports.Hello{
	116	+	Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: sess.name, Size: defaultSize,
	117	+	RemoteTarget: &target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
	118	+	}, winnerTransport)
	119	+	require.NoError(t, err)
	120	+	}
	121	+	return func() *attachedClient {
	122	+	t.Helper()
	123	+	require.NotNil(t, winner)
	124	+	return winner
	125	+	}, winnerTransport
	126	+	}
	127	+
53	128		func TestRouteRemoteTargetRejectsSameNameReplacement(t *testing.T) {
54	129		d := newTestDaemon(t, nil, stubClock{})
55	130		sess := addControlSession(d, "work", "tab-new", "pane-new")


@@ -50,6 +50,81 @@ func TestFinishRouteAttachRollsBackCreatedSession(t *testing.T) {
50	require.False(t, retained, "failed route attach must remove its newly created session")	50		require.False(t, retained, "failed route attach must remove its newly created session")
51	}	51		}
52		52
		53	+	func TestFinishRouteAttachPreservesConcurrentAttachment(t *testing.T) {
		54	+	pty, release := newBlockingPTY(t)
		55	+	defer release()
		56	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
		57	+	sess, err := createSessionForTest(d, "work", false, "/tmp/work", defaultSize, terminalEnv{}, d.baseEnv)
		58	+	require.NoError(t, err)
		59	+	winner, winnerTransport := attachWhenRouteCleanupSnapshots(t, d, sess)
		60	+
		61	+	missing := domain.RemoteSessionTarget{LifecycleID: sess.incarnation, SessionName: "work", LiveTabID: "missing-tab"}
		62	+	d.mu.Lock()
		63	+	_, err = d.finishRouteAttach(sess, &closeTrackingTransport{}, defaultSize, ports.Hello{
		64	+	Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: defaultSize,
		65	+	RemoteTarget: &missing, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
		66	+	}, true, true)
		67	+
		68	+	var protocol *protoErr
		69	+	require.ErrorAs(t, err, &protocol)
		70	+	require.Equal(t, ports.ErrNoSuchTarget, protocol.code)
		71	+	require.Same(t, sess, winner().currentAttachmentSession())
		72	+	d.mu.Lock()
		73	+	require.Same(t, sess, d.sessions[sess.id])
		74	+	d.mu.Unlock()
		75	+	d.clientGone(sess, winner(), winnerTransport, false)
		76	+	}
		77	+
		78	+	func TestFailedHandshakeCleanupPreservesConcurrentAttachment(t *testing.T) {
		79	+	pty, release := newBlockingPTY(t)
		80	+	defer release()
		81	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
		82	+	sess, err := createSessionForTest(d, "work", false, "/tmp/work", defaultSize, terminalEnv{}, d.baseEnv)
		83	+	require.NoError(t, err)
		84	+	initialTransport, _ := newCapturingTransport(t)
		85	+	target := domain.RemoteSessionTarget{Endpoint: "arch", DisplayOrigin: "arch", LifecycleID: sess.incarnation, SessionName: "work", LiveTabID: domain.TabStableID(sess.tabs[0].stableID)}
		86	+	_, initial, err := d.routeWithContext(context.Background(), ports.Hello{
		87	+	Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: defaultSize,
		88	+	RemoteTarget: &target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
		89	+	}, initialTransport)
		90	+	require.NoError(t, err)
		91	+	initial.routeCreatedSession = true
		92	+	initial.routeSessionPurge = true
		93	+	winner, winnerTransport := attachWhenRouteCleanupSnapshots(t, d, sess)
		94	+
		95	+	d.failHandshakeAttachment(sess, initial, initialTransport, false)
		96	+
		97	+	require.Same(t, sess, winner().currentAttachmentSession())
		98	+	d.mu.Lock()
		99	+	require.Same(t, sess, d.sessions[sess.id])
		100	+	d.mu.Unlock()
		101	+	d.clientGone(sess, winner(), winnerTransport, false)
		102	+	}
		103	+
		104	+	func attachWhenRouteCleanupSnapshots(t testing.T, d Daemon, sess session) (func() attachedClient, ports.Transport) {
		105	+	t.Helper()
		106	+	winnerTransport, _ := newCapturingTransport(t)
		107	+	var winner *attachedClient
		108	+	d.afterAttachmentEffectParticipantsSnapshotted = func(string, []*attachedClient) {
		109	+	d.afterAttachmentEffectParticipantsSnapshotted = nil
		110	+	target := domain.RemoteSessionTarget{
		111	+	Endpoint: "arch", DisplayOrigin: "arch", LifecycleID: sess.incarnation,
		112	+	SessionName: sess.name, LiveTabID: domain.TabStableID(sess.tabs[0].stableID),
		113	+	}
		114	+	var err error
		115	+	_, winner, err = d.routeWithContext(context.Background(), ports.Hello{
		116	+	Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: sess.name, Size: defaultSize,
		117	+	RemoteTarget: &target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
		118	+	}, winnerTransport)
		119	+	require.NoError(t, err)
		120	+	}
		121	+	return func() *attachedClient {
		122	+	t.Helper()
		123	+	require.NotNil(t, winner)
		124	+	return winner
		125	+	}, winnerTransport
		126	+	}
		127	+
53	func TestRouteRemoteTargetRejectsSameNameReplacement(t *testing.T) {	128		func TestRouteRemoteTargetRejectsSameNameReplacement(t *testing.T) {
54	d := newTestDaemon(t, nil, stubClock{})	129		d := newTestDaemon(t, nil, stubClock{})
55	sess := addControlSession(d, "work", "tab-new", "pane-new")	130		sess := addControlSession(d, "work", "tab-new", "pane-new")

M internal/usecase/daemon/session.go +19 -2


@@ -1309,7 +1309,7 @@ func (d Daemon) killSessionForAttachment(sess session, reason uint8, purge boo
1309	1309		// architecture locks, but releases every lock before any attachment gate is frozen or
1310	1310		// drained. The initiator is part of the same deduplicated global gate set even
1311	1311		// when it belongs to a different source session.
1312		-	func (d Daemon) snapshotSessionKillParticipants(target session, admission *sessionKillAdmission) (sessionKillParticipants, bool) {
	1312	+	func (d Daemon) snapshotSessionKillParticipants(target session, admission *sessionKillAdmission, requireEmpty bool) (sessionKillParticipants, bool) {
1313	1313		snapshot := sessionKillParticipants{target: target}
1314	1314		if admission != nil {
1315	1315		token := admission.token

@@ -1334,6 +1334,12 @@ func (d Daemon) snapshotSessionKillParticipants(target session, admission *ses
1334	1334		source = target
1335	1335		}
1336	1336		unlockSessions := lockAttachmentSessions(source, target)
	1337	+	if requireEmpty && len(target.attachments) != 0 {
	1338	+	unlockSessions()
	1339	+	d.notices.routingMu.Unlock()
	1340	+	d.mu.Unlock()
	1341	+	return sessionKillParticipants{}, false
	1342	+	}
1337	1343		snapshot.attachments = make(map[*attachedClient]struct{}, len(target.attachments))
1338	1344		snapshot.transports = make(map[*attachedClient]transportSnapshot, len(target.attachments))
1339	1345		for ac := range target.attachments {

@@ -1524,13 +1530,24 @@ func (s *session) finishTeardown() {
1524	1530		// only immutable capture state; it never observes closed worker channels or
1525	1531		// session-owned cache that teardown has released.
1526	1532		func (d Daemon) killSessionWithSnapshotDeadline(sess session, reason uint8, purge bool, deadline snapshotShutdownDeadline, admission sessionKillAdmission) error {
	1533	+	return d.killSessionWithSnapshotDeadlineAndCondition(sess, reason, purge, deadline, admission, false)
	1534	+	}
	1535	+
	1536	+	// killSessionIfEmpty tears down sess only while its exact registry entry has no
	1537	+	// attachments. Membership is captured under the normal architecture locks and
	1538	+	// fenced again at publication, so an attachment winning either side survives.
	1539	+	func (d Daemon) killSessionIfEmpty(sess session, reason uint8, purge bool) error {
	1540	+	return d.killSessionWithSnapshotDeadlineAndCondition(sess, reason, purge, nil, nil, true)
	1541	+	}
	1542	+
	1543	+	func (d Daemon) killSessionWithSnapshotDeadlineAndCondition(sess session, reason uint8, purge bool, deadline snapshotShutdownDeadline, admission sessionKillAdmission, requireEmpty bool) error {
1527	1544		if sess == nil {
1528	1545		return nil
1529	1546		}
1530	1547		// Discover every attachment whose membership can be invalidated before freezing
1531	1548		// any one of them. The common gate helper deduplicates and freezes the whole
1532	1549		// immutable identity set in the same global order used by transitions.
1533		-	participants, ok := d.snapshotSessionKillParticipants(sess, admission)
	1550	+	participants, ok := d.snapshotSessionKillParticipants(sess, admission, requireEmpty)
1534	1551		if !ok {
1535	1552		return nil
1536	1553		}

 // architecture locks, but releases every lock before any attachment gate is frozen or
 // drained. The initiator is part of the same deduplicated global gate set even
 // when it belongs to a different source session.
-func (d *Daemon) snapshotSessionKillParticipants(target *session, admission *sessionKillAdmission) (sessionKillParticipants, bool) {
+func (d *Daemon) snapshotSessionKillParticipants(target *session, admission *sessionKillAdmission, requireEmpty bool) (sessionKillParticipants, bool) {
 	snapshot := sessionKillParticipants{target: target}
 	if admission != nil {
 		token := admission.token
 		source = target
 	}
 	unlockSessions := lockAttachmentSessions(source, target)
+	if requireEmpty && len(target.attachments) != 0 {
+		unlockSessions()
+		d.notices.routingMu.Unlock()
+		d.mu.Unlock()
+		return sessionKillParticipants{}, false
+	}
 	snapshot.attachments = make(map[*attachedClient]struct{}, len(target.attachments))
 	snapshot.transports = make(map[*attachedClient]transportSnapshot, len(target.attachments))
 	for ac := range target.attachments {
 // only immutable capture state; it never observes closed worker channels or
 // session-owned cache that teardown has released.
 func (d *Daemon) killSessionWithSnapshotDeadline(sess *session, reason uint8, purge bool, deadline *snapshotShutdownDeadline, admission *sessionKillAdmission) error {
+	return d.killSessionWithSnapshotDeadlineAndCondition(sess, reason, purge, deadline, admission, false)
+}
+// killSessionIfEmpty tears down sess only while its exact registry entry has no
+// attachments. Membership is captured under the normal architecture locks and
+// fenced again at publication, so an attachment winning either side survives.
+func (d *Daemon) killSessionIfEmpty(sess *session, reason uint8, purge bool) error {
+	return d.killSessionWithSnapshotDeadlineAndCondition(sess, reason, purge, nil, nil, true)
+}
+func (d *Daemon) killSessionWithSnapshotDeadlineAndCondition(sess *session, reason uint8, purge bool, deadline *snapshotShutdownDeadline, admission *sessionKillAdmission, requireEmpty bool) error {
 	if sess == nil {
 		return nil
 	}
 	// Discover every attachment whose membership can be invalidated before freezing
 	// any one of them. The common gate helper deduplicates and freezes the whole
 	// immutable identity set in the same global order used by transitions.
-	participants, ok := d.snapshotSessionKillParticipants(sess, admission)
+	participants, ok := d.snapshotSessionKillParticipants(sess, admission, requireEmpty)
 	if !ok {
 		return nil
 	}