[pull] master from DataDog:master#653
Merged
Merged
Conversation
* Add async GitHub client rate limiting for the Dispatcher (AI-6475) - Add InstrumentedAsyncLimiter wrapper around aiolimiter.AsyncLimiter that fires on_throttled/on_acquired callbacks for observability without touching aiolimiter internals - Add RateLimiterFactory (dispatcher-specific) that holds two shared limiter instances (default 360 req/hr, slow 120 req/hr) so all processors in a run compete for the same token buckets; validates default+slow <= total_max_rate at construction time - Wire rate_limiter into AsyncGitHubClient._request() and async_github_client() context manager - Add aiolimiter==1.2.1 dependency * Add changelog entry for #24179 * Update changelog entry for #24179 * Move RateLimiterConfig to dispatcher module and expose time_period * Convert rate limiter config objects to Pydantic models * Rename max_rate to hourly_max_rate and derive AsyncLimiter bucket size from it * Revert hourly_max_rate rename; normalize to req/hr only in factory validation * Address Codex review: keyword-only limiter arg and rate-limit artifact redirects - Make rate_limiter, default_timeout, and transport keyword-only on AsyncGitHubClient and the async_github_client factory. - Route the authenticated artifact-redirect GET through the rate-limited _request path so it counts against the shared limiter budget. * Add header-driven budget governor with a configurable pacing cap - BudgetGovernor reacts to GitHub's x-ratelimit/retry-after response headers, rationing requests as the shared budget runs low and hard-pausing on exhaustion or secondary limits; one governor is shared across both tiers. - max_wait_seconds caps the voluntary pacing interval (factory default 300s) so a request never paces itself out toward the full reset window; hard pauses are still honored in full. - The async client feeds each response's rate-limit headers to the governor. - Test cleanup: shared assert_blocks helper, clock/governor fixtures and a snapshot factory, parametrized cases, and added branch/coverage tests. * Address Codex review: commutative budget merges, drop underscore attributes - observe() now keeps the lowest remaining within a window and the longest retry_after pause, so out-of-order concurrent responses converge to the strictest constraint regardless of arrival order. - Drop leading underscores from InstrumentedAsyncLimiter and RateLimiterFactory instance attributes per the AGENTS.md naming rule. * Fold the monotonic-remaining rule into BudgetSnapshot.merged_with The within-window invariant (remaining is non-increasing until reset_at advances) is intrinsic to a rate-limit budget, so combining two observations enforces it directly rather than in a separate reconcile step. observe() drops back to a plain merged_with call and the monotonic tests move to that level. * Unify rate-limit observability into a single on_event callback Replace the four separate callbacks (on_budget_low, on_secondary_limit, on_throttled, on_acquired) with one on_event that receives a discriminated RateLimitEvent union (BUCKET, BUDGET, SECONDARY_LIMIT, PACING). Events fire on every acquire/observe/wait carrying a boolean or reason, so a single handler can emit continuous 0/1 metrics instead of sparse per-state callbacks. The factory shares one handler across the governor and both tier limiters. * Collapse reserve()/reserve_with_reason() into a single reserve() The bare reserve() had no production callers left (wait() is the only path that reserves, and it needs the reason), so merge the two into one reserve() that returns (target, reason). Callers take what they need. * Harden BudgetGovernor and InstrumentedAsyncLimiter after review - merged_with discards a stale, out-of-order response from an earlier window (smaller reset_at) instead of rolling budget state back, still coalescing retry_after. - Acquire the local bucket token AFTER the governor wait, not before, so the token is taken immediately before the request fires rather than being spent across a long pause. - wait() raises instead of silently proceeding if it fails to converge within MAX_WAIT_ITERATIONS, and emits a PacingEvent when a pause extends it mid-sleep. - Rename budget_target_with_reason -> claim_budget_target_with_reason to signal the pacing-cursor mutation; sample now() once in observe's retry_after branch. - Test suite: drop redundant/trivial tests, parametrize the merged_with overlay and wait() pacing-reason cases, assert behavior over internals, and add stale-window, concurrency, mid-sleep-extension, fail-loud, and acquire-order coverage. * Track the pause floor for extension events; drop a misleading concurrency test - wait() emits the mid-sleep SECONDARY_LIMIT PacingEvent only when the pause floor actually grows, instead of on every post-first iteration. This avoids a spurious event from monotonic-vs-wall-clock skew (asyncio.sleep uses the loop's monotonic clock while now() is wall-clock) and keeps the fail-loud path quiet. - Remove test_concurrent_waits_stagger_by_pacing_interval: the shared-advancing fake clock cannot model overlapping sleeps, so the test only passed because it ran sequentially; making it genuinely interleave collapses all finish times. The cursor invariant is structural (no await between reading and advancing next_slot) and covered by the sequential pacing test. - Nits: rename test to match its contract, use get_running_loop/create_task, and cancel the in-flight task in a finally. * Replace max_wait_seconds cap with a window-reset clamp max_wait_seconds was an arbitrary bound. The principled bound is the window itself: a rationed request never needs to wait past reset_at, because the budget refills then. claim_budget_target_with_reason now clamps at the boundary — once the pacing cursor reaches reset_at the window is spent, so overflow requests wait exactly until reset+buffer (EXHAUSTED) without advancing the cursor, so they all get the same target. Removes the config field, the governor parameter, and their tests; drops the pointless conditional advance in test_reserve_returns_now. * Forbid unknown keys in the rate limiter config models * Shorten the rate limiting changelog entry * Drop retry_after from the persisted budget and tighten rate-limit tests - BudgetSnapshot.merged_with no longer carries retry_after; it is a transient secondary-limit signal consumed only in observe, which makes the budget merge fully commutative. - Normalize snapshot construction in tests onto the make_snapshot factory and give it explicit optional fields instead of opaque **fields. - Rewrite the convergence test to assert an explicit expected state and name the failing permutation. - Docstring cleanup: state final behavior only, no development history. * Enforce the synchronous pacing-claim invariant with tests Add two guards for the "no await between reading and advancing next_slot" rule that reserve()'s inline comment relies on: - test_pacing_claim_path_is_synchronous asserts the claim path stays synchronous and, on failure, tells the editor what mechanism they owe in exchange. - test_reserve_assigns_distinct_slots_under_concurrent_claims gathers concurrent claims on the real event loop and asserts distinct, interval-spaced slots. Point the inline comment at both tests so it reads as enforced, not advisory. * Replace the wait iteration cap's flood role with a time budget The iteration count bounded nothing meaningful in time: one iteration costs anywhere from seconds to a retry_after of many minutes. Add an opt-in max_wait_seconds killswitch on BudgetGovernor that bounds the total time a single wait() may target, from when it starts, across both exhausted-window waits and accumulated secondary-limit extensions. - Raise RateLimitWaitAbandoned (subclasses TimeoutError) carrying waited_seconds and remaining_seconds; fail fast the moment the floor crosses the budget rather than sleeping into a doomed wait. - Emit a PacingEvent with the new PacingReason.ABANDONED before raising. - Default None keeps the current wait-indefinitely behavior. - Keep MAX_WAIT_ITERATIONS as the RuntimeError backstop for the no-progress bug (frozen clock) that a time budget cannot detect. - Fold in the single-current_floor cleanup in the wait loop. * Make GitHub client rate-limit protection the default and add rate-limit-aware retries Protection is now always on: constructing the async client without a rate_limiter builds a default one (a permissive bucket fronting a reactive BudgetGovernor) instead of running unprotected. Because octo-sts mints the token against one installation for the whole company, an unprotected client would degrade every other consumer. - Add github_async/defaults.py with default_github_rate_limiter and log_rate_limit_events (GitHub-shaped defaults; the rate_limiting module stays provider-agnostic). - Observe response headers before raise_for_status so a failed response's rate-limit signal is learned even when the caller swallows the exception. - Retry header-confirmed rate-limit responses (403/429) in _request; re-acquiring the limiter is the backoff, so there is no sleep or backoff math in the client. Transport errors and non-rate-limit statuses are never retried; RateLimitWaitAbandoned propagates. Bounded by max_rate_limit_retries (default 2). - Lower MAX_WAIT_ITERATIONS to 100 now that the time budget guards real floods.
* Add DispatcherConfig model read from .ddev/config.toml Holds max_jobs_per_batch, global_timeout_seconds, and the GitHub rate limits (RateLimiterFactoryConfig). Loaded via the /dispatcher JSON pointer of the repo config. * Add changelog entry for #24434 * Type-annotate test fixture and functions * Document the batch and timeout defaults * Drop -> None return annotations from tests * Make the changelog entry a patch-level fixed entry * Cap max_jobs_per_batch at the safe workflow maximum
…memory budget (#24508) * kafka_consumer: cap broker_timestamps retained history by a per-cluster budget The DSM broker_timestamps cache grows unbounded with partition count (up to timestamp_history_size samples for every partition), reaching multiple GiB on large clusters. Bound it to a fixed entry budget (~0.5 GiB/cluster): the per-partition history depth is scaled down as partition count grows (budget // num_partitions, clamped to [2, timestamp_history_size]). Small clusters keep full history; large clusters shrink to a shallow depth, so total memory stays bounded. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Add changelog entry Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
See Commits and Changes for more details.
Created by
pull[bot] (v2.0.0-alpha.4)
Can you help keep this open source service alive? 💖 Please sponsor : )