From b150640428acf6430a59cb5e219e6e4c3f310043 Mon Sep 17 00:00:00 2001 From: ysyneu Date: Tue, 15 Sep 2026 05:22:10 -0700 Subject: [PATCH 1/2] cli: classify broker control-channel closure vs dial refusal The broker dialer collapsed two distinct handshake failures into one opaque message. A dead control channel (runner gone, or channel reclaimed after the spawning command finished) surfaced as "broker handshake send: broken pipe" or the misleading "broker refused connection (code [])" on recv EOF, while a live broker declining the dial (0xFF byte) used the same "refused" wording. - Add exported sentinel ErrBrokerClosed; Sendmsg EPIPE/ECONNREFUSED and Recvmsg EOF (n==0) now wrap it, so errors.Is keeps working through the http.Transport/url.Error wrapping. - The 0xFF refusal gets its own clear message and no longer prints a bogus "code"; other unexpected response bytes get an explicit hex byte instead. - Add dialer-level tests for EOF, EPIPE, and refusal classification. --- internal/cli/broker_dial_unix.go | 30 ++++++++- internal/cli/broker_dial_unix_test.go | 92 +++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 2 deletions(-) diff --git a/internal/cli/broker_dial_unix.go b/internal/cli/broker_dial_unix.go index 4e631d2..d530d01 100644 --- a/internal/cli/broker_dial_unix.go +++ b/internal/cli/broker_dial_unix.go @@ -19,6 +19,15 @@ import ( // never returns nil), but defaultNewClient references it on every platform. var errBrokerUnsupported = errors.New("flashduty: broker mode is not supported on this platform") +// ErrBrokerClosed is returned (wrapped) when the runner-side broker control +// channel is gone: the runner exited, or reclaimed the channel once the +// command that started this process finished, so fduty calls from a +// long-lived background process fail this way. A live broker declining a +// dial is a different failure (see the 0xFF path in dial). Callers and +// automation can tell the two apart with errors.Is(err, ErrBrokerClosed); +// the wrapping by http.Transport and url.Error preserves that. +var ErrBrokerClosed = errors.New("flashduty: broker control channel closed: the broker that started this process is no longer available") + // brokerEgressCapable reports whether this build can act as a broker-mode client // (read FLASHDUTY_CRED_FD and dial over the inherited control fd). The runner // probes it via `fduty version --json` and only advertises broker mode to safari @@ -39,6 +48,14 @@ func (d *brokerDialer) dial(_ context.Context, _, _ string) (net.Conn, error) { defer d.mu.Unlock() if err := syscall.Sendmsg(d.credFD, []byte{0x01}, nil, nil, 0); err != nil { + // A dead control channel surfaces on send as EPIPE (stream-style + // sockets) or ECONNREFUSED (datagram-style sockets, where the closed + // peer answers the datagram): classify both as ErrBrokerClosed so + // callers can tell "broker gone" apart from "broker alive but declined + // this dial" (the 0xFF path below). + if errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNREFUSED) { + return nil, fmt.Errorf("%w (handshake send: %v)", ErrBrokerClosed, err) + } return nil, fmt.Errorf("broker handshake send: %w", err) } body := make([]byte, 1) @@ -47,8 +64,17 @@ func (d *brokerDialer) dial(_ context.Context, _, _ string) (net.Conn, error) { if err != nil { return nil, fmt.Errorf("broker handshake recv: %w", err) } - if n < 1 || body[0] != 0x01 { - return nil, fmt.Errorf("broker refused connection (code %v)", body[:n]) + if n == 0 { + // Orderly EOF: the broker closed the control channel. + return nil, fmt.Errorf("%w (handshake recv: connection closed by peer)", ErrBrokerClosed) + } + if body[0] == 0xFF { + // ctrlRespErr: the broker is alive but declined this dial (e.g. it + // could not mint a connection). Deliberately not ErrBrokerClosed. + return nil, errors.New("flashduty: broker refused the dial request") + } + if body[0] != 0x01 { + return nil, fmt.Errorf("broker handshake: unexpected response byte 0x%02x", body[0]) } scms, err := syscall.ParseSocketControlMessage(oob[:oobn]) if err != nil { diff --git a/internal/cli/broker_dial_unix_test.go b/internal/cli/broker_dial_unix_test.go index 6f5d13a..aa3aaa0 100644 --- a/internal/cli/broker_dial_unix_test.go +++ b/internal/cli/broker_dial_unix_test.go @@ -5,6 +5,7 @@ package cli import ( "bufio" "context" + "errors" "io" "net" "net/http" @@ -12,6 +13,7 @@ import ( "net/url" "os" "strconv" + "strings" "sync" "syscall" "testing" @@ -242,6 +244,96 @@ func TestBrokerHTTPClient_RefusedReturnsError(t *testing.T) { _ = syscall.Close(parentFD) } +// TestBrokerDialer_RecvEOF_ClassifiesBrokerClosed covers a broker that closes +// the control channel mid-handshake: Recvmsg returns 0 (orderly EOF) and the +// dial must classify as ErrBrokerClosed with a message that explains the +// channel is gone. +func TestBrokerDialer_RecvEOF_ClassifiesBrokerClosed(t *testing.T) { + // SOCK_STREAM because closing the peer end must wake the dialer's blocked + // Recvmsg with EOF on both darwin and Linux; datagram sockets carry no EOF + // signalling. Production SEQPACKET hits this same n==0 code path. + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + childFD, parentFD := pair[0], pair[1] + defer func() { _ = syscall.Close(childFD) }() + + parentGone := make(chan struct{}) + go func() { + defer close(parentGone) + defer func() { _ = syscall.Close(parentFD) }() + buf := make([]byte, 1) + // Consume the handshake byte, then close: the dialer's Recvmsg + // returns EOF. + _, _, _, _, _ = syscall.Recvmsg(parentFD, buf, nil, 0) + }() + + d := &brokerDialer{credFD: childFD} + _, err = d.dial(context.Background(), "", "") + <-parentGone + if !errors.Is(err, ErrBrokerClosed) { + t.Fatalf("dial after broker close: want ErrBrokerClosed, got: %v", err) + } + if msg := err.Error(); !strings.Contains(msg, "broker control channel") { + t.Fatalf("user-facing message must mention the closed control channel, got: %v", msg) + } +} + +// TestBrokerDialer_SendEPIPE_ClassifiesBrokerClosed covers a broker whose +// control channel is already gone when the handshake starts: Sendmsg fails +// with EPIPE (stream sockets) and the dial must classify as ErrBrokerClosed. +func TestBrokerDialer_SendEPIPE_ClassifiesBrokerClosed(t *testing.T) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + childFD, parentFD := pair[0], pair[1] + defer func() { _ = syscall.Close(childFD) }() + _ = syscall.Close(parentFD) // peer gone before the handshake + + d := &brokerDialer{credFD: childFD} + _, err = d.dial(context.Background(), "", "") + if !errors.Is(err, ErrBrokerClosed) { + t.Fatalf("dial with dead broker: want ErrBrokerClosed, got: %v", err) + } + if msg := err.Error(); !strings.Contains(msg, "no longer available") { + t.Fatalf("user-facing message must state the broker is gone, got: %v", msg) + } +} + +// TestBrokerDialer_RefusalIsNotBrokerClosed covers a live broker that answers +// the handshake with the 0xFF refusal byte: it must NOT classify as +// ErrBrokerClosed, and the message must read as a refusal. +func TestBrokerDialer_RefusalIsNotBrokerClosed(t *testing.T) { + pair, err := syscall.Socketpair(syscall.AF_UNIX, controlSockType, 0) + if err != nil { + t.Fatalf("socketpair: %v", err) + } + childFD, parentFD := pair[0], pair[1] + defer func() { _ = syscall.Close(childFD) }() + defer func() { _ = syscall.Close(parentFD) }() + + // Queue the refusal before the dial: the dialer's Sendmsg succeeds and its + // Recvmsg consumes the already-buffered 0xFF datagram, so no goroutine is + // needed. + if err := syscall.Sendmsg(parentFD, []byte{0xFF}, nil, nil, 0); err != nil { + t.Fatalf("queue refusal: %v", err) + } + + d := &brokerDialer{credFD: childFD} + _, err = d.dial(context.Background(), "", "") + if err == nil { + t.Fatal("dial must fail when the broker refuses") + } + if errors.Is(err, ErrBrokerClosed) { + t.Fatalf("a live broker's refusal must not classify as ErrBrokerClosed: %v", err) + } + if msg := err.Error(); !strings.Contains(msg, "broker refused the dial request") { + t.Fatalf("user-facing message must say the dial was refused, got: %v", msg) + } +} + // serveProxyConn is a tiny test upstream-proxy used by fakeBroker; the real // implementation lives in the runner, this mirrors it for the CLI test. func serveProxyConn(conn net.Conn, upstream, realKey string) { From 27307243cff8675b0b36d3572098174f08228d1b Mon Sep 17 00:00:00 2001 From: ysyneu Date: Tue, 15 Sep 2026 06:43:03 -0700 Subject: [PATCH 2/2] cli: apply review feedback to broker dial error classification - Demote ErrBrokerClosed to unexported errBrokerClosed: the SDK flattens dial errors into text, so error identity never survives end-to-end and an exported sentinel has no consumer. Classification rides the user-visible message; only in-module tests use errors.Is. - Drop the "flashduty: " prefix from the new messages (the SDK adds it), and give the 0xFF refusal its final actionable wording. - Fix "broker parse rights: %w" rendering %!w() when the rights parse succeeds but yields no fd. - Reword comments to state production errno behavior (Linux SEQPACKET send -> EPIPE; datagram-style peer death -> ECONNREFUSED) and note the sentinel covers the handshake phase only. --- internal/cli/broker_dial_unix.go | 38 +++++++++++++++++---------- internal/cli/broker_dial_unix_test.go | 30 ++++++++++----------- 2 files changed, 39 insertions(+), 29 deletions(-) diff --git a/internal/cli/broker_dial_unix.go b/internal/cli/broker_dial_unix.go index d530d01..1669151 100644 --- a/internal/cli/broker_dial_unix.go +++ b/internal/cli/broker_dial_unix.go @@ -19,14 +19,21 @@ import ( // never returns nil), but defaultNewClient references it on every platform. var errBrokerUnsupported = errors.New("flashduty: broker mode is not supported on this platform") -// ErrBrokerClosed is returned (wrapped) when the runner-side broker control +// errBrokerClosed is returned (wrapped) when the runner-side broker control // channel is gone: the runner exited, or reclaimed the channel once the // command that started this process finished, so fduty calls from a // long-lived background process fail this way. A live broker declining a -// dial is a different failure (see the 0xFF path in dial). Callers and -// automation can tell the two apart with errors.Is(err, ErrBrokerClosed); -// the wrapping by http.Transport and url.Error preserves that. -var ErrBrokerClosed = errors.New("flashduty: broker control channel closed: the broker that started this process is no longer available") +// dial is a different failure (see the 0xFF path in dial). +// +// The distinction rides the user-visible message, not error identity: the +// SDK flattens dial errors into text ("%v"), so only in-module tests +// classify via errors.Is; humans and agents reading the output tell the two +// failures apart by wording. +// +// It covers the handshake phase only. When an already-dispatched connection +// is torn down, in-flight requests see EOF/reset first; the next dial +// reports this error. +var errBrokerClosed = errors.New("broker channel closed: the command that started this process has finished — rerun it in the foreground of a live session") // brokerEgressCapable reports whether this build can act as a broker-mode client // (read FLASHDUTY_CRED_FD and dial over the inherited control fd). The runner @@ -48,13 +55,13 @@ func (d *brokerDialer) dial(_ context.Context, _, _ string) (net.Conn, error) { defer d.mu.Unlock() if err := syscall.Sendmsg(d.credFD, []byte{0x01}, nil, nil, 0); err != nil { - // A dead control channel surfaces on send as EPIPE (stream-style - // sockets) or ECONNREFUSED (datagram-style sockets, where the closed - // peer answers the datagram): classify both as ErrBrokerClosed so - // callers can tell "broker gone" apart from "broker alive but declined + // A dead control channel surfaces on send as EPIPE on the + // production Linux SEQPACKET socket, or ECONNREFUSED when a + // datagram-style peer is gone; classify both as errBrokerClosed so + // "broker gone" reads differently from "broker alive but declined // this dial" (the 0xFF path below). if errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNREFUSED) { - return nil, fmt.Errorf("%w (handshake send: %v)", ErrBrokerClosed, err) + return nil, fmt.Errorf("%w (handshake send: %v)", errBrokerClosed, err) } return nil, fmt.Errorf("broker handshake send: %w", err) } @@ -66,12 +73,12 @@ func (d *brokerDialer) dial(_ context.Context, _, _ string) (net.Conn, error) { } if n == 0 { // Orderly EOF: the broker closed the control channel. - return nil, fmt.Errorf("%w (handshake recv: connection closed by peer)", ErrBrokerClosed) + return nil, fmt.Errorf("%w (recvmsg: EOF)", errBrokerClosed) } if body[0] == 0xFF { // ctrlRespErr: the broker is alive but declined this dial (e.g. it - // could not mint a connection). Deliberately not ErrBrokerClosed. - return nil, errors.New("flashduty: broker refused the dial request") + // could not mint a connection). Deliberately not errBrokerClosed. + return nil, errors.New("broker refused the dial request (no request reached Flashduty; retrying is safe)") } if body[0] != 0x01 { return nil, fmt.Errorf("broker handshake: unexpected response byte 0x%02x", body[0]) @@ -84,9 +91,12 @@ func (d *brokerDialer) dial(_ context.Context, _, _ string) (net.Conn, error) { return nil, fmt.Errorf("broker sent no fd") } fds, err := syscall.ParseUnixRights(&scms[0]) - if err != nil || len(fds) == 0 { + if err != nil { return nil, fmt.Errorf("broker parse rights: %w", err) } + if len(fds) == 0 { + return nil, fmt.Errorf("broker sent no usable fd") + } f := os.NewFile(uintptr(fds[0]), "broker-conn") conn, err := net.FileConn(f) // dups + registers with the netpoller _ = f.Close() diff --git a/internal/cli/broker_dial_unix_test.go b/internal/cli/broker_dial_unix_test.go index aa3aaa0..1ac1ba3 100644 --- a/internal/cli/broker_dial_unix_test.go +++ b/internal/cli/broker_dial_unix_test.go @@ -246,8 +246,8 @@ func TestBrokerHTTPClient_RefusedReturnsError(t *testing.T) { // TestBrokerDialer_RecvEOF_ClassifiesBrokerClosed covers a broker that closes // the control channel mid-handshake: Recvmsg returns 0 (orderly EOF) and the -// dial must classify as ErrBrokerClosed with a message that explains the -// channel is gone. +// dial must classify as errBrokerClosed with an actionable user-visible +// message. func TestBrokerDialer_RecvEOF_ClassifiesBrokerClosed(t *testing.T) { // SOCK_STREAM because closing the peer end must wake the dialer's blocked // Recvmsg with EOF on both darwin and Linux; datagram sockets carry no EOF @@ -272,17 +272,17 @@ func TestBrokerDialer_RecvEOF_ClassifiesBrokerClosed(t *testing.T) { d := &brokerDialer{credFD: childFD} _, err = d.dial(context.Background(), "", "") <-parentGone - if !errors.Is(err, ErrBrokerClosed) { - t.Fatalf("dial after broker close: want ErrBrokerClosed, got: %v", err) + if !errors.Is(err, errBrokerClosed) { + t.Fatalf("dial after broker close: want errBrokerClosed, got: %v", err) } - if msg := err.Error(); !strings.Contains(msg, "broker control channel") { - t.Fatalf("user-facing message must mention the closed control channel, got: %v", msg) + if msg := err.Error(); !strings.Contains(msg, "rerun it in the foreground of a live session") { + t.Fatalf("user-facing message must tell the user how to recover, got: %v", msg) } } // TestBrokerDialer_SendEPIPE_ClassifiesBrokerClosed covers a broker whose // control channel is already gone when the handshake starts: Sendmsg fails -// with EPIPE (stream sockets) and the dial must classify as ErrBrokerClosed. +// with EPIPE (stream sockets) and the dial must classify as errBrokerClosed. func TestBrokerDialer_SendEPIPE_ClassifiesBrokerClosed(t *testing.T) { pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) if err != nil { @@ -294,17 +294,17 @@ func TestBrokerDialer_SendEPIPE_ClassifiesBrokerClosed(t *testing.T) { d := &brokerDialer{credFD: childFD} _, err = d.dial(context.Background(), "", "") - if !errors.Is(err, ErrBrokerClosed) { - t.Fatalf("dial with dead broker: want ErrBrokerClosed, got: %v", err) + if !errors.Is(err, errBrokerClosed) { + t.Fatalf("dial with dead broker: want errBrokerClosed, got: %v", err) } - if msg := err.Error(); !strings.Contains(msg, "no longer available") { - t.Fatalf("user-facing message must state the broker is gone, got: %v", msg) + if msg := err.Error(); !strings.Contains(msg, "the command that started this process has finished") { + t.Fatalf("user-facing message must state why the channel is gone, got: %v", msg) } } // TestBrokerDialer_RefusalIsNotBrokerClosed covers a live broker that answers // the handshake with the 0xFF refusal byte: it must NOT classify as -// ErrBrokerClosed, and the message must read as a refusal. +// errBrokerClosed, and the message must read as a refusal. func TestBrokerDialer_RefusalIsNotBrokerClosed(t *testing.T) { pair, err := syscall.Socketpair(syscall.AF_UNIX, controlSockType, 0) if err != nil { @@ -326,10 +326,10 @@ func TestBrokerDialer_RefusalIsNotBrokerClosed(t *testing.T) { if err == nil { t.Fatal("dial must fail when the broker refuses") } - if errors.Is(err, ErrBrokerClosed) { - t.Fatalf("a live broker's refusal must not classify as ErrBrokerClosed: %v", err) + if errors.Is(err, errBrokerClosed) { + t.Fatalf("a live broker's refusal must not classify as errBrokerClosed: %v", err) } - if msg := err.Error(); !strings.Contains(msg, "broker refused the dial request") { + if msg := err.Error(); !strings.Contains(msg, "broker refused the dial request (no request reached Flashduty; retrying is safe)") { t.Fatalf("user-facing message must say the dial was refused, got: %v", msg) } }