CXP-533 fix event feed DeadlineExceeded from unbounded per-user Reports API fan-out - #131
CXP-533 fix event feed DeadlineExceeded from unbounded per-user Reports API fan-out#131JavierCarnelli-ConductorOne wants to merge 6 commits into
Conversation
…an-out usage_event_feed's OAuth lookup issued one sequential Reports API call per authorized app with no cap, so a batch of users with many authorized apps could force a single ListEvents call to make hundreds of rate-limited calls and blow the SDK's RPC deadline (observed in production). Add a per-call budget (maxLookupCallsPerEventFeedCall) shared across all three "last login" event feeds, with resumable per-user progress via cursor state, so a user needing more calls than the budget allows pauses mid-user instead of stalling or restarting from scratch. Also run a user's per-app lookups concurrently (bounded) so network latency overlaps instead of stacking on top of the rate-limiter wait. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Connector PR Review: CXP-533 fix event feed DeadlineExceeded from unbounded per-user Reports API fan-outBlocking Issues: 0 | Suggestions: 2 | Threads Resolved: 0 Review SummaryThe new commit ( Security IssuesNone found. Correctness IssuesNone found. Suggestions
Prompt for AI agents |
- Clamp resume-state index to 0 to prevent a panic on a negative/corrupted cursor value in usage_event_feed's per-app resume logic. - Stop advancing the cursor in scanUsersForEvents' error branch: the SDK discards the returned StreamState whenever err != nil, so the prior advancement had no effect and only muddied the comment. - Add a wall-clock soft deadline (maxEventFeedCallDuration) alongside the call-count budget, since a call budget alone doesn't bound elapsed time under a shared, contended rate limiter with retry/backoff. - Make the heavy multi-app resume test use an unlimited rate limiter so it doesn't compete for or depend on the shared 220/min quota other tests use. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Tokens.list is re-fetched fresh on every resumed call and its ordering isn't guaranteed stable, so a positional resume index could silently skip or re-visit apps whenever a user's authorized-app set changed between calls (e.g. an app revoked mid-pagination shifts every later app's index down by one). Resume by the last-processed app's client_id instead, re-locating it in the fresh list each time. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…budget maxEventFeedCallDuration was only checked between users in scanUsersForEvents, never during a single user's own lookup. Since usageEventFeed.lookupUser could fan out a full budget's worth of Reports calls (up to 60) in one go, each retrying up to reportsMaxRetries times against a rate limiter shared across all three event feeds, one heavy contended user could still blow well past the intended wall-clock budget before the check ever ran again — reproducing the DeadlineExceeded this branch exists to fix. Thread a deadline (derived from the soft budget, tightened against ctx's own deadline when set) down into each feed's lookup. usageEventFeed's fan-out now runs in fixed-size chunks and checks the deadline between chunks (never before the first, so a call always makes progress), stopping early and resuming from the last app actually finished once time is up. The two single-call feeds (google login, SAML) just accept and ignore the new parameter. Also reworded a misleading comment: the error-path cursor in scanUsersForEvents isn't "untouched" when a fresh directory page was already fetched earlier in the same call — it's simply moot, since the SDK discards the returned StreamState on error. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
- scanUsersForEvents now checks the ctx-derived deadline instead of the fixed maxEventFeedCallDuration, so a short RPC deadline is respected uniformly by the SAML and Google-login feeds (which never used the deadline internally), not just the usage feed's per-user fan-out. - usageEventFeed.lookupUser drops the fixed-size chunk barrier in favor of a single continuous errgroup, checking the deadline before dispatching each app rather than between chunks, so a slow/rate-limited app no longer idles the other concurrency slots for a whole chunk. - Guard against indexing toProcess[consumed-1] when consumed == 0. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
| if budget <= 0 || !time.Now().Before(deadline) { | ||
| // Budget or time spent before starting u; resume here fresh next call. | ||
| cursor.PendingUsers = cursor.PendingUsers[i:] | ||
| cursor.ResumeState = "" | ||
| return finish(events, true) |
There was a problem hiding this comment.
🟡 Suggestion (confidence: high): switching this guard to the ctx-derived deadline makes the i == 0 branch reachable for the first time, and that branch clears cursor.ResumeState for a user it is keeping at the front. Previously time.Since(start) >= maxEventFeedCallDuration was effectively always false at i == 0 (time.Since(start) ≈ 0), so the reset was harmless; now, if ctx.Deadline() is under eventFeedDeadlineSafetyMargin (5s) away when scanUsersForEvents starts, deadline is already in the past and the loop returns at i == 0 after wiping the head user's mid-user progress — so the next call restarts that user from app 0 and re-issues every lookup it already paid for. budget <= 0 can't fire at i == 0 (budget starts at 60), so this is purely the deadline path.
Also note that this exit makes zero forward progress: PendingUsers is unchanged and HasMore is true, so a persistently short RPC deadline would spin without advancing.
| if budget <= 0 || !time.Now().Before(deadline) { | |
| // Budget or time spent before starting u; resume here fresh next call. | |
| cursor.PendingUsers = cursor.PendingUsers[i:] | |
| cursor.ResumeState = "" | |
| return finish(events, true) | |
| if budget <= 0 || !time.Now().Before(deadline) { | |
| // Budget or time spent before starting u; resume here fresh next call. | |
| cursor.PendingUsers = cursor.PendingUsers[i:] | |
| if i > 0 { | |
| // u's own resumeState (if any) belonged to batch[0], which is already done. | |
| cursor.ResumeState = "" | |
| } | |
| return finish(events, true) | |
| } |
| for i, app := range toProcess { | ||
| if i > 0 && !deadline.IsZero() && time.Now().After(deadline) { | ||
| // Past budget: stop dispatching new lookups. Ones already in flight are awaited | ||
| // below; nextResume picks up right after the last app actually dispatched. | ||
| break | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: the dispatch loop only breaks on deadline, never on gctx, so a single failing app lookup no longer stops the fan-out. When app 3 of 60 errors, errgroup cancels gctx but g.Go keeps accepting the remaining 57 apps; each one calls sharedReportsRateLimiter.Wait, which returns nil (and decrements a token) whenever tokens >= 1 without checking ctx.Done() first, then fails immediately on the cancelled context. Net effect: up to 57 tokens burned out of the process-global 220/min bucket on requests that were never going to be issued — throttling the next legitimate call. The chunked version stopped at the failing chunk boundary. Adding gctx.Err() != nil to the break condition restores that.
| for i, app := range toProcess { | |
| if i > 0 && !deadline.IsZero() && time.Now().After(deadline) { | |
| // Past budget: stop dispatching new lookups. Ones already in flight are awaited | |
| // below; nextResume picks up right after the last app actually dispatched. | |
| break | |
| } | |
| for i, app := range toProcess { | |
| if gctx.Err() != nil { | |
| // A prior lookup already failed and cancelled gctx; don't dispatch doomed work | |
| // (each would still spend a shared rate-limiter token before failing). | |
| break | |
| } | |
| if i > 0 && !deadline.IsZero() && time.Now().After(deadline) { | |
| // Past budget: stop dispatching new lookups. Ones already in flight are awaited | |
| // below; nextResume picks up right after the last app actually dispatched. | |
| break | |
| } |
| // maxConcurrentAppLookups bounds concurrent per-app Reports lookups for one user, and also sizes | ||
| // the fan-out chunks in lookupUser (see below) between which the deadline budget is checked. The | ||
| // shared reportsRateLimiter still caps overall quota use; this just overlaps network latency. | ||
| const maxConcurrentAppLookups = 8 |
There was a problem hiding this comment.
🟡 Suggestion: this doc comment is now stale — the chunked fan-out was replaced by a single errgroup with SetLimit, so maxConcurrentAppLookups no longer "sizes the fan-out chunks ... between which the deadline budget is checked"; the deadline is checked before each dispatch. Same staleness in event_feed_common_test.go: the test is still named TestUsageEventFeed_LookupUserStopsAtDeadlineBetweenChunks and numApps's comment still reads // more than one chunk's worth.
| // maxConcurrentAppLookups bounds concurrent per-app Reports lookups for one user, and also sizes | |
| // the fan-out chunks in lookupUser (see below) between which the deadline budget is checked. The | |
| // shared reportsRateLimiter still caps overall quota use; this just overlaps network latency. | |
| const maxConcurrentAppLookups = 8 | |
| // maxConcurrentAppLookups bounds how many per-app Reports lookups for one user are in flight at | |
| // once in lookupUser (see below). The shared reportsRateLimiter still caps overall quota use; | |
| // this just overlaps network latency. |
| deadline := start.Add(maxEventFeedCallDuration) | ||
| if ctxDeadline, ok := ctx.Deadline(); ok { | ||
| if safe := ctxDeadline.Add(-eventFeedDeadlineSafetyMargin); safe.Before(deadline) { | ||
| deadline = safe | ||
| } | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion (test gap): the ctx-derived deadline is the behavior this commit exists to activate, and nothing exercises it — no test in pkg/connector constructs a context.WithTimeout/WithDeadline for scanUsersForEvents, so neither the eventFeedDeadlineSafetyMargin subtraction, the safe.Before(deadline) min-selection, nor the new !time.Now().Before(deadline) between-users guard is covered. A table test with ctx deadlines of (say) 3s and 120s asserting which of maxEventFeedCallDuration vs ctxDeadline - margin wins, plus one asserting the between-users loop stops early under a near-expired ctx, would lock this in cheaply and would have surfaced the ResumeState reset flagged below.
| startIdx := 0 | ||
| if resumeState != "" { | ||
| for i, app := range apps { | ||
| if app.ClientID == resumeState { | ||
| startIdx = i + 1 | ||
| break | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: the doc comment justifies client_id resume because "ordering is not guaranteed stable across calls", then claims it "never skips one" — but that only holds if relative order is preserved. Under a genuine reorder, any not-yet-processed app that now sorts before the anchor falls into apps[:startIdx] and is skipped for this pass (its usage event is silently missed until the next full directory walk). Only the anchor-removed case is handled by the restart-from-0 fallback. Consider tracking processed client_ids in the resume state (or a processed-count/set) rather than a single positional anchor, or at least soften the "never skips" claim to "never skips when ordering is preserved."
| if err := g.Wait(); err != nil { | ||
| return nil, "", 0, err | ||
| } |
There was a problem hiding this comment.
🟡 Suggestion: a single failing app discards the whole user — consumed: 0 and no events — so up to budget-1 already-completed Reports lookups (and their quota) are thrown away, and because the SDK drops the StreamState on error the next call replays this user from the same anchor. If one app returns a non-retryable error persistently (e.g. a 403 scoped to that app), the cursor can never advance past this user and the feed stalls indefinitely. Per R7/F3, consider degrading per app: log a Warn for the failing client_id, keep the successful results, and still return consumed/nextResume so the walk makes forward progress.
1 participant
usage_event_feed's OAuth lookup issued one sequential Reports API call per
authorized app with no cap, so a batch of users with many authorized apps
could force a single ListEvents call to make hundreds of rate-limited calls
and blow the SDK's RPC deadline (observed in production).
Add a per-call budget (maxLookupCallsPerEventFeedCall) shared across all
three "last login" event feeds, with resumable per-user progress via cursor
state, so a user needing more calls than the budget allows pauses mid-user
instead of stalling or restarting from scratch. Also run a user's per-app
lookups concurrently (bounded) so network latency overlaps instead of
stacking on top of the rate-limiter wait.
Co-Authored-By: Claude Sonnet 5 noreply@anthropic.com