Queue for a resilience slot instead of failing parallel invocations

internal/commands/files.go

@@ -1504,8 +1504,7 @@ You can pass either an item ID or a Basecamp URL:
 
 				// If all probes failed, check if first error was 404 or something else
 				if result == nil && firstErr != nil {
-					sdkErr := basecamp.AsError(firstErr)
-					if sdkErr.Code != basecamp.CodeNotFound {
+					if sdkErr := basecamp.AsError(firstErr); sdkErr == nil || sdkErr.Code != basecamp.CodeNotFound {
 						// Return actual error (auth, permission, network, etc.)
 						return convertSDKError(firstErr)
 					}
@@ -2155,8 +2154,7 @@ You can pass either an item ID or a Basecamp URL:
 							detectedType = "upload"
 						} else {
 							// All probes failed - check if first error was 404 or something else
-							sdkErr := basecamp.AsError(firstErr)
-							if sdkErr.Code != basecamp.CodeNotFound {
+							if sdkErr := basecamp.AsError(firstErr); sdkErr == nil || sdkErr.Code != basecamp.CodeNotFound {
 								// Return actual error (auth, permission, network, etc.)
 								return convertSDKError(firstErr)
 							}

internal/commands/projects.go

@@ -433,6 +433,11 @@ func convertSDKError(err error) error {
 		return nil
 	}
 
+	// A gate that queued and gave up says which limit, how long, and what to do
+	if gateErr := output.AsGateError(err); gateErr != nil {
+		return gateErr
+	}
+
 	// Handle resilience sentinel errors (use errors.Is for wrapped errors)
 	if errors.Is(err, basecamp.ErrRateLimited) {
 		return &output.Error{

internal/commands/projects_test.go

@@ -2,13 +2,16 @@ package commands
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -20,6 +23,7 @@ import (
 	"github.com/basecamp/basecamp-cli/internal/config"
 	"github.com/basecamp/basecamp-cli/internal/names"
 	"github.com/basecamp/basecamp-cli/internal/output"
+	"github.com/basecamp/basecamp-cli/internal/resilience"
 )
 
 type mockProjectUpdateTransport struct {
@@ -190,3 +194,26 @@ func TestProjectsCreateErrorEnvelopeCarriesRetryable(t *testing.T) {
 		})
 	}
 }
+
+// A gate that queued and gave up reaches the user with its own message and
+// hint (which limit, how long it waited, what to do) rather than the generic
+// sentinel wording, while keeping the rate-limit code and retryable flag.
+func TestConvertSDKErrorCarriesTheGateMessageAndHint(t *testing.T) {
+	store := resilience.NewStore(t.TempDir())
+	require.NoError(t, store.Update(func(state *resilience.State) error {
+		state.Bulkhead.ActivePIDs = []int{os.Getppid()}
+		return nil
+	}))
+	bh := resilience.NewBulkhead(store, resilience.BulkheadConfig{MaxConcurrent: 1})
+	gateErr := bh.Wait(context.Background(), time.Now())
+	require.Error(t, gateErr)
+
+	err := convertSDKError(fmt.Errorf("listing projects: %w", gateErr))
+
+	var outErr *output.Error
+	require.ErrorAs(t, err, &outErr)
+	assert.Equal(t, basecamp.CodeRateLimit, outErr.Code)
+	assert.Equal(t, "Too many concurrent basecamp processes (limit 1); waited 0s", outErr.Message)
+	assert.Equal(t, "Re-run, or lower parallelism.", outErr.Hint)
+	assert.True(t, outErr.Retryable)
+}

internal/names/resolver.go

@@ -689,6 +689,11 @@ func convertSDKError(err error) error {
 		return nil
 	}
 
+	// A gate that queued and gave up says which limit, how long, and what to do
+	if gateErr := output.AsGateError(err); gateErr != nil {
+		return gateErr
+	}
+
 	// Handle resilience sentinel errors (use errors.Is for wrapped errors)
 	if errors.Is(err, basecamp.ErrRateLimited) {
 		return &output.Error{

internal/output/errors.go

@@ -7,6 +7,8 @@ import (
 
 	"github.com/basecamp/basecamp-sdk/go/pkg/basecamp"
 	clioutput "github.com/basecamp/cli/output"
+
+	"github.com/basecamp/basecamp-cli/internal/resilience"
 )
 
 // Error is a structured error with code, message, and optional hint.
@@ -32,6 +34,9 @@ func ErrAmbiguous(resource string, matches []string) *Error {
 }
 
 func AsError(err error) *Error {
+	if gateErr := AsGateError(err); gateErr != nil {
+		return gateErr
+	}
 	var sdkErr *basecamp.Error
 	if errors.As(err, &sdkErr) {
 		message := err.Error()
@@ -53,6 +58,23 @@ func AsError(err error) *Error {
 	return clioutput.AsError(err)
 }
 
+// AsGateError converts a resilience gate rejection, which arrives through
+// any SDK operation, into the rate-limit error the user sees: the gate's own
+// message and hint (which limit, how long it waited, what to do), retryable.
+// Nil when err is not a gate rejection.
+func AsGateError(err error) *Error {
+	var gateErr *resilience.GateError
+	if !errors.As(err, &gateErr) {
+		return nil
+	}
+	return &Error{
+		Code:      CodeRateLimit,
+		Message:   gateErr.Message,
+		Hint:      gateErr.Hint,
+		Retryable: true,
+	}
+}
+
 // RequestID returns the SDK request ID carried by err, if present.
 func RequestID(err error) string {
 	var sdkErr *basecamp.Error

internal/output/gate_error_test.go

@@ -0,0 +1,34 @@
+package output
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/basecamp/basecamp-cli/internal/resilience"
+)
+
+// A gate rejection reaches the user with its own message and hint through
+// the central conversion too, so commands that return SDK errors directly
+// render the same thing as those that call convertSDKError.
+func TestAsErrorCarriesTheGateMessageAndHint(t *testing.T) {
+	store := resilience.NewStore(t.TempDir())
+	require.NoError(t, store.Update(func(state *resilience.State) error {
+		state.Bulkhead.ActivePIDs = []int{os.Getppid()}
+		return nil
+	}))
+	gateErr := resilience.NewBulkhead(store, resilience.BulkheadConfig{MaxConcurrent: 1}).Wait(context.Background(), time.Now())
+	require.Error(t, gateErr)
+
+	err := AsError(fmt.Errorf("listing chat lines: %w", gateErr))
+
+	assert.Equal(t, CodeRateLimit, err.Code)
+	assert.Equal(t, "Too many concurrent basecamp processes (limit 1); waited 0s", err.Message)
+	assert.Equal(t, "Re-run, or lower parallelism.", err.Hint)
+	assert.True(t, err.Retryable)
+}

internal/resilience/bulkhead.go

@@ -1,8 +1,12 @@
 package resilience
 
 import (
+	"context"
+	"fmt"
 	"os"
 	"time"
+
+	"github.com/basecamp/basecamp-sdk/go/pkg/basecamp"
 )
 
 // Bulkhead implements the bulkhead pattern with cross-process persistence.
@@ -82,6 +86,44 @@ func (b *Bulkhead) Acquire() (bool, error) {
 	return acquired, nil
 }
 
+// slotPoll is the base interval between slot attempts while queued. A slot
+// frees when another process finishes its operation, which is network-bound,
+// so tens of milliseconds is fine-grained enough without hammering the lock.
+const slotPoll = 25 * time.Millisecond
+
+// Wait acquires a slot, polling with jitter until deadline. It returns nil
+// once the slot is held (Acquire fails open on a store error), a *GateError
+// once the deadline has passed, or ctx.Err(). Cancellation and the deadline
+// are checked before every attempt, so an expired or canceled gate reserves
+// nothing, and cancellation outranks the deadline.
+func (b *Bulkhead) Wait(ctx context.Context, deadline time.Time) error {
+	return b.waitSince(ctx, b.now(), deadline)
+}
+
+// waitSince is Wait for a gate that started queueing at start, so the wait
+// it reports covers the whole gate and not only the slot phase.
+func (b *Bulkhead) waitSince(ctx context.Context, start, deadline time.Time) error {
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		remaining := deadline.Sub(b.now())
+		if remaining <= 0 {
+			return &GateError{
+				Message:  fmt.Sprintf("Too many concurrent basecamp processes (limit %d); waited %s", b.config.MaxConcurrent, b.now().Sub(start).Round(time.Second)),
+				Hint:     "Re-run, or lower parallelism.",
+				sentinel: basecamp.ErrBulkheadFull,
+			}
+		}
+		if acquired, _ := b.Acquire(); acquired { //nolint:contextcheck // lock acquisition is context-independent by design
+			return nil
+		}
+		if err := pause(ctx, min(jittered(slotPoll), remaining)); err != nil {
+			return err
+		}
+	}
+}
+
 // Release releases the slot held by this process.
 func (b *Bulkhead) Release() error {
 	return b.store.Update(func(state *State) error {

internal/resilience/circuit_breaker.go

@@ -40,6 +40,15 @@ func (cb *CircuitBreaker) now() time.Time {
 	return cb.nowFn()
 }
 
+// Tripped reports whether the circuit is open with its timeout still running:
+// the one state Allow rejects without writing anything. Read here so a gate
+// can fail fast before queueing for a token it would only be refused with.
+func (cb *CircuitBreaker) Tripped() bool {
+	state, err := cb.store.Load()
+	return err == nil && state.CircuitBreaker.IsOpen() &&
+		cb.now().Sub(state.CircuitBreaker.OpenedAt) < cb.config.OpenTimeout
+}
+
 // Allow checks if a request should be allowed.
 // Returns true if the request can proceed, false if it should be rejected.
 // In half-open state, atomically reserves an attempt slot to prevent thundering herd.

internal/resilience/config.go

@@ -6,6 +6,13 @@ import (
 
 // Config holds configuration for all resilience primitives.
 type Config struct {
+	// MaxWait bounds how long a gated operation queues for a rate-limiter
+	// token or a bulkhead slot before it is rejected. Parallel invocations
+	// of the CLI share one bucket and one slot table, so a burst that would
+	// otherwise fail fast waits its turn instead.
+	// Default: 10 seconds
+	MaxWait time.Duration
+
 	// CircuitBreaker configures the circuit breaker pattern.
 	CircuitBreaker CircuitBreakerConfig
 
@@ -67,6 +74,7 @@ type BulkheadConfig struct {
 // DefaultConfig returns a Config with sensible defaults for the Basecamp API.
 func DefaultConfig() *Config {
 	return &Config{
+		MaxWait: DefaultMaxWait,
 		CircuitBreaker: CircuitBreakerConfig{
 			FailureThreshold:    5,
 			SuccessThreshold:    2,

internal/resilience/gate.go

@@ -0,0 +1,60 @@
+package resilience
+
+import (
+	"context"
+	"math/rand/v2"
+	"time"
+)
+
+// DefaultMaxWait is how long OnOperationGate queues for a token or a slot
+// before giving up. It is long enough to absorb a burst of parallel CLI
+// invocations (10 workers draining a 50-token bucket refill at 10/s in well
+// under a second) and short enough that a genuinely saturated machine still
+// fails within one attention span.
+const DefaultMaxWait = 10 * time.Second
+
+// GateError is a gate rejection that already waited its turn. It unwraps to
+// the SDK sentinel (basecamp.ErrRateLimited or basecamp.ErrBulkheadFull) so
+// existing errors.Is checks keep working, and carries the message and hint
+// the CLI shows: which limit, how long it waited, and what to do.
+type GateError struct {
+	Message  string
+	Hint     string
+	sentinel error
+}
+
+func (e *GateError) Error() string { return e.Message }
+
+// Unwrap returns the SDK sentinel the rejection stands for.
+func (e *GateError) Unwrap() error { return e.sentinel }
+
+// pause sleeps for d or until ctx is done, whichever comes first.
+func pause(ctx context.Context, d time.Duration) error {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+// jittered stretches d by up to half again so that processes which woke
+// together do not retry in lockstep and starve each other at the lock.
+func jittered(d time.Duration) time.Duration {
+	return d + jitter(d)
+}
+
+// jitter is the random stretch jittered adds; a variable so tests can pin it.
+var jitter = func(d time.Duration) time.Duration {
+	return rand.N(d/2 + 1) //nolint:gosec // jitter spreads retries; it guards nothing
+}
+
+// ceilSeconds rounds d up to a whole second.
+func ceilSeconds(d time.Duration) time.Duration {
+	if d%time.Second == 0 {
+		return d
+	}
+	return d.Truncate(time.Second) + time.Second
+}