diff --git a/go.mod b/go.mod index dcdc55df..f7554f88 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( charm.land/bubbletea/v2 v2.0.9 charm.land/glamour/v2 v2.0.1 charm.land/lipgloss/v2 v2.0.6 - github.com/basecamp/actioncable-go v0.0.0-20260824145920-822e6cf08655 + github.com/basecamp/actioncable-go v1.0.0 github.com/basecamp/hey-sdk/go v0.29.0 github.com/basecamp/mcp v0.0.0-20260828100356-2d6f44b51e9d github.com/charmbracelet/x/ansi v0.11.8 diff --git a/go.sum b/go.sum index 1718d44d..4109caf1 100644 --- a/go.sum +++ b/go.sum @@ -87,8 +87,8 @@ github.com/aymanbagabas/go-udiff v0.4.1 h1:OEIrQ8maEeDBXQDoGCbbTTXYJMYRCRO1fnodZ github.com/aymanbagabas/go-udiff v0.4.1/go.mod h1:0L9PGwj20lrtmEMeyw4WKJ/TMyDtvAoK9bf2u/mNo3w= github.com/aymerick/douceur v0.2.0 h1:Mv+mAeH1Q+n9Fr+oyamOlAkUNPWPlA8PPGR0QAaYuPk= github.com/aymerick/douceur v0.2.0/go.mod h1:wlT5vV2O3h55X9m7iVYN0TBM0NH/MmbLnd30/FjWUq4= -github.com/basecamp/actioncable-go v0.0.0-20260824145920-822e6cf08655 h1:zz0WUSEmjURj0T+soXuTtgX291nYouqa+UoyYY3Xxk8= -github.com/basecamp/actioncable-go v0.0.0-20260824145920-822e6cf08655/go.mod h1:ezaV5z1GXQAsqyejqTs6wCFl2D8Wj+COLQkHc/kwoRs= +github.com/basecamp/actioncable-go v1.0.0 h1:m8UGFYfBfa/YivQNayc2VoS+oVKLG+ixKzhIzdACWDM= +github.com/basecamp/actioncable-go v1.0.0/go.mod h1:ezaV5z1GXQAsqyejqTs6wCFl2D8Wj+COLQkHc/kwoRs= github.com/basecamp/hey-sdk/go v0.29.0 h1:/u9iD5x2Rm2IqVpF55XDfGMNwVyvtC+rJjpYC9BrFeo= github.com/basecamp/hey-sdk/go v0.29.0/go.mod h1:k6sO2XhMkU3UY8lD2ozp0735Ic3q8xoMQt7YUT3TlYk= github.com/basecamp/mcp v0.0.0-20260828100356-2d6f44b51e9d h1:zEQVGq1x1nhKMZ2TudFAcSJ32CHT8richI1vQakIKz4= diff --git a/internal/cable/cable.go b/internal/cable/cable.go index 6cddfdf7..3f7b2910 100644 --- a/internal/cable/cable.go +++ b/internal/cable/cable.go @@ -9,6 +9,7 @@ import ( "net/url" "os" "strings" + "time" "github.com/basecamp/actioncable-go" @@ -16,6 +17,13 @@ import ( "github.com/basecamp/hey-cli/internal/version" ) +// openTimeout is how long Dial keeps trying to open a connection before it reports +// why it couldn't. Credentials the server won't take, and a server that isn't there, +// both fail every attempt, and a command has to say so rather than retry under a +// caller who gave it no deadline at all. It bounds the opening and nothing else: a +// connection that got through outlives it, and reconnects on its own until Close. +const openTimeout = 15 * time.Second + // Dial connects to the cable server for a HEY base URL, authorizing the upgrade // request with the same credentials the SDK sends on an API request. // @@ -29,24 +37,17 @@ func Dial(ctx context.Context, baseURL string, authMgr *auth.Manager, options .. return nil, err } - // The first dial's header is taken here so that credentials the server won't take - // are reported now, rather than becoming a reconnect loop inside the client. - if _, err := authHeader(ctx, baseURL, authMgr); err != nil { - return nil, err - } - settings := make([]actioncable.Option, 0, 1+len(options)) settings = append(settings, actioncable.WithHeaderFunc(func(ctx context.Context) (http.Header, error) { return authHeader(ctx, baseURL, authMgr) })) settings = append(settings, options...) + opening, giveUp := context.WithTimeout(ctx, openTimeout) + defer giveUp() + client := actioncable.New(cableURL, settings...) - if err := client.Connect(ctx); err != nil { - // Connect's context bounds the caller's wait rather than the client's lifetime. - // A dial that did not complete has no owner to close it, so stop its retry loop - // before returning the error. - _ = client.Close() + if err := client.Connect(opening); err != nil { return nil, err } diff --git a/internal/cable/cable_test.go b/internal/cable/cable_test.go index 727f3ece..ed76fc65 100644 --- a/internal/cable/cable_test.go +++ b/internal/cable/cable_test.go @@ -6,6 +6,7 @@ import ( "net/http" "os" "slices" + "strings" "sync" "testing" "time" @@ -121,8 +122,8 @@ func TestEveryDialCarriesCurrentCredentials(t *testing.T) { t.Errorf("redial Origin = %q, want the client's own headers kept", got) } - // Connect's deadline only bounds its wait; Dial owns and stops a client that never - // connected so no retry goroutine is left behind after the error. + // A dial that gave up leaves nothing behind it: the client stops itself rather + // than retrying under a caller who has already been handed the error. dialsAtReturn := len(headers) time.Sleep(10 * time.Millisecond) if got := len(recorded.recorded()); got != dialsAtReturn { @@ -130,12 +131,25 @@ func TestEveryDialCarriesCurrentCredentials(t *testing.T) { } } -func TestDialWithoutCredentialsFailsBeforeConnecting(t *testing.T) { +func TestDialWithoutCredentialsSaysSo(t *testing.T) { t.Setenv("HEY_NO_KEYRING", "1") t.Setenv("HEY_TOKEN", "") - _, err := Dial(t.Context(), "https://app.hey.com", auth.NewManager("https://app.hey.com", http.DefaultClient, t.TempDir())) + recorded := &recordingTransport{dialed: make(chan struct{}, 1)} + dialing, stopDialing := context.WithTimeout(t.Context(), 200*time.Millisecond) + defer stopDialing() + + _, err := Dial(dialing, "https://app.hey.com", auth.NewManager("https://app.hey.com", http.DefaultClient, t.TempDir()), + actioncable.WithTransport(recorded), actioncable.WithBackoff(time.Millisecond, time.Millisecond)) if err == nil { t.Fatal("expected a dial with no credentials to fail") } + // The upgrade request is never built without credentials, so the reason has to + // come back with the error rather than being retried away out of sight. + if !strings.Contains(err.Error(), "not authenticated") { + t.Errorf("error = %q, want it to name the credentials that could not be built", err.Error()) + } + if dials := len(recorded.recorded()); dials != 0 { + t.Errorf("dials = %d, want no upgrade request attempted without credentials", dials) + } } diff --git a/internal/cmd/tui_watch.go b/internal/cmd/tui_watch.go index 840fd21b..811bbc57 100644 --- a/internal/cmd/tui_watch.go +++ b/internal/cmd/tui_watch.go @@ -2,7 +2,6 @@ package cmd import ( "context" - "errors" "sync" "time" @@ -23,11 +22,12 @@ const mailChangeBacklog = 16 // reconnectBacklog is one, because a Screener reconnect always says the same thing. It // is a channel of its own so that the relay goroutine is the only writer to the channel -// it closes: the cable client drains callbacks queued before it was told to stop, and a -// send on a closed channel panics whatever the select around it says — off a goroutine -// Bubble Tea knows nothing about, which takes the terminal down in raw mode. Nothing -// closes this one, so a callback arriving after the relay is gone rings into the buffer -// and is collected with it. +// it closes: the relay ends with the watch's context and closes as it goes, while the +// subscription it was reading lives until the goodbye behind it, so a callback can still +// fire once the relay is gone. A send on a closed channel panics whatever the select +// around it says — off a goroutine Bubble Tea knows nothing about, which takes the +// terminal down in raw mode. Nothing closes this one, so a late callback rings into the +// buffer and is collected with it. const reconnectBacklog = 1 type mailConnectionNotifier struct { @@ -58,16 +58,10 @@ func (n *mailConnectionNotifier) after(version uint64) (tui.MailWatchEvent, uint return n.event, n.version, n.version != version } -const ( - // unsubscribeTimeout bounds the goodbye sent for a watch that is over. Nothing waits - // on it, and a connection that has gone away is reason to stop trying rather than hang. - unsubscribeTimeout = 5 * time.Second - - // tuiCableDialTimeout turns an unreachable cable server into app state instead of - // leaving the startup command inside Action Cable's retry loop forever. The model - // owns the retries after this first bounded attempt. - tuiCableDialTimeout = 5 * time.Second -) +// tuiCableDialTimeout turns an unreachable cable server into app state instead of +// leaving the startup command inside Action Cable's retry loop for as long as the +// package allows. The model owns the retries after this first bounded attempt. +const tuiCableDialTimeout = 5 * time.Second // tuiWatchers are the streams `hey tui` follows to stay live. func tuiWatchers() tui.Watchers { @@ -98,7 +92,7 @@ func watchMailChanges(ctx context.Context) (<-chan tui.MailWatchEvent, error) { events := make(chan tui.MailWatchEvent, mailChangeBacklog) go func() { - defer unsubscribe(ctx, subscription) + defer unsubscribe(subscription) relayMailChanges(ctx, subscription.Messages(), connection, events) }() @@ -156,7 +150,7 @@ func watchScreenerChanges(ctx, connectionCtx context.Context, signedStreamName s changes := make(chan struct{}, 1) go func() { - defer unsubscribe(ctx, subscription) + defer unsubscribe(subscription) relayScreenerChanges(ctx, subscription.Messages(), reconnects, changes) }() @@ -260,7 +254,7 @@ func (w *calendarStreamWatch) subscribe(ctx, connectionCtx context.Context, cale w.stops[calendar.Calendar.Id] = stop go func() { - defer unsubscribe(subCtx, subscription) + defer unsubscribe(subscription) for { select { case <-subCtx.Done(): @@ -353,12 +347,9 @@ func (w *calendarStreamWatch) stopAll() { // unsubscribe drops a subscription whose watch is over. Cancelling the watch's context // ends the relay, but the subscription itself belongs to the shared client: left // registered it holds its buffered channel and its callback dispatcher for as long as the -// TUI runs, and goes on being handed messages nobody reads. The watch's context is what -// ended, so the goodbye is sent under one that outlives it. -func unsubscribe(ctx context.Context, subscription *actioncable.Subscription) { - goodbye, giveUp := context.WithTimeout(context.WithoutCancel(ctx), unsubscribeTimeout) - defer giveUp() - _ = subscription.Unsubscribe(goodbye) +// TUI runs, and goes on being handed messages nobody reads. +func unsubscribe(subscription *actioncable.Subscription) { + _ = subscription.Unsubscribe() } // ringMailWatchEvent keeps connection state ahead of stale box doorbells. Box events can @@ -423,9 +414,8 @@ func ring[T any](notifications chan<- T, notification T) { } // tuiSubscribe subscribes over the connection the TUI's watches share, dialling a new one -// when the one on hand has stopped itself. A stopped client preserves its terminal failure -// and never dials again on its own, so a reopened stream replaces it with a connection that -// carries current credentials. +// when the one on hand has stopped itself. A stopped client never dials again, so a +// reopened stream replaces it with a connection that carries current credentials. func tuiSubscribe(ctx, connectionCtx context.Context, identifier actioncable.Identifier, options ...actioncable.SubscriptionOption) (*actioncable.Subscription, error) { client, err := tuiCableClient(connectionCtx) if err != nil { @@ -445,17 +435,17 @@ func tuiSubscribe(ctx, connectionCtx context.Context, identifier actioncable.Ide return subscription, err } -// subscribeTuiCable returns stopped when the shared client needs replacing. Every shared -// client has connected before it is cached, so Connect reports ErrAlreadyConnected while -// it is live or reconnecting and preserves the terminal failure after it stops. +// subscribeTuiCable returns stopped when the shared client needs replacing. A client that +// has stopped keeps why for good and never dials again, which Err reports and every other +// failure — a rejection, a subscribe the context ran out on — leaves nil. func subscribeTuiCable(ctx context.Context, client *actioncable.Client, identifier actioncable.Identifier, options ...actioncable.SubscriptionOption) (*actioncable.Subscription, bool, error) { subscription, err := client.Subscribe(ctx, identifier, options...) if err == nil { return subscription, false, nil } - stoppedBecause := client.Connect(ctx) - if errors.Is(stoppedBecause, actioncable.ErrAlreadyConnected) { + stoppedBecause := client.Err() + if stoppedBecause == nil { return nil, false, err } diff --git a/internal/cmd/tui_watch_test.go b/internal/cmd/tui_watch_test.go index 9de31f21..540e925a 100644 --- a/internal/cmd/tui_watch_test.go +++ b/internal/cmd/tui_watch_test.go @@ -78,10 +78,11 @@ func TestRelayScreenerChangesRingsOnEveryBroadcast(t *testing.T) { } func TestARelayIsTheOnlyWriterToTheStreamItCloses(t *testing.T) { - // The cable client runs the callbacks it queued before it was told to stop, so a - // reconnect can be announced after the relay closed the channel the TUI reads. - // Sending on a closed channel panics, off a goroutine Bubble Tea can't recover, - // which leaves the terminal in raw mode — so only the relay may write to it. + // A relay ends with its watch's context while the subscription it was reading is + // still registered, so a reconnect can be announced after the relay closed the + // channel the TUI reads. Sending on a closed channel panics, off a goroutine Bubble + // Tea can't recover, which leaves the terminal in raw mode — so only the relay may + // write to it. relaying, stop := context.WithCancel(t.Context()) mailMessages := make(chan actioncable.Message) @@ -195,6 +196,36 @@ func TestSubscribeTuiCableRecognizesAnUnenumeratedTerminalFailure(t *testing.T) } } +func TestSubscribeTuiCableKeepsALiveClientThatTurnedASubscriptionDown(t *testing.T) { + // A channel that says no is about that one subscription. The connection under it is + // still good and still carrying the TUI's other watches, so throwing it away would + // cost every one of them a reconnect over a stream that was never going to open. + conn := newScriptedCableConn() + conn.reads <- []byte(`{"type":"welcome"}`) + client := actioncable.New("ws://cable.example.test/cable", actioncable.WithTransport(scriptedCableTransport{conn: conn})) + if err := client.Connect(t.Context()); err != nil { + t.Fatalf("connect: %v", err) + } + defer client.Close() + tuiCable.client = client + t.Cleanup(func() { tuiCable.client = nil }) + + go func() { + conn.reads <- []byte(`{"type":"reject_subscription","identifier":"{\"channel\":\"Postings::ChangesChannel\"}"}`) + }() + + _, stopped, err := subscribeTuiCable(t.Context(), client, actioncable.Identifier{Channel: changesChannel}) + if stopped { + t.Fatal("a rejection should not condemn the connection the other watches share") + } + if !errors.Is(err, actioncable.ErrRejected) { + t.Errorf("error = %v, want the rejection reported as it is", err) + } + if tuiCable.client != client { + t.Error("a live client should stay cached after a rejected subscription") + } +} + func TestServerStoppedActionCableClientIsReplaceable(t *testing.T) { conn := newScriptedCableConn() conn.reads <- []byte(`{"type":"welcome"}`) diff --git a/internal/cmd/watch.go b/internal/cmd/watch.go index 0a720e1d..cc4f005c 100644 --- a/internal/cmd/watch.go +++ b/internal/cmd/watch.go @@ -16,7 +16,6 @@ import ( "strconv" "strings" "sync" - "sync/atomic" "syscall" "time" @@ -201,8 +200,7 @@ func (c *watchCommand) run(cmd *cobra.Command, args []string) error { watch.noteConnection(true) } }), - actioncable.OnDisconnected(func(willReconnect bool) { watch.noteConnection(false) }), - actioncable.OnRejected(func() { watch.rejected.Store(true) })) + actioncable.OnDisconnected(func(willReconnect bool) { watch.noteConnection(false) })) if err != nil { return apierr.ErrAPI(0, fmt.Sprintf("could not subscribe to posting changes: %v", err)) } @@ -423,7 +421,6 @@ type postingsWatch struct { connection chan struct{} transitionsMu sync.Mutex transitions []bool - rejected atomic.Bool catchingUp bool unread map[int64]bool backoff time.Duration @@ -464,7 +461,7 @@ func (w *postingsWatch) listen(ctx context.Context, subscription *actioncable.Su } case message, open := <-subscription.Messages(): if !open { - return w.closedError(ctx) + return w.closedError(ctx, subscription.Err()) } if err := w.read(ctx, message); err != nil { return err @@ -475,18 +472,19 @@ func (w *postingsWatch) listen(ctx context.Context, subscription *actioncable.Su return nil } -// closedError tells the two ways the subscription's messages dry up apart: the watch was +// closedError tells the ways the subscription's messages dry up apart: the watch was // interrupted or timed out, which is how it's meant to end, or the connection went away // for good and there is nothing left listening — which a watch left running unattended -// has to hear about rather than exiting quietly. -func (w *postingsWatch) closedError(ctx context.Context) error { +// has to hear about rather than exiting quietly. ended is what the subscription says +// closed it. +func (w *postingsWatch) closedError(ctx context.Context, ended error) error { switch { case ctx.Err() != nil: return nil //nolint:nilerr // an interrupt or a --timeout is how a watch is meant to end - case w.rejected.Load(): + case errors.Is(ended, actioncable.ErrRejected): return apierr.ErrAuth("HEY's cable server turned this subscription down — run `hey auth login` again, or log in with `hey auth login --cookie` if the server doesn't take access tokens on a websocket yet") default: - return apierr.ErrNetwork(errors.New("HEY's cable server hung up for good — nothing is watching for changes any more")) + return apierr.ErrNetwork(fmt.Errorf("HEY's cable server hung up for good — nothing is watching for changes any more: %w", ended)) } } diff --git a/internal/cmd/watch_calendar.go b/internal/cmd/watch_calendar.go index c91995a1..35b26fd5 100644 --- a/internal/cmd/watch_calendar.go +++ b/internal/cmd/watch_calendar.go @@ -251,7 +251,7 @@ func (w *postingsWatch) subscribeCalendar(ctx context.Context, calendar *watched id := calendar.id go func() { - defer unsubscribe(subCtx, subscription) + defer unsubscribe(subscription) for { select { case <-subCtx.Done(): diff --git a/internal/cmd/watch_test.go b/internal/cmd/watch_test.go index 5da2a78a..8fab476b 100644 --- a/internal/cmd/watch_test.go +++ b/internal/cmd/watch_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "io" "net/http" "net/http/httptest" @@ -708,11 +709,11 @@ func TestWatchClosedSubscriptionIsOnlyFineWhenItWasInterrupted(t *testing.T) { interrupted, interrupt := context.WithCancel(context.Background()) interrupt() - if err := watch.closedError(interrupted); err != nil { + if err := watch.closedError(interrupted, actioncable.ErrUnsubscribed); err != nil { t.Errorf("error = %v, want an interrupted watch to end cleanly", err) } - err := watch.closedError(context.Background()) + err := watch.closedError(context.Background(), actioncable.ErrClosed) if err == nil { t.Fatal("a connection that went away for good should be reported") } @@ -720,8 +721,7 @@ func TestWatchClosedSubscriptionIsOnlyFineWhenItWasInterrupted(t *testing.T) { t.Errorf("error = %q, want it to say the server hung up", err.Error()) } - watch.rejected.Store(true) - err = watch.closedError(context.Background()) + err = watch.closedError(context.Background(), fmt.Errorf("%w: %s", actioncable.ErrRejected, changesChannel)) if err == nil || !strings.Contains(err.Error(), "turned this subscription down") { t.Errorf("error = %v, want a rejected subscription reported as an auth failure", err) } diff --git a/nix/package.nix b/nix/package.nix index 398a0162..78566181 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -18,7 +18,7 @@ buildGoModule.override { inherit go; } (finalAttrs: { # To update: run `make update-nix-hash` (Docker). It rewrites this quoted # value in place, so keep it a string literal rather than lib.fakeHash. - vendorHash = "sha256-Nupd+16J+aXwNnstS6W86jjx1Usuwsw4FJSU6tdcQ3o="; + vendorHash = "sha256-Lm+yjYS2EFgsbzc3E0HSSKyk4+zM/JBAULVrgI69yHk="; subPackages = [ "cmd/hey" ];