pico

fix(pipe): cleanly terminate pipe sessions when peer disconnects rss
created pr with 136.1 on 2026-08-15T18:06:19Z · by c8ef7d19
cmds
checkout latest patchset: 
ssh pr.pico.sh print 136 | git am -3
checkout any patchset in a patch request: 
ssh pr.pico.sh print 136.[rev] | git am -3
add changes to patch request: 
git format-patch main --stdout | ssh pr.pico.sh pr add 136

Patchset 136.1 on 2026-08-15T18:06:19Z · commit c58eb5b

→ fix(pipe): cleanly terminate pipe sessions when peer disconnects
Eric Bower 2026-08-15T17:41:43Z
 
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`
Semantic diff summary
3 added, 14 modified, 0 signature changed, 0 removed across 4 analyzed files
pkg/apps/pipe/cli.go
pkg/apps/pipe/ssh_test.go
pkg/pubsub/broker.go
pkg/pubsub/client.go
+7 -0 pkg/apps/pipe/cli.go #
......@@ -932,6 +932,13 @@ func (t *throttledMonitorRW) Write(p []byte) (int, error) {
932932 return n, err
933933 }
934934
935+func (t *throttledMonitorRW) Close() error {
936+ if closer, ok := t.rw.(io.Closer); ok {
937+ return closer.Close()
938+ }
939+ return nil
940+}
941+
935942 func (handler *CliHandler) sub(cmd *CliCmd, topic string, clientID string) error {
936943 subCmd := flagSet("sub", cmd.sesh)
937944 access := subCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")
+95 -12 pkg/apps/pipe/ssh_test.go #
......@@ -26,6 +26,7 @@ import (
2626
2727 type TestDB struct {
2828 *stub.StubDB
29+ mu sync.RWMutex
2930 Users []*db.User
3031 Pubkeys []*db.PublicKey
3132 Features []*db.FeatureFlag
......@@ -39,36 +40,51 @@ func NewTestDB(logger *slog.Logger) *TestDB {
3940 }
4041
4142 func (t *TestDB) FindUserByPubkey(key string) (*db.User, error) {
43+ t.mu.RLock()
44+ defer t.mu.RUnlock()
4245 for _, pk := range t.Pubkeys {
4346 if pk.Key == key {
44- return t.FindUser(pk.UserID)
47+ return t.findUserLocked(pk.UserID)
4548 }
4649 }
4750 return nil, fmt.Errorf("user not found for pubkey")
4851 }
4952
50-func (t *TestDB) FindUser(userID string) (*db.User, error) {
53+func (t *TestDB) findUserLocked(userID string) (*db.User, error) {
5154 for _, user := range t.Users {
5255 if user.ID == userID {
53- return user, nil
56+ cp := *user
57+ return &cp, nil
5458 }
5559 }
5660 return nil, fmt.Errorf("user not found")
5761 }
5862
63+func (t *TestDB) FindUser(userID string) (*db.User, error) {
64+ t.mu.RLock()
65+ defer t.mu.RUnlock()
66+ return t.findUserLocked(userID)
67+}
68+
5969 func (t *TestDB) FindUserByName(name string) (*db.User, error) {
70+ t.mu.RLock()
71+ defer t.mu.RUnlock()
6072 for _, user := range t.Users {
6173 if user.Name == name {
62- return user, nil
74+ cp := *user
75+ return &cp, nil
6376 }
6477 }
6578 return nil, fmt.Errorf("user not found")
6679 }
6780
6881 func (t *TestDB) FindFeature(userID, name string) (*db.FeatureFlag, error) {
82+ t.mu.RLock()
83+ defer t.mu.RUnlock()
6984 for _, ff := range t.Features {
7085 if ff.UserID == userID && ff.Name == name {
71- return ff, nil
86+ cp := *ff
87+ return &cp, nil
7288 }
7389 }
7490 return nil, fmt.Errorf("feature not found")
......@@ -91,18 +107,31 @@ func (t *TestDB) Close() error {
91107 }
92108
93109 func (t *TestDB) AddUser(user *db.User) {
94- t.Users = append(t.Users, user)
110+ t.mu.Lock()
111+ defer t.mu.Unlock()
112+ cp := *user
113+ t.Users = append(t.Users, &cp)
95114 }
96115
97116 func (t *TestDB) AddPubkey(pubkey *db.PublicKey) {
98- t.Pubkeys = append(t.Pubkeys, pubkey)
117+ t.mu.Lock()
118+ defer t.mu.Unlock()
119+ cp := *pubkey
120+ t.Pubkeys = append(t.Pubkeys, &cp)
99121 }
100122
101123 func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
124+ t.mu.Lock()
125+ defer t.mu.Unlock()
126+ var winEndCopy *time.Time
127+ if winEnd != nil {
128+ w := *winEnd
129+ winEndCopy = &w
130+ }
102131 for _, m := range t.PipeMonitors {
103132 if m.UserId == userID && m.Topic == topic {
104133 m.WindowDur = dur
105- m.WindowEnd = winEnd
134+ m.WindowEnd = winEndCopy
106135 now := time.Now()
107136 m.UpdatedAt = &now
108137 return nil
......@@ -114,7 +143,7 @@ func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winE
114143 UserId: userID,
115144 Topic: topic,
116145 WindowDur: dur,
117- WindowEnd: winEnd,
146+ WindowEnd: winEndCopy,
118147 CreatedAt: &now,
119148 UpdatedAt: &now,
120149 })
......@@ -122,9 +151,16 @@ func (t *TestDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winE
122151 }
123152
124153 func (t *TestDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
154+ t.mu.Lock()
155+ defer t.mu.Unlock()
156+ var lastPingCopy *time.Time
157+ if lastPing != nil {
158+ p := *lastPing
159+ lastPingCopy = &p
160+ }
125161 for _, m := range t.PipeMonitors {
126162 if m.UserId == userID && m.Topic == topic {
127- m.LastPing = lastPing
163+ m.LastPing = lastPingCopy
128164 now := time.Now()
129165 m.UpdatedAt = &now
130166 return nil
......@@ -134,6 +170,8 @@ func (t *TestDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.
134170 }
135171
136172 func (t *TestDB) RemovePipeMonitor(userID, topic string) error {
173+ t.mu.Lock()
174+ defer t.mu.Unlock()
137175 for i, m := range t.PipeMonitors {
138176 if m.UserId == userID && m.Topic == topic {
139177 t.PipeMonitors = append(t.PipeMonitors[:i], t.PipeMonitors[i+1:]...)
......@@ -143,20 +181,48 @@ func (t *TestDB) RemovePipeMonitor(userID, topic string) error {
143181 return fmt.Errorf("monitor not found")
144182 }
145183
184+func copyPipeMonitor(m *db.PipeMonitor) *db.PipeMonitor {
185+ if m == nil {
186+ return nil
187+ }
188+ cp := *m
189+ if m.WindowEnd != nil {
190+ w := *m.WindowEnd
191+ cp.WindowEnd = &w
192+ }
193+ if m.LastPing != nil {
194+ p := *m.LastPing
195+ cp.LastPing = &p
196+ }
197+ if m.CreatedAt != nil {
198+ c := *m.CreatedAt
199+ cp.CreatedAt = &c
200+ }
201+ if m.UpdatedAt != nil {
202+ u := *m.UpdatedAt
203+ cp.UpdatedAt = &u
204+ }
205+ return &cp
206+}
207+
146208 func (t *TestDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
209+ t.mu.RLock()
210+ defer t.mu.RUnlock()
147211 for _, m := range t.PipeMonitors {
148212 if m.UserId == userID && m.Topic == topic {
149- return m, nil
213+ return copyPipeMonitor(m), nil
150214 }
151215 }
152216 return nil, fmt.Errorf("monitor not found")
153217 }
154218
155219 func (t *TestDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
220+ t.mu.RLock()
221+ defer t.mu.RUnlock()
156222 var monitors []*db.PipeMonitor
157223 for _, m := range t.PipeMonitors {
158224 if m.UserId == userID {
159- monitors = append(monitors, m)
225+ monitors = append(monitors, copyPipeMonitor(m))
160226 }
161227 }
162228 return monitors, nil
......@@ -643,6 +709,23 @@ func TestPipe_Bidirectional(t *testing.T) {
643709 if !strings.Contains(string(aliceReceived[:n]), "hello from bob") {
644710 t.Errorf("alice did not receive bob's message, got: %q", string(aliceReceived[:n]))
645711 }
712+
713+ // When alice disconnects, bob's session should terminate cleanly without hanging
714+ _ = aliceStdin.Close()
715+ _ = aliceSession.Close()
716+ _ = bobStdin.Close()
717+
718+ bobDone := make(chan error, 1)
719+ go func() {
720+ bobDone <- bobSession.Wait()
721+ }()
722+
723+ select {
724+ case <-bobDone:
725+ // Bob's session terminated cleanly
726+ case <-time.After(3 * time.Second):
727+ t.Fatal("bob's pipe session hung after alice disconnected")
728+ }
646729 }
647730
648731 func TestPipe_AutoGeneratedTopic(t *testing.T) {
+20 -5 pkg/pubsub/broker.go #
......@@ -108,14 +108,18 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
108108
109109 client.Cleanup()
110110
111- count := 0
111+ inputCount := 0
112+ pipeCount := 0
112113 for _, cl := range dataChannel.GetClients() {
113- if cl.Direction == ChannelDirectionInput || cl.Direction == ChannelDirectionInputOutput {
114- count++
114+ switch cl.Direction {
115+ case ChannelDirectionInput:
116+ inputCount++
117+ case ChannelDirectionInputOutput:
118+ pipeCount++
115119 }
116120 }
117121
118- if count == 0 {
122+ if inputCount == 0 && pipeCount <= 1 {
119123 for _, cl := range dataChannel.GetClients() {
120124 if !cl.KeepAlive {
121125 otherChannels := 0
......@@ -200,6 +204,7 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
200204 sendwg.Wait()
201205
202206 if err != nil {
207+ client.Cleanup()
203208 if errors.Is(err, io.EOF) {
204209 return
205210 }
......@@ -222,6 +227,7 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
222227 _, err := client.ReadWriter.Write(data.Data)
223228 if err != nil {
224229 outputErr = err
230+ client.Cleanup()
225231 break mainLoop
226232 }
227233
......@@ -235,7 +241,16 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
235241 }()
236242 }
237243
238- wg.Wait()
244+ done := make(chan struct{})
245+ go func() {
246+ wg.Wait()
247+ close(done)
248+ }()
249+
250+ select {
251+ case <-done:
252+ case <-client.Done:
253+ }
239254
240255 return inputErr, outputErr
241256 }
+3 -0 pkg/pubsub/client.go #
......@@ -48,5 +48,8 @@ func (c *Client) GetChannels() iter.Seq2[string, *Channel] {
4848 func (c *Client) Cleanup() {
4949 c.once.Do(func() {
5050 close(c.Done)
51+ if closer, ok := c.ReadWriter.(io.Closer); ok {
52+ _ = closer.Close()
53+ }
5154 })
5255 }
Back to top