From be4d6af80a6f2d81bf8127733cf603aa4c0bae53 Mon Sep 17 00:00:00 2001 From: Paymahn Moghadasian Date: Tue, 8 Sep 2026 19:54:18 -0500 Subject: [PATCH] fix: honour unlisted 1/min history and replies quotas (#5) Unlisted Slack apps get 1 req/min and 15 msgs/page on conversations.history and conversations.replies. Wait Retry-After, cap the page, and share that slot across CLI processes. --- README.md | 9 +- internal/cli/root.go | 2 +- internal/config/config.go | 2 +- pkg/limiter/unlisted.go | 277 +++++++++++++++++++++++++++++++++++ pkg/limiter/unlisted_test.go | 154 +++++++++++++++++++ pkg/provider/api.go | 42 +++++- skills/slack-cli/SKILL.md | 3 + 7 files changed, 484 insertions(+), 5 deletions(-) create mode 100644 pkg/limiter/unlisted.go create mode 100644 pkg/limiter/unlisted_test.go diff --git a/README.md b/README.md index 2ca68f6..6db0817 100644 --- a/README.md +++ b/README.md @@ -119,7 +119,14 @@ added becomes the default. Combining explicit tokens with `--profile` is rejected as ambiguous. Global flags: `--govslack` (route to slack-gov.com), `--no-cache`, `--raw` -(print tool output verbatim), `--verbose`, `--timeout` (default 30s). +(print tool output verbatim), `--verbose`, `--timeout` (default 2m). + +Unlisted (non-Marketplace) Slack apps are capped at **1 req/min and 15 +messages/page** on `conversations.history` / `conversations.replies`. The CLI +caps the page, waits `Retry-After`, and shares that 1/min slot across processes +via a file in the cache dir. Set `SLACK_MCP_UNLISTED_HISTORY=1` to force that +tier without waiting for a 429; `=0` disables it. Paginate with `--cursor` for +the rest of the thread. Default `--timeout` is 2m so one wait fits. ## Cache diff --git a/internal/cli/root.go b/internal/cli/root.go index c6d3340..40fd43b 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -61,7 +61,7 @@ func NewRootCommand() *cobra.Command { f.BoolVar(&cfg.NoCache, "no-cache", false, "Skip user/channel cache; only channel/user IDs resolve (no #name/@name lookup)") f.BoolVar(&cfg.Raw, "raw", false, "Print tool output verbatim (no JSON pretty-print)") f.BoolVarP(&cfg.Verbose, "verbose", "v", false, "Verbose logging to stderr") - f.DurationVar(&cfg.Timeout, "timeout", 30*time.Second, "request timeout") + f.DurationVar(&cfg.Timeout, "timeout", 2*time.Minute, "request timeout (2m so one unlisted history 429 can wait Retry-After)") root.AddCommand(newAuthCommand(&cfg)) cmds.AddCommands(root, &cfg) diff --git a/internal/config/config.go b/internal/config/config.go index 8ecf95d..2a06446 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -27,7 +27,7 @@ type Config struct { } func FromEnv() Config { - return Config{Timeout: 30 * time.Second} + return Config{Timeout: 2 * time.Minute} } // ApplyEnv fills any unset credential field from the environment. diff --git a/pkg/limiter/unlisted.go b/pkg/limiter/unlisted.go new file mode 100644 index 0000000..16df5d3 --- /dev/null +++ b/pkg/limiter/unlisted.go @@ -0,0 +1,277 @@ +package limiter + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// Unlisted commercially-distributed Slack apps get a special tier on +// conversations.history and conversations.replies: 1 request/minute and at +// most 15 objects per page. Marketplace and internal apps stay on Tier 3. +// +// https://docs.slack.dev/changelog/2025/05/29/rate-limit-changes-for-non-marketplace-apps/ +const ( + UnlistedPageSize = 15 + UnlistedInterval = time.Minute +) + +// SLACK_MCP_UNLISTED_HISTORY=1 forces the special tier (cap 15, 1/min) without +// waiting for a 429. =0 disables it even after a long Retry-After. +const unlistedEnv = "SLACK_MCP_UNLISTED_HISTORY" + +type quotaState struct { + Unlisted bool `json:"unlisted"` + LastUnixNano int64 `json:"last_unix_nano"` +} + +// UnlistedQuota is a process-shared special-tier gate for one Slack method +// (conversations.history or conversations.replies) in one workspace. +type UnlistedQuota struct { + Dir string + TeamID string + Method string + Interval time.Duration + PageSize int + MarkAfter time.Duration + now func() time.Time + + force bool + disabled bool +} + +func NewUnlistedQuota(teamID, method string) *UnlistedQuota { + q := &UnlistedQuota{ + TeamID: teamID, + Method: method, + Interval: UnlistedInterval, + PageSize: UnlistedPageSize, + now: time.Now, + } + switch strings.ToLower(strings.TrimSpace(os.Getenv(unlistedEnv))) { + case "1", "true", "yes": + q.force = true + case "0", "false", "no": + q.disabled = true + } + return q +} + +func (q *UnlistedQuota) CapPage(n int) int { + if !q.unlisted() { + return n + } + if q.PageSize <= 0 { + return UnlistedPageSize + } + if n <= 0 || n > q.PageSize { + return q.PageSize + } + return n +} + +func (q *UnlistedQuota) unlisted() bool { + if q.disabled { + return false + } + if q.force { + return true + } + st, _ := q.read() + return st.Unlisted +} + +func (q *UnlistedQuota) path() string { + dir := q.Dir + if dir == "" { + cache, err := os.UserCacheDir() + if err != nil { + cache = "." + } + dir = filepath.Join(cache, "slack-mcp-server") + } + team := q.TeamID + if team == "" { + team = "unknown" + } + method := strings.ReplaceAll(q.Method, ".", "_") + if method == "" { + method = "history" + } + return filepath.Join(dir, team+"_"+method+"_quota.json") +} + +func (q *UnlistedQuota) read() (quotaState, error) { + var st quotaState + b, err := os.ReadFile(q.path()) + if err != nil { + return st, err + } + if err := json.Unmarshal(b, &st); err != nil { + return quotaState{}, err + } + return st, nil +} + +func (q *UnlistedQuota) write(st quotaState) error { + path := q.path() + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + b, err := json.Marshal(st) + if err != nil { + return err + } + tmp, err := os.CreateTemp(filepath.Dir(path), "quota-") + if err != nil { + return err + } + tmpName := tmp.Name() + if _, err := tmp.Write(b); err != nil { + tmp.Close() + os.Remove(tmpName) + return err + } + if err := tmp.Close(); err != nil { + os.Remove(tmpName) + return err + } + return os.Rename(tmpName, path) +} + +func (q *UnlistedQuota) MarkUnlisted() { + if q.disabled { + return + } + st, _ := q.read() + st.Unlisted = true + _ = q.write(st) +} +func (q *UnlistedQuota) Touch() { + st, _ := q.read() + if q.force { + st.Unlisted = true + } + st.LastUnixNano = q.now().UnixNano() + _ = q.write(st) +} + +// Lock serializes history/replies calls across slack-cli processes so they +// share the 1/min budget instead of all 429ing. Stale locks older than 2 +// intervals are stolen. +func (q *UnlistedQuota) Lock(ctx context.Context) (func(), error) { + path := q.path() + ".lock" + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return func() {}, err + } + for { + f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) + if err == nil { + _, _ = fmt.Fprintf(f, "%d\n", q.now().Unix()) + _ = f.Close() + return func() { _ = os.Remove(path) }, nil + } + if info, statErr := os.Stat(path); statErr == nil { + age := q.now().Sub(info.ModTime()) + interval := q.Interval + if interval <= 0 { + interval = UnlistedInterval + } + if age > 2*interval { + _ = os.Remove(path) + continue + } + } + select { + case <-ctx.Done(): + return func() {}, ctx.Err() + case <-time.After(50 * time.Millisecond): + } + } +} +func (q *UnlistedQuota) Wait(ctx context.Context) error { + if !q.unlisted() { + return nil + } + st, _ := q.read() + if st.LastUnixNano == 0 { + return nil + } + interval := q.Interval + if interval <= 0 { + interval = UnlistedInterval + } + next := time.Unix(0, st.LastUnixNano).Add(interval) + now := q.now() + if !now.Before(next) { + return nil + } + return sleepCtx(ctx, next.Sub(now)) +} + +func sleepCtx(ctx context.Context, d time.Duration) error { + if d <= 0 { + return nil + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } +} + +func (q *UnlistedQuota) markThreshold() time.Duration { + if q.MarkAfter > 0 { + return q.MarkAfter + } + return 20 * time.Second +} + +// DoUnlisted runs fn under the special-tier lock: wait out the 1/min slot, +// retry 429s (honouring Retry-After), and remember a long Retry-After as +// unlisted so later processes cap the page size at 15. +func DoUnlisted[T any](ctx context.Context, q *UnlistedQuota, retryAfter func(error) time.Duration, fn func() (T, error)) (T, error) { + var zero T + unlock, err := q.Lock(ctx) + if err != nil { + return zero, err + } + defer unlock() + if err := q.Wait(ctx); err != nil { + return zero, err + } + const maxRetries = 3 + var last error + for attempt := 0; attempt <= maxRetries; attempt++ { + res, err := fn() + if err == nil { + q.Touch() + return res, nil + } + last = err + wait := time.Duration(0) + if retryAfter != nil { + wait = retryAfter(err) + } + if wait <= 0 { + return res, err + } + if wait >= q.markThreshold() { + q.MarkUnlisted() + } + if attempt == maxRetries { + return res, err + } + if err := sleepCtx(ctx, wait); err != nil { + return zero, err + } + } + return zero, last +} diff --git a/pkg/limiter/unlisted_test.go b/pkg/limiter/unlisted_test.go new file mode 100644 index 0000000..4ad9efc --- /dev/null +++ b/pkg/limiter/unlisted_test.go @@ -0,0 +1,154 @@ +package limiter + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestCapPage(t *testing.T) { + dir := t.TempDir() + q := &UnlistedQuota{Dir: dir, TeamID: "T1", Method: "conversations.history", PageSize: UnlistedPageSize, now: time.Now} + if q.CapPage(100) != 100 { + t.Fatalf("listed cap: got %d", q.CapPage(100)) + } + q.MarkUnlisted() + if got := q.CapPage(100); got != 15 { + t.Fatalf("unlisted cap 100: got %d", got) + } + if got := q.CapPage(0); got != 15 { + t.Fatalf("unlisted cap 0: got %d", got) + } + if got := q.CapPage(7); got != 7 { + t.Fatalf("unlisted cap 7: got %d", got) + } +} + +func TestCapPageDisabled(t *testing.T) { + q := &UnlistedQuota{Dir: t.TempDir(), disabled: true, PageSize: 15, now: time.Now} + q.MarkUnlisted() + if q.CapPage(100) != 100 { + t.Fatal("disabled quota should not cap") + } +} + +func TestCapPageForced(t *testing.T) { + q := &UnlistedQuota{Dir: t.TempDir(), force: true, PageSize: 15, now: time.Now} + if q.CapPage(100) != 15 { + t.Fatal("forced unlisted should cap before any 429") + } +} + +func TestWaitSkipsWhenFresh(t *testing.T) { + q := &UnlistedQuota{ + Dir: t.TempDir(), + TeamID: "T1", + Method: "history", + Interval: time.Hour, + now: time.Now, + force: true, + } + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + if err := q.Wait(ctx); err != nil { + t.Fatal(err) + } +} + +func TestWaitSleepsUntilInterval(t *testing.T) { + now := time.Now() + q := &UnlistedQuota{ + Dir: t.TempDir(), + TeamID: "T1", + Method: "history", + Interval: 80 * time.Millisecond, + now: func() time.Time { return now }, + force: true, + } + q.Touch() + q.now = time.Now + start := time.Now() + if err := q.Wait(context.Background()); err != nil { + t.Fatal(err) + } + if time.Since(start) < 50*time.Millisecond { + t.Fatalf("wait returned too quickly: %s", time.Since(start)) + } +} +func TestDoUnlistedRetryAndMark(t *testing.T) { + q := &UnlistedQuota{ + Dir: t.TempDir(), + TeamID: "T1", + Method: "conversations.history", + Interval: time.Millisecond, + PageSize: 15, + MarkAfter: 10 * time.Millisecond, + now: time.Now, + } + var n atomic.Int32 + res, err := DoUnlisted(context.Background(), q, func(error) time.Duration { return 25 * time.Millisecond }, func() (int, error) { + if n.Add(1) == 1 { + return 0, errors.New("rate limited") + } + return 7, nil + }) + if err != nil { + t.Fatal(err) + } + if res != 7 || n.Load() != 2 { + t.Fatalf("res=%d n=%d", res, n.Load()) + } + if !q.unlisted() { + t.Fatal("long Retry-After should mark unlisted") + } +} + +func TestDoUnlistedShortRetryDoesNotMark(t *testing.T) { + q := &UnlistedQuota{ + Dir: t.TempDir(), + TeamID: "T1", + Method: "conversations.replies", + Interval: time.Millisecond, + now: time.Now, + } + _, err := DoUnlisted(context.Background(), q, func(error) time.Duration { return 5 * time.Millisecond }, func() (int, error) { + return 1, nil + }) + if err != nil { + t.Fatal(err) + } + if q.unlisted() { + t.Fatal("success should not mark unlisted") + } +} + +func TestLockSerializes(t *testing.T) { + q := &UnlistedQuota{Dir: t.TempDir(), TeamID: "T1", Method: "history", Interval: time.Second, now: time.Now} + ctx := context.Background() + unlock, err := q.Lock(ctx) + if err != nil { + t.Fatal(err) + } + var gotLock atomic.Bool + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + u, err := q.Lock(ctx) + if err == nil { + gotLock.Store(true) + u() + } + }() + time.Sleep(40 * time.Millisecond) + if gotLock.Load() { + t.Fatal("second lock acquired while first held") + } + unlock() + wg.Wait() +} diff --git a/pkg/provider/api.go b/pkg/provider/api.go index a14cfa9..483c0ce 100644 --- a/pkg/provider/api.go +++ b/pkg/provider/api.go @@ -517,12 +517,50 @@ func (c *MCPSlackClient) GetConversationsForUserContext(ctx context.Context, par return c.slackClient.GetConversationsForUserContext(ctx, params) } +func (c *MCPSlackClient) teamID() string { + if c.authResponse != nil && c.authResponse.TeamID != "" { + return c.authResponse.TeamID + } + return "unknown" +} + +func slackRetryAfter(err error) time.Duration { + var rle *slack.RateLimitedError + if errors.As(err, &rle) { + return rle.RetryAfter + } + return 0 +} + func (c *MCPSlackClient) GetConversationHistoryContext(ctx context.Context, params *slack.GetConversationHistoryParameters) (*slack.GetConversationHistoryResponse, error) { - return c.slackClient.GetConversationHistoryContext(ctx, params) + if params == nil { + params = &slack.GetConversationHistoryParameters{} + } + q := limiter.NewUnlistedQuota(c.teamID(), "conversations.history") + return limiter.DoUnlisted(ctx, q, slackRetryAfter, func() (*slack.GetConversationHistoryResponse, error) { + p := *params + p.Limit = q.CapPage(p.Limit) + return c.slackClient.GetConversationHistoryContext(ctx, &p) + }) } func (c *MCPSlackClient) GetConversationRepliesContext(ctx context.Context, params *slack.GetConversationRepliesParameters) (msgs []slack.Message, hasMore bool, nextCursor string, err error) { - return c.slackClient.GetConversationRepliesContext(ctx, params) + if params == nil { + params = &slack.GetConversationRepliesParameters{} + } + q := limiter.NewUnlistedQuota(c.teamID(), "conversations.replies") + type replies struct { + msgs []slack.Message + hasMore bool + nextCursor string + } + res, err := limiter.DoUnlisted(ctx, q, slackRetryAfter, func() (replies, error) { + p := *params + p.Limit = q.CapPage(p.Limit) + m, more, cursor, err := c.slackClient.GetConversationRepliesContext(ctx, &p) + return replies{m, more, cursor}, err + }) + return res.msgs, res.hasMore, res.nextCursor, err } func (c *MCPSlackClient) SearchContext(ctx context.Context, query string, params slack.SearchParameters) (*slack.SearchMessages, *slack.SearchFiles, error) { diff --git a/skills/slack-cli/SKILL.md b/skills/slack-cli/SKILL.md index 9bc6043..176dda2 100644 --- a/skills/slack-cli/SKILL.md +++ b/skills/slack-cli/SKILL.md @@ -87,6 +87,9 @@ slack-cli channels me # channels you belong to slack-cli conversations history [--limit 1d|1w|30d|] [--cursor C] [--activity] slack-cli conversations replies # Pagination: read the Cursor field of the last element, then pass --limit='' --cursor . +# Unlisted Slack apps: 1 req/min and 15 msgs/page on history/replies. The CLI waits +# Retry-After and shares the slot across processes. Set SLACK_MCP_UNLISTED_HISTORY=1 +# to force that cap. Default --timeout is 2m. Do not hammer these in a loop. # Search (needs xoxp or browser token; not bot) slack-cli conversations search [query] \