Commit 08f4e64

Antonio Mika  ·  2026-10-10 22:46:19 -0400 EDT
parent af2fdb0 
fix(pipe): send an exit status when commands end
3 files changed,  +87, -4
M pkg/apps/pipe/api.go
M pkg/apps/pipe/ssh_test.go
M pkg/pubsub/client.go
M pkg/apps/pipe/api.go
+1, -1
......@@ -217,7 +217,7 @@ func (web *webRouter) handlePub(pubsub bool) http.HandlerFunc {
217217 }
218218
219219 // pub sends the message once its stdin ends, and exits when it has.
220- // pipe closes its sessions without an exit status.
220+ // Older pipe servers close the session without an exit status.
221221 _ = p.CloseWrite()
222222 var missing *ssh.ExitMissingError
223223 if err := p.Wait(); err != nil && !errors.As(err, &missing) {
M pkg/apps/pipe/ssh_test.go
+84, -0
......@@ -5,6 +5,7 @@ import (
55 "context"
66 "crypto/ed25519"
77 "crypto/rand"
8+ "errors"
89 "fmt"
910 "io"
1011 "log/slog"
......@@ -2459,3 +2460,86 @@ func TestSSH_WildcardSub(t *testing.T) {
24592460 t.Errorf("expected SSH wildcard subscriber output to contain 'prose-event', got: %q", output)
24602461 }
24612462 }
2463+
2464+// runStatus runs cmd with stdin and returns its exit status, or -1 when the
2465+// session ended without one.
2466+func runStatus(t *testing.T, client *ssh.Client, cmd, stdin string) int {
2467+ t.Helper()
2468+ session, err := client.NewSession()
2469+ if err != nil {
2470+ t.Fatal(err)
2471+ }
2472+ defer func() { _ = session.Close() }()
2473+ session.Stdin = strings.NewReader(stdin)
2474+ return exitStatus(t, session.Run(cmd))
2475+}
2476+
2477+func exitStatus(t *testing.T, err error) int {
2478+ t.Helper()
2479+ var exitErr *ssh.ExitError
2480+ var missing *ssh.ExitMissingError
2481+ switch {
2482+ case err == nil:
2483+ return 0
2484+ case errors.As(err, &exitErr):
2485+ return exitErr.ExitStatus()
2486+ case errors.As(err, &missing):
2487+ return -1
2488+ }
2489+ t.Fatalf("command failed: %v", err)
2490+ return 0
2491+}
2492+
2493+func TestCommandsExitWithStatus(t *testing.T) {
2494+ server := NewTestSSHServer(t)
2495+ defer server.Shutdown()
2496+
2497+ alice := GenerateUser("alice")
2498+ charlie := GenerateUser("charlie")
2499+ RegisterUserWithServer(server, alice)
2500+ RegisterUserWithServer(server, charlie)
2501+ aliceClient, err := alice.NewClient()
2502+ if err != nil {
2503+ t.Fatal(err)
2504+ }
2505+ defer func() { _ = aliceClient.Close() }()
2506+ charlieClient, err := charlie.NewClient()
2507+ if err != nil {
2508+ t.Fatal(err)
2509+ }
2510+ defer func() { _ = charlieClient.Close() }()
2511+
2512+ if got := runStatus(t, aliceClient, "pub nobody -b=false", "hello"); got != 0 {
2513+ t.Errorf("pub exited %d, want 0", got)
2514+ }
2515+
2516+ sub, err := aliceClient.NewSession()
2517+ if err != nil {
2518+ t.Fatal(err)
2519+ }
2520+ defer func() { _ = sub.Close() }()
2521+ if err := sub.Start("sub news -a alice"); err != nil {
2522+ t.Fatal(err)
2523+ }
2524+ server.waitForClients(t, 1)
2525+ if got := runStatus(t, aliceClient, "pub news", "hello"); got != 0 {
2526+ t.Errorf("pub to a sub exited %d, want 0", got)
2527+ }
2528+ if got := exitStatus(t, sub.Wait()); got != 0 {
2529+ t.Errorf("sub exited %d after its publisher left, want 0", got)
2530+ }
2531+
2532+ if err := (func() error {
2533+ s, err := aliceClient.NewSession()
2534+ if err != nil {
2535+ return err
2536+ }
2537+ return s.Start("sub restricted -p -a alice -k")
2538+ })(); err != nil {
2539+ t.Fatal(err)
2540+ }
2541+ server.waitForClients(t, 1)
2542+ if got := runStatus(t, charlieClient, "sub restricted -p", ""); got != 1 {
2543+ t.Errorf("denied sub exited %d, want 1", got)
2544+ }
2545+}
M pkg/pubsub/client.go
+2, -3
......@@ -48,12 +48,11 @@ func (c *Client) GetChannels() iter.Seq2[string, *Channel] {
4848 return c.Channels.Range
4949 }
5050
51+// Cleanup ends the client. Its ReadWriter stays open for whoever opened it
52+// to close, so an ssh session can still send its exit status.
5153 func (c *Client) Cleanup() {
5254 c.once.Do(func() {
5355 close(c.Done)
54- if closer, ok := c.ReadWriter.(io.Closer); ok {
55- _ = closer.Close()
56- }
5756 })
5857 }
5958