From 414dcc7ced7c3f6f7a49db55aae9e5e4f4ff7e6a Mon Sep 17 00:00:00 2001 From: Sourya Vatsyayan Date: Fri, 14 Aug 2026 16:59:53 +0530 Subject: [PATCH] fix: back off and retry when Slack rate limits the channel listing conversations.list is a Tier 2 method, so paginating a large workspace trips Slack's rate limit partway through. The listing treated the 429 as a fatal 5xx and abandoned every page it had already fetched, which made the Slack integration impossible to install on those workspaces. Rate limited pages are now retried against the same cursor, honouring Retry-After within a cap. Once retries are exhausted the pages already fetched are returned rather than failing, since a partial channel list still lets the install complete. Slack's `ok: false` application errors on a 200 are also detected now, where previously they surfaced as an empty channel list. Also fixes handleHTTPFailure, where a dead `err != nil` branch under `StatusCode > 500` classified every failure as permanent and logged it as a 5xx, the 429s included. Ref ENG-4785 --- provider/slack/client.go | 156 +++++++++++++++++-- provider/slack/client_test.go | 279 +++++++++++++++++++++++++++++++--- 2 files changed, 401 insertions(+), 34 deletions(-) diff --git a/provider/slack/client.go b/provider/slack/client.go index 8fb7c99..272c1d1 100644 --- a/provider/slack/client.go +++ b/provider/slack/client.go @@ -7,6 +7,9 @@ import ( "io" "net/http" "net/url" + "strconv" + "strings" + "time" "github.com/deepsourcelabs/hermes/domain" "github.com/deepsourcelabs/hermes/provider" @@ -18,6 +21,17 @@ const postMessageURL = "https://slack.com/api/chat.postMessage" type Client struct { HTTPClient provider.IHTTPClient + + // Sleep is swapped out in tests so backoff does not slow them down. + Sleep func(time.Duration) +} + +func (c *Client) sleep(d time.Duration) { + if c.Sleep != nil { + c.Sleep(d) + return + } + time.Sleep(d) } type SendMessageRequest struct { @@ -71,19 +85,41 @@ func handleHTTPFailure(response *http.Response) domain.IError { return errFailedSendPermanent(err.Error()) } - if response.StatusCode > 500 { - if err != nil { - log.Errorf("slack: failed with 5xx response code: %v", err) - return errFailedSendTemporary(fmt.Sprintf("received 5xx, error=%s", string(b))) - } + body := strings.TrimSpace(string(b)) + + // Rate limits and server errors are worth retrying. Everything else is a + // permanent failure for this request. + if response.StatusCode == http.StatusTooManyRequests || response.StatusCode >= 500 { + log.Errorf("slack: retryable failure, status=%d error=%s", response.StatusCode, body) + return errFailedSendTemporary(fmt.Sprintf("received %d, error=%s", response.StatusCode, body)) } - log.Errorf("slack: failed with 5xx response code: %v", err) - return errFailedSendPermanent(fmt.Sprintf("received 5xx, error=%s", string(b))) + log.Errorf("slack: permanent failure, status=%d error=%s", response.StatusCode, body) + return errFailedSendPermanent(fmt.Sprintf("received %d, error=%s", response.StatusCode, body)) } const getChannelsURL = "https://slack.com/api/conversations.list?types=public_channel,private_channel&exclude_archived=true&limit=1000" +const ( + // conversations.list is a Slack Tier 2 method, which allows roughly 20 + // requests a minute. A workspace large enough to need several pages will + // trip that limit mid-pagination, so the listing has to back off and retry + // instead of abandoning the whole thing. + maxRateLimitRetries = 2 + + // Slack's Retry-After for Tier 2 methods is often 30s or more. This runs + // inside the synchronous OAuth callback, so the wait is capped and partial + // results are preferred over holding the request open indefinitely. + maxRetryAfter = 15 * time.Second + defaultRetryAfter = 5 * time.Second + + // Runaway guard in case Slack keeps handing back a next_cursor. + maxChannelPages = 200 + + // The `error` Slack sets on a rate limited response body. + slackRateLimitedError = "ratelimited" +) + type GetChannelsRequest struct { BearerToken string `json:"_"` } @@ -99,10 +135,41 @@ type ResponseMetadata struct { type GetChannelsResponse struct { Ok bool `json:"ok"` + Error string `json:"error"` Channels []Channel `json:"channels"` ResponseMetadata ResponseMetadata `json:"response_metadata"` } +// rateLimitedError marks a channel page fetch that Slack rate limited, and +// carries the wait Slack asked for so pagination can retry the same cursor. +type rateLimitedError struct { + domain.IError + retryAfter time.Duration +} + +func newRateLimitedError(retryAfter time.Duration, internal string) *rateLimitedError { + return &rateLimitedError{IError: errFailedOptsFetch(internal), retryAfter: retryAfter} +} + +func isRateLimited(err domain.IError) bool { + _, ok := err.(*rateLimitedError) + return ok +} + +// retryAfterFrom reads Slack's Retry-After header, falling back to a default +// when it is missing or unparseable, and clamping it to maxRetryAfter. +func retryAfterFrom(header http.Header) time.Duration { + seconds, err := strconv.Atoi(strings.TrimSpace(header.Get("Retry-After"))) + if err != nil || seconds <= 0 { + return defaultRetryAfter + } + + if retryAfter := time.Duration(seconds) * time.Second; retryAfter < maxRetryAfter { + return retryAfter + } + return maxRetryAfter +} + func (c *Client) getChannelsPage(request *GetChannelsRequest, cursor string) (*GetChannelsResponse, domain.IError) { var response = new(GetChannelsResponse) @@ -127,26 +194,85 @@ func (c *Client) getChannelsPage(request *GetChannelsRequest, cursor string) (*G } defer resp.Body.Close() + if resp.StatusCode == http.StatusTooManyRequests { + b, _ := io.ReadAll(resp.Body) + return response, newRateLimitedError( + retryAfterFrom(resp.Header), + fmt.Sprintf("slack rate limited the channel listing, error=%s", strings.TrimSpace(string(b))), + ) + } + if resp.StatusCode < 200 || resp.StatusCode > 399 { - log.Errorf("slack: Non 2xx response while fetching options: %v", err) + log.Errorf("slack: non-2xx response while fetching options: status=%d", resp.StatusCode) return response, handleHTTPFailure(resp) } if err := json.NewDecoder(resp.Body).Decode(response); err != nil { - log.Errorf("slack: Non 2xx response while fetching options: %v", err) + log.Errorf("slack: failed decoding options response: %v", err) return response, errFailedOptsFetch(err.Error()) } + // Slack reports application level failures as `ok: false` on a 200, so the + // status code alone is not enough to tell whether the page came back. + if !response.Ok { + if response.Error == slackRateLimitedError { + return response, newRateLimitedError( + retryAfterFrom(resp.Header), + "slack rate limited the channel listing", + ) + } + log.Errorf("slack: channel listing failed with error=%s", response.Error) + return response, errFailedOptsFetch(fmt.Sprintf("slack returned error=%s", response.Error)) + } + return response, nil } +// getChannelsPageWithBackoff fetches a single page, retrying a bounded number +// of times while Slack rate limits us. Non rate limit failures are returned +// straight away. +func (c *Client) getChannelsPageWithBackoff(request *GetChannelsRequest, cursor string) (*GetChannelsResponse, domain.IError) { + for attempt := 0; ; attempt++ { + response, err := c.getChannelsPage(request, cursor) + if err == nil { + return response, nil + } + + rateLimited, ok := err.(*rateLimitedError) + if !ok { + return response, err + } + + if attempt >= maxRateLimitRetries { + return response, rateLimited + } + + log.Warnf( + "slack: rate limited fetching channel page %q, retrying in %v (attempt %d of %d)", + cursor, rateLimited.retryAfter, attempt+1, maxRateLimitRetries, + ) + c.sleep(rateLimited.retryAfter) + } +} + func (c *Client) GetChannels(request *GetChannelsRequest) ([]map[string]string, domain.IError) { - var channels []map[string]string + channels := make([]map[string]string, 0) cursor := "" - for { - response, err := c.getChannelsPage(request, cursor) + for page := 0; page < maxChannelPages; page++ { + response, err := c.getChannelsPageWithBackoff(request, cursor) if err != nil { + // Slack kept rate limiting us. The pages that did come back are far + // more useful than a hard failure, which stops the integration from + // being installed at all. + if isRateLimited(err) && len(channels) > 0 { + log.Warnf( + "slack: rate limited while paginating channels, returning the %d channels fetched so far", + len(channels), + ) + return channels, nil + } + log.Errorf("slack: Error fetching page %v: %v", cursor, err) return channels, err } @@ -160,9 +286,13 @@ func (c *Client) GetChannels(request *GetChannelsRequest) ([]map[string]string, cursor = response.ResponseMetadata.NextCursor if cursor == "" { - break + return channels, nil } } + log.Warnf( + "slack: hit the %d page cap while paginating channels, returning %d channels", + maxChannelPages, len(channels), + ) return channels, nil } diff --git a/provider/slack/client_test.go b/provider/slack/client_test.go index 53c412e..50989b9 100644 --- a/provider/slack/client_test.go +++ b/provider/slack/client_test.go @@ -1,40 +1,277 @@ package slack import ( + "io" + "net/http" "reflect" + "strings" "testing" - - "github.com/deepsourcelabs/hermes/domain" - "github.com/deepsourcelabs/hermes/provider" + "time" ) -func TestClient_GetChannels(t *testing.T) { - type fields struct { - HTTPClient provider.IHTTPClient +// stubHTTPClient replays a canned list of responses, one per request, and +// records the request URLs it was asked for. +type stubHTTPClient struct { + responses []*http.Response + requested []string + calls int +} + +func (s *stubHTTPClient) Do(req *http.Request) (*http.Response, error) { + s.requested = append(s.requested, req.URL.String()) + + if s.calls >= len(s.responses) { + s.calls++ + return nil, io.ErrUnexpectedEOF } - type args struct { - request *GetChannelsRequest + + resp := s.responses[s.calls] + s.calls++ + return resp, nil +} + +func response(status int, body string, header http.Header) *http.Response { + if header == nil { + header = http.Header{} + } + return &http.Response{ + StatusCode: status, + Body: io.NopCloser(strings.NewReader(body)), + Header: header, + } +} + +func rateLimitedResponse(retryAfter string) *http.Response { + header := http.Header{} + if retryAfter != "" { + header.Set("Retry-After", retryAfter) + } + return response(http.StatusTooManyRequests, `{"ok":false,"error":"ratelimited"}`, header) +} + +// newTestClient returns a client whose backoff is recorded rather than slept. +func newTestClient(responses ...*http.Response) (*Client, *stubHTTPClient, *[]time.Duration) { + stub := &stubHTTPClient{responses: responses} + var slept []time.Duration + client := &Client{ + HTTPClient: stub, + Sleep: func(d time.Duration) { slept = append(slept, d) }, + } + return client, stub, &slept +} + +func channelNames(channels []map[string]string) []string { + names := make([]string, 0, len(channels)) + for _, channel := range channels { + names = append(names, channel["name"]) + } + return names +} + +func TestClient_GetChannels_PaginatesUntilCursorIsEmpty(t *testing.T) { + client, stub, _ := newTestClient( + response(200, `{"ok":true,"channels":[{"id":"C1","name":"general"}],"response_metadata":{"next_cursor":"page2"}}`, nil), + response(200, `{"ok":true,"channels":[{"id":"C2","name":"random"}]}`, nil), + ) + + got, err := client.GetChannels(&GetChannelsRequest{BearerToken: "xoxb-test"}) + if err != nil { + t.Fatalf("GetChannels() unexpected error = %v", err) } + + if want := []string{"general", "random"}; !reflect.DeepEqual(channelNames(got), want) { + t.Errorf("GetChannels() = %v, want %v", channelNames(got), want) + } + if stub.calls != 2 { + t.Errorf("GetChannels() made %d requests, want 2", stub.calls) + } + if !strings.Contains(stub.requested[1], "cursor=page2") { + t.Errorf("second request = %q, want it to carry cursor=page2", stub.requested[1]) + } +} + +// The regression this fixes: Slack rate limits a page mid-pagination and the +// whole channel listing used to be abandoned, which silently blocked the +// integration from installing. +func TestClient_GetChannels_RetriesRateLimitedPage(t *testing.T) { + client, stub, slept := newTestClient( + response(200, `{"ok":true,"channels":[{"id":"C1","name":"general"}],"response_metadata":{"next_cursor":"page2"}}`, nil), + rateLimitedResponse("3"), + response(200, `{"ok":true,"channels":[{"id":"C2","name":"random"}]}`, nil), + ) + + got, err := client.GetChannels(&GetChannelsRequest{BearerToken: "xoxb-test"}) + if err != nil { + t.Fatalf("GetChannels() unexpected error = %v", err) + } + + if want := []string{"general", "random"}; !reflect.DeepEqual(channelNames(got), want) { + t.Errorf("GetChannels() = %v, want %v", channelNames(got), want) + } + if want := []time.Duration{3 * time.Second}; !reflect.DeepEqual(*slept, want) { + t.Errorf("backoff = %v, want %v", *slept, want) + } + // The retry must re-request the same cursor, not skip the page. + if !strings.Contains(stub.requested[2], "cursor=page2") { + t.Errorf("retry request = %q, want it to carry cursor=page2", stub.requested[2]) + } +} + +func TestClient_GetChannels_RetriesRateLimitedBodyOn200(t *testing.T) { + client, _, slept := newTestClient( + response(200, `{"ok":false,"error":"ratelimited"}`, nil), + response(200, `{"ok":true,"channels":[{"id":"C1","name":"general"}]}`, nil), + ) + + got, err := client.GetChannels(&GetChannelsRequest{BearerToken: "xoxb-test"}) + if err != nil { + t.Fatalf("GetChannels() unexpected error = %v", err) + } + + if want := []string{"general"}; !reflect.DeepEqual(channelNames(got), want) { + t.Errorf("GetChannels() = %v, want %v", channelNames(got), want) + } + if want := []time.Duration{defaultRetryAfter}; !reflect.DeepEqual(*slept, want) { + t.Errorf("backoff = %v, want %v", *slept, want) + } +} + +// Once retries are exhausted, an install is still possible with the pages that +// did come back, so partial results beat a hard failure. +func TestClient_GetChannels_ReturnsPartialResultsWhenRetriesExhausted(t *testing.T) { + client, _, slept := newTestClient( + response(200, `{"ok":true,"channels":[{"id":"C1","name":"general"}],"response_metadata":{"next_cursor":"page2"}}`, nil), + rateLimitedResponse("1"), + rateLimitedResponse("1"), + rateLimitedResponse("1"), + ) + + got, err := client.GetChannels(&GetChannelsRequest{BearerToken: "xoxb-test"}) + if err != nil { + t.Fatalf("GetChannels() unexpected error = %v, want partial success", err) + } + + if want := []string{"general"}; !reflect.DeepEqual(channelNames(got), want) { + t.Errorf("GetChannels() = %v, want %v", channelNames(got), want) + } + if len(*slept) != maxRateLimitRetries { + t.Errorf("backoff attempts = %d, want %d", len(*slept), maxRateLimitRetries) + } +} + +// With nothing at all to show, the rate limit is a real failure. +func TestClient_GetChannels_ErrorsWhenRateLimitedWithNoResults(t *testing.T) { + client, _, _ := newTestClient( + rateLimitedResponse("1"), + rateLimitedResponse("1"), + rateLimitedResponse("1"), + ) + + got, err := client.GetChannels(&GetChannelsRequest{BearerToken: "xoxb-test"}) + if err == nil { + t.Fatalf("GetChannels() error = nil, want a rate limit error") + } + if !isRateLimited(err) { + t.Errorf("GetChannels() error = %v, want it to be a rate limit error", err) + } + if len(got) != 0 { + t.Errorf("GetChannels() = %v, want no channels", got) + } +} + +func TestClient_GetChannels_ErrorsOnSlackApplicationError(t *testing.T) { + client, stub, _ := newTestClient( + response(200, `{"ok":false,"error":"invalid_auth"}`, nil), + ) + + got, err := client.GetChannels(&GetChannelsRequest{BearerToken: "xoxb-test"}) + if err == nil { + t.Fatalf("GetChannels() error = nil, want an error for ok:false") + } + if isRateLimited(err) { + t.Errorf("GetChannels() error = %v, want a non rate limit error", err) + } + if !strings.Contains(err.Error(), "invalid_auth") { + t.Errorf("GetChannels() error = %q, want it to mention invalid_auth", err.Error()) + } + if len(got) != 0 { + t.Errorf("GetChannels() = %v, want no channels", got) + } + // An application error is not retried. + if stub.calls != 1 { + t.Errorf("GetChannels() made %d requests, want 1", stub.calls) + } +} + +// An empty workspace must come back as an empty list rather than a nil one, so +// callers iterating the options do not trip over a null. +func TestClient_GetChannels_ReturnsEmptySliceForNoChannels(t *testing.T) { + client, _, _ := newTestClient( + response(200, `{"ok":true,"channels":[]}`, nil), + ) + + got, err := client.GetChannels(&GetChannelsRequest{BearerToken: "xoxb-test"}) + if err != nil { + t.Fatalf("GetChannels() unexpected error = %v", err) + } + if got == nil { + t.Fatal("GetChannels() = nil, want an empty slice") + } + if len(got) != 0 { + t.Errorf("GetChannels() = %v, want no channels", got) + } +} + +func TestRetryAfterFrom(t *testing.T) { tests := []struct { - name string - fields fields - args args - want interface{} - want1 domain.IError + name string + retryAfter string + want time.Duration }{ - // TODO: Add test cases. + {name: "honours the header", retryAfter: "7", want: 7 * time.Second}, + {name: "falls back when absent", retryAfter: "", want: defaultRetryAfter}, + {name: "falls back when unparseable", retryAfter: "later", want: defaultRetryAfter}, + {name: "falls back when not positive", retryAfter: "0", want: defaultRetryAfter}, + {name: "clamps a long wait", retryAfter: "600", want: maxRetryAfter}, } + for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - c := &Client{ - HTTPClient: tt.fields.HTTPClient, + header := http.Header{} + if tt.retryAfter != "" { + header.Set("Retry-After", tt.retryAfter) + } + if got := retryAfterFrom(header); got != tt.want { + t.Errorf("retryAfterFrom() = %v, want %v", got, tt.want) } - got, got1 := c.GetChannels(tt.args.request) - if !reflect.DeepEqual(got, tt.want) { - t.Errorf("Client.GetChannels() got = %v, want %v", got, tt.want) + }) + } +} + +func TestHandleHTTPFailure(t *testing.T) { + tests := []struct { + name string + status int + wantFatal bool + wantMessage string + }{ + // A 429 used to fall through to the permanent branch and get logged as + // a 5xx, which is what made the rate limit so hard to spot. + {name: "rate limited is retryable", status: http.StatusTooManyRequests, wantFatal: false}, + {name: "server error is retryable", status: http.StatusInternalServerError, wantFatal: false}, + {name: "bad gateway is retryable", status: http.StatusBadGateway, wantFatal: false}, + {name: "unauthorized is permanent", status: http.StatusUnauthorized, wantFatal: true}, + {name: "bad request is permanent", status: http.StatusBadRequest, wantFatal: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := handleHTTPFailure(response(tt.status, `{"ok":false,"error":"boom"}`, nil)) + if err.IsFatal() != tt.wantFatal { + t.Errorf("handleHTTPFailure(%d).IsFatal() = %v, want %v", tt.status, err.IsFatal(), tt.wantFatal) } - if !reflect.DeepEqual(got1, tt.want1) { - t.Errorf("Client.GetChannels() got1 = %v, want %v", got1, tt.want1) + if !strings.Contains(err.Error(), "boom") { + t.Errorf("handleHTTPFailure(%d) internal = %q, want it to include the response body", tt.status, err.Error()) } }) }