Commit c58eb5b

fix(pipe): cleanly terminate pipe sessions when peer disconnects

Previously, when one party in a bidirectional `pipe` disconnected, the
broker treated the remaining client as an active publisher (because its
direction is `InputOutput`), leaving the remaining client hanging in a
half-closed state with frozen terminal input.

- broker: unblock `Connect()` immediately when `client.Done` closes and
  trigger `client.Cleanup()` on read/write errors
- client: close underlying `ReadWriter` in `Cleanup()` if it implements `io.Closer`
- pipe/cli: implement `Close()` on `throttledMonitorRW`

 1@@ -932,6 +932,13 @@ func (t *throttledMonitorRW) Write(p []byte) (int, error) {
 2 	return n, err
 3 }
 4 
 5+func (t *throttledMonitorRW) Close() error {
 6+	if closer, ok := t.rw.(io.Closer); ok {
 7+		return closer.Close()
 8+	}
 9+	return nil
10+}
11+
12 func (handler *CliHandler) sub(cmd *CliCmd, topic string, clientID string) error {
13 	subCmd := flagSet("sub", cmd.sesh)
14 	access := subCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")

  1@@ -26,6 +26,7 @@ import (
  2 
  3 type TestDB struct {
  4 	*stub.StubDB
  5+	mu           sync.RWMutex
  6 	Users        []*db.User
  7 	Pubkeys      []*db.PublicKey
  8 	Features     []*db.FeatureFlag
  9@@ -39,36 +40,51 @@ func NewTestDB(logger *slog.Logger) *TestDB {
 10 }
 11 
 12 func (t *TestDB) FindUserByPubkey(key string) (*db.User, error) {
 13+	t.mu.RLock()
 14+	defer t.mu.RUnlock()
 15 	for _, pk := range t.Pubkeys {
 16 		if pk.Key == key {
 17-			return t.FindUser(pk.UserID)
 18+			return t.findUserLocked(pk.UserID)
 19 		}
 20 	}
 21 	return nil, fmt.Errorf("user not found for pubkey")
 22 }
 23 
 24-func (t *TestDB) FindUser(userID string) (*db.User, error) {
 25+func (t *TestDB) findUserLocked(userID string) (*db.User, error) {
 26 	for _, user := range t.Users {
 27 		if user.ID == userID {
 28-			return user, nil
 29+			cp := *user
 30+			return &cp, nil
 31 		}
 32 	}
 33 	return nil, fmt.Errorf("user not found")
 34 }
 35 
 36+func (t *TestDB) FindUser(userID string) (*db.User, error) {
 37+	t.mu.RLock()
 38+	defer t.mu.RUnlock()
 39+	return t.findUserLocked(userID)
 40+}
 41+
 42 func (t *TestDB) FindUserByName(name string) (*db.User, error) {
 43+	t.mu.RLock()
 44+	defer t.mu.RUnlock()
 45 	for _, user := range t.Users {
 46 		if user.Name == name {
 47-			return user, nil
 48+			cp := *user
 49+			return &cp, nil
 50 		}
 51 	}
 52 	return nil, fmt.Errorf("user not found")
 53 }
 54 
 55 func (t *TestDB) FindFeature(userID, name string) (*db.FeatureFlag, error) {
 56+	t.mu.RLock()
 57+	defer t.mu.RUnlock()
 58 	for _, ff := range t.Features {
 59 		if ff.UserID == userID && ff.Name == name {
 60-			return ff, nil
 61+			cp := *ff
 62+			return &cp, nil
 63 		}
 64 	}
 65 	return nil, fmt.Errorf("feature not found")
 66@@ -91,18 +107,31 @@ func (t *TestDB) Close() error {
 67 }
 68 
 69 func (t *TestDB) AddUser(user *db.User) {
 70-	t.Users = append(t.Users, user)
 71+	t.mu.Lock()
 72+	defer t.mu.Unlock()
 73+	cp := *user
 74+	t.Users = append(t.Users, &cp)
 75 }
 76 
 77 func (t *TestDB) AddPubkey(pubkey *db.PublicKey) {
 78-	t.Pubkeys = append(t.Pubkeys, pubkey)
 79+	t.mu.Lock()
 80+	defer t.mu.Unlock()
 81+	cp := *pubkey
 82+	t.Pubkeys = append(t.Pubkeys, &cp)
 83 }
 84 
 85 func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
 86+	t.mu.Lock()
 87+	defer t.mu.Unlock()
 88+	var winEndCopy *time.Time
 89+	if winEnd != nil {
 90+		w := *winEnd
 91+		winEndCopy = &w
 92+	}
 93 	for _, m := range t.PipeMonitors {
 94 		if m.UserId == userID && m.Topic == topic {
 95 			m.WindowDur = dur
 96-			m.WindowEnd = winEnd
 97+			m.WindowEnd = winEndCopy
 98 			now := time.Now()
 99 			m.UpdatedAt = &now
100 			return nil
101@@ -114,7 +143,7 @@ func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winE
102 		UserId:    userID,
103 		Topic:     topic,
104 		WindowDur: dur,
105-		WindowEnd: winEnd,
106+		WindowEnd: winEndCopy,
107 		CreatedAt: &now,
108 		UpdatedAt: &now,
109 	})
110@@ -122,9 +151,16 @@ func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winE
111 }
112 
113 func (t *TestDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
114+	t.mu.Lock()
115+	defer t.mu.Unlock()
116+	var lastPingCopy *time.Time
117+	if lastPing != nil {
118+		p := *lastPing
119+		lastPingCopy = &p
120+	}
121 	for _, m := range t.PipeMonitors {
122 		if m.UserId == userID && m.Topic == topic {
123-			m.LastPing = lastPing
124+			m.LastPing = lastPingCopy
125 			now := time.Now()
126 			m.UpdatedAt = &now
127 			return nil
128@@ -134,6 +170,8 @@ func (t *TestDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.
129 }
130 
131 func (t *TestDB) RemovePipeMonitor(userID, topic string) error {
132+	t.mu.Lock()
133+	defer t.mu.Unlock()
134 	for i, m := range t.PipeMonitors {
135 		if m.UserId == userID && m.Topic == topic {
136 			t.PipeMonitors = append(t.PipeMonitors[:i], t.PipeMonitors[i+1:]...)
137@@ -143,20 +181,48 @@ func (t *TestDB) RemovePipeMonitor(userID, topic string) error {
138 	return fmt.Errorf("monitor not found")
139 }
140 
141+func copyPipeMonitor(m *db.PipeMonitor) *db.PipeMonitor {
142+	if m == nil {
143+		return nil
144+	}
145+	cp := *m
146+	if m.WindowEnd != nil {
147+		w := *m.WindowEnd
148+		cp.WindowEnd = &w
149+	}
150+	if m.LastPing != nil {
151+		p := *m.LastPing
152+		cp.LastPing = &p
153+	}
154+	if m.CreatedAt != nil {
155+		c := *m.CreatedAt
156+		cp.CreatedAt = &c
157+	}
158+	if m.UpdatedAt != nil {
159+		u := *m.UpdatedAt
160+		cp.UpdatedAt = &u
161+	}
162+	return &cp
163+}
164+
165 func (t *TestDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
166+	t.mu.RLock()
167+	defer t.mu.RUnlock()
168 	for _, m := range t.PipeMonitors {
169 		if m.UserId == userID && m.Topic == topic {
170-			return m, nil
171+			return copyPipeMonitor(m), nil
172 		}
173 	}
174 	return nil, fmt.Errorf("monitor not found")
175 }
176 
177 func (t *TestDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
178+	t.mu.RLock()
179+	defer t.mu.RUnlock()
180 	var monitors []*db.PipeMonitor
181 	for _, m := range t.PipeMonitors {
182 		if m.UserId == userID {
183-			monitors = append(monitors, m)
184+			monitors = append(monitors, copyPipeMonitor(m))
185 		}
186 	}
187 	return monitors, nil
188@@ -643,6 +709,23 @@ func TestPipe_Bidirectional(t *testing.T) {
189 	if !strings.Contains(string(aliceReceived[:n]), "hello from bob") {
190 		t.Errorf("alice did not receive bob's message, got: %q", string(aliceReceived[:n]))
191 	}
192+
193+	// When alice disconnects, bob's session should terminate cleanly without hanging
194+	_ = aliceStdin.Close()
195+	_ = aliceSession.Close()
196+	_ = bobStdin.Close()
197+
198+	bobDone := make(chan error, 1)
199+	go func() {
200+		bobDone <- bobSession.Wait()
201+	}()
202+
203+	select {
204+	case <-bobDone:
205+		// Bob's session terminated cleanly
206+	case <-time.After(3 * time.Second):
207+		t.Fatal("bob's pipe session hung after alice disconnected")
208+	}
209 }
210 
211 func TestPipe_AutoGeneratedTopic(t *testing.T) {

 1@@ -108,14 +108,18 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
 2 
 3 			client.Cleanup()
 4 
 5-			count := 0
 6+			inputCount := 0
 7+			pipeCount := 0
 8 			for _, cl := range dataChannel.GetClients() {
 9-				if cl.Direction == ChannelDirectionInput || cl.Direction == ChannelDirectionInputOutput {
10-					count++
11+				switch cl.Direction {
12+				case ChannelDirectionInput:
13+					inputCount++
14+				case ChannelDirectionInputOutput:
15+					pipeCount++
16 				}
17 			}
18 
19-			if count == 0 {
20+			if inputCount == 0 && pipeCount <= 1 {
21 				for _, cl := range dataChannel.GetClients() {
22 					if !cl.KeepAlive {
23 						otherChannels := 0
24@@ -200,6 +204,7 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
25 				sendwg.Wait()
26 
27 				if err != nil {
28+					client.Cleanup()
29 					if errors.Is(err, io.EOF) {
30 						return
31 					}
32@@ -222,6 +227,7 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
33 					_, err := client.ReadWriter.Write(data.Data)
34 					if err != nil {
35 						outputErr = err
36+						client.Cleanup()
37 						break mainLoop
38 					}
39 
40@@ -235,7 +241,16 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
41 		}()
42 	}
43 
44-	wg.Wait()
45+	done := make(chan struct{})
46+	go func() {
47+		wg.Wait()
48+		close(done)
49+	}()
50+
51+	select {
52+	case <-done:
53+	case <-client.Done:
54+	}
55 
56 	return inputErr, outputErr
57 }

1@@ -48,5 +48,8 @@ func (c *Client) GetChannels() iter.Seq2[string, *Channel] {
2 func (c *Client) Cleanup() {
3 	c.once.Do(func() {
4 		close(c.Done)
5+		if closer, ok := c.ReadWriter.(io.Closer); ok {
6+			_ = closer.Close()
7+		}
8 	})
9 }