refactor: unify push provider auth + outcome handling; bound retry policy#233
Merged
Conversation
…e retry engine - Clamp provider-supplied Retry-After at a new MAX_RETRY_AFTER (60s) ceiling so an untrusted header can no longer pin a send task (holding a decrypted device token and an in-flight guard) for an arbitrary duration. The floor and jitter from PR #204 are preserved. Fixes the policy gap in #162. - Add SendAttemptResult::AuthRejected and a one-shot immediate retry in with_retry: after a provider auth rejection (cached credential already invalidated by the client), the in-hand notification is retried exactly once with a freshly minted credential instead of being dropped (#85). The auth-retry budget is separate from the transient 429/5xx budget and counts toward push_retries_total. - Redact transport-error URLs inside with_transport_retry: reqwest errors can embed the request URL and the APNs URL contains the raw device token, so every arm now strips it via without_url()/redact_transport_url() before logging or propagation, independent of caller discipline (#172). Adds Error::redact_transport_url. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…nery New src/push/auth.rs owns the cached-credential semantics both push clients previously duplicated (and had already let drift, #158): - AuthTokenGenerator: provider-specific mint step (ES256 sign for APNs, OAuth round-trip for FCM), with mint failures classified by the shared TokenAcquisitionError (permanent vs transient) lifted from the FCM client. - TokenCache<G>: read-lock fast path plus a single-flight refresh that runs OUTSIDE the cache write lock — readers keep serving a still-valid token during a refresh and concurrent misses coalesce onto one mint via a dedicated refresh mutex and double-check (#86). - invalidate_if_matches: evicts a provider-rejected credential only while it is still the cached one; whether a rejection is credential-related is the caller's decision (APNs 403 reason gating per #145 lands with the client delegation). Also generalizes the bounded JSON body reader (parse_bounded_json_body) so the FCM OAuth success body can reuse it (#154 groundwork), buffering through a zeroizing Vec since that body carries a bearer credential. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The APNs client no longer carries its own cached-JWT machinery; an ApnsTokenGenerator (ES256 signing) plugs into the shared TokenCache<G> (#158). Behavioral changes: - 403 handling now parses the reason BEFORE deciding to evict. Only ExpiredProviderToken/InvalidProviderToken evict the cached JWT and return the new SendAttemptResult::AuthRejected (fresh-token retry via the retry engine, #85); every other 403 (BadCertificateEnvironment, Forbidden, ...) leaves the cache intact and is a permanent error, ending the per-notification re-sign stampede (#145). - The request template (ApnsRequestParts + zeroizing device-token URL) is built once in send() and borrowed by the retry closure; transport retries reuse it instead of rebuilding (#198). reqwest still owns the serialized non-zeroized body/headers — the documented accepted #126 posture. - Per-logical-push duration is recorded once in send() after retries resolve; handle_response records only the per-attempt response-status counter (#168). Metric semantic change: push_request_duration_seconds now counts notifications, not HTTP attempts. - Transport errors are stripped of their URL (which embeds the device token) before propagation/logging (#172). - Removed the unreachable is_valid_device_token validator; a debug_assert documents TokenPayload as the single source of token well-formedness (#199). Tests updated for the intentional semantic changes (403 reason gating, AuthRejected, per-push duration) and to drive the cache via test-only seed/inspect helpers plus a test base-URL override. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The FCM client's bespoke OAuth cache is replaced by an FcmTokenGenerator plugged into the shared TokenCache<G> (#158). The generator lifts the former refresh_token_inner wholesale, keeping the TokenAcquisitionError retriable/permanent classification (PR #209) and expires_in TTL (PR #212). Additional fixes folded in: - POST the OAuth request to the service-account token_uri so the signed assertion aud and the request endpoint always agree, falling back to the constant only when token_uri is empty (#153). - Read the OAuth success body through the shared bounded, zeroizing reader (MAX_OAUTH_BODY_BYTES) instead of unbounded response.json() (#154). - Single-flight refresh outside the cache write lock via TokenCache, so concurrent senders keep serving the valid token during a refresh (#86). - 401 now evicts and returns AuthRejected for a one-shot fresh-token retry (#85) instead of dropping the notification. - The request body template (FcmRequest borrowing the device token) is built once in send_once, outside the transport-retry closure, so a transport retry no longer rebuilds the map/struct or re-copies the token (#198). reqwest still owns the serialized non-zeroized body — documented accepted #126 posture. - Per-logical-push duration recorded once in send(); handle_response records only the per-attempt status counter (#168, same semantic change as APNs). - Transport errors are URL-stripped before propagation/logging (#172). Adds FCM tests for token_uri targeting, empty-token_uri fallback, expires_in TTL, and the bounded OAuth-body read. Existing 401 and duration tests updated for the intentional semantic changes; cache tests now drive the shared cache via seed/inspect helpers. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Dispatcher (#160): a second semaphore (MAX_LIVE_SEND_TASKS = 2x MAX_CONCURRENT_PUSHES) is acquired at task spawn in the recv loop and held for the task's whole life — NOT released during backoff. Since a retrying task releases its concurrency permit during backoff sleeps but stays alive holding a decrypted token, the old design let a provider 429/5xx storm drain the entire 10k queue into thousands of live token-holding tasks. The live-task semaphore caps live tasks (active + sleeping-in-backoff) at a small constant, applying backpressure to the queue instead. InFlightTracker drain semantics are unchanged. The recv-loop clients/semaphores are grouped into a DispatchWorkerContext. Secret hygiene (#172): both dispatcher send-error debug! sites now log through Error::redact_transport_url, defense-in-depth on top of the send-path stripping. Adds error.rs tests asserting a mapped reqwest error's Display carries no device token, plus the no-op case. Adapted send_push_records_invalid_token_metrics: the APNs malformed-token branch is gone (#199 — the dispatcher only feeds even-length hex), so the test now covers only the FCM short-circuit. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…nification # Conflicts: # src/push/dispatcher.rs
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 45 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. WalkthroughThis PR extracts a shared ChangesPush provider auth/retry/concurrency refactor
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
|
Ready to review this PR? Stage has broken it down into 6 individual chapters for you: 
Chapters generated by Stage for commit 5e12f8f on Jul 5, 2026 1:29pm UTC. |
✅ Code Coverage Report
View detailed coverageDownload the Or run locally: cargo llvm-cov --html
open target/llvm-cov/html/index.html |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/push/apns.rs (1)
341-357: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winDocument the new
sendtoken-format precondition.
sendis public and now relies on upstream token normalization in release builds. Please add this precondition to the method docs so external callers do not expect malformed tokens to be rejected locally. As per coding guidelines,**/*.rs: Update documentation for public API changes.Suggested documentation update
/// When `backoff_permit` is `Some`, the dispatcher concurrency permit is /// released during backoff sleeps and re-acquired before each retry, so a /// sleeping retry does not occupy an in-flight concurrency slot. + /// + /// # Preconditions + /// + /// `device_token` must be an even-length hex APNs device token produced by + /// the dispatcher token pipeline (`TokenPayload::device_token_hex()`). This + /// method does not perform release-mode token-format validation. pub async fn send(🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/push/apns.rs` around lines 341 - 357, The public `send` method in `apns.rs` now assumes tokens are already normalized upstream, so its docs need to state that precondition explicitly. Update the documentation for `Push::send` to note that `device_token` must be even-length lowercase hex produced by `TokenPayload::device_token_hex()` and that malformed tokens are not rejected locally in release builds. Mention the existing `debug_assert!` behavior so callers understand the contract change.Source: Coding guidelines
src/push/fcm.rs (1)
322-346: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winUse the resolved OAuth URL for the JWT audience too.
Line 325 still signs
audwith rawsa.token_uri, while Line 345 posts tooauth_token_url(sa). Whentoken_uriis empty, the POST falls back toOAUTH_TOKEN_URLbut the JWT audience stays empty, so fallback-configured service accounts will mint invalid assertions.🐛 Proposed fix
+ let token_url = self.oauth_token_url(sa); + let claims = OAuthClaims { iss: sa.client_email.clone(), scope: FCM_SCOPE.to_string(), - aud: sa.token_uri.clone(), + aud: token_url.to_string(), iat, exp, }; @@ .http_client - .post(self.oauth_token_url(sa)) + .post(token_url)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/push/fcm.rs` around lines 322 - 346, The JWT audience in the FCM OAuth flow is still built from sa.token_uri instead of the resolved endpoint used by oauth_token_url(sa), which breaks fallback configurations when token_uri is empty. Update the token-creation path in the FCM code so OAuthClaims.aud is derived from the same resolved OAuth URL that the request posts to, keeping the JWT audience and POST target consistent.
🧹 Nitpick comments (2)
src/push/fcm.rs (1)
639-643: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAlign the 401 comment with guarded invalidation.
Line 641 says the cached token is evicted "unconditionally", but Line 643 uses
invalidate_if_matches, preserving a concurrently refreshed token. Please update the comment so future changes don’t remove that guard accidentally.📝 Proposed wording fix
- // cached token unconditionally and ask the retry engine to + // cached token only if it still matches the rejected credential, + // then ask the retry engine to🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/push/fcm.rs` around lines 639 - 643, The comment in the FCM auth-error handling is inaccurate because `self.token_cache.invalidate_if_matches(access_token).await` is guarded, not unconditional. Update the wording near this `invalidate_if_matches` call to describe conditional eviction based on the matched `access_token`, so the intent stays aligned with the existing concurrency-safe behavior and future edits don’t remove the guard.src/push/mod.rs (1)
4-4: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winKeep
authcrate-private unless this is intended public API.The shared auth machinery is consumed internally and its types are
pub(crate)in the provided context. Declaring the module aspubstill exposes a new public namespace; preferpub(crate) mod auth;unless external callers are meant to depend on it. If it is intentional public API, add/update API docs. As per coding guidelines, "Update documentation for public API changes."♻️ Proposed fix
-pub mod auth; +pub(crate) mod auth;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/push/mod.rs` at line 4, Change the `auth` module declaration in `mod.rs` from public to crate-private so it does not create an unintended external API surface. Update the module visibility to `pub(crate) mod auth;` unless `auth` is meant to be consumed by external callers; if it must remain public, then add or update the relevant public API documentation for the `auth` namespace and its exposed items.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/push/apns.rs`:
- Around line 1426-1500: The current tests cover only handle_response, but the
retry path in send() also needs coverage. Add an integration-style test around
ApnsClient::send that exercises with_retry and send_once: start with a seeded
JWT, return 403 ExpiredProviderToken on the first APNs response, verify the
cached token is evicted, then ensure send() retries with a fresh JWT and
succeeds on the second response. Use the existing test helpers and symbols like
ApnsClient::mock, seed_token, handle_response, send, and with_retry so the new
test directly validates the auth-retry flow end to end.
In `@src/push/dispatcher.rs`:
- Around line 49-53: The semaphore comment in dispatcher should be reworded
because it overstates the decrypted-token residency guarantee; update the
wording around the live-task cap in the dispatcher logic to say it limits
spawned live tasks rather than total token residency, and keep the explanation
aligned with the Zeroizing<String> device tokens that remain in the queue before
task spawn.
---
Outside diff comments:
In `@src/push/apns.rs`:
- Around line 341-357: The public `send` method in `apns.rs` now assumes tokens
are already normalized upstream, so its docs need to state that precondition
explicitly. Update the documentation for `Push::send` to note that
`device_token` must be even-length lowercase hex produced by
`TokenPayload::device_token_hex()` and that malformed tokens are not rejected
locally in release builds. Mention the existing `debug_assert!` behavior so
callers understand the contract change.
In `@src/push/fcm.rs`:
- Around line 322-346: The JWT audience in the FCM OAuth flow is still built
from sa.token_uri instead of the resolved endpoint used by oauth_token_url(sa),
which breaks fallback configurations when token_uri is empty. Update the
token-creation path in the FCM code so OAuthClaims.aud is derived from the same
resolved OAuth URL that the request posts to, keeping the JWT audience and POST
target consistent.
---
Nitpick comments:
In `@src/push/fcm.rs`:
- Around line 639-643: The comment in the FCM auth-error handling is inaccurate
because `self.token_cache.invalidate_if_matches(access_token).await` is guarded,
not unconditional. Update the wording near this `invalidate_if_matches` call to
describe conditional eviction based on the matched `access_token`, so the intent
stays aligned with the existing concurrency-safe behavior and future edits don’t
remove the guard.
In `@src/push/mod.rs`:
- Line 4: Change the `auth` module declaration in `mod.rs` from public to
crate-private so it does not create an unintended external API surface. Update
the module visibility to `pub(crate) mod auth;` unless `auth` is meant to be
consumed by external callers; if it must remain public, then add or update the
relevant public API documentation for the `auth` namespace and its exposed
items.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 03b60c61-bf65-4ac3-aa2e-501dabacb445
📒 Files selected for processing (7)
src/error.rssrc/push/apns.rssrc/push/auth.rssrc/push/dispatcher.rssrc/push/fcm.rssrc/push/mod.rssrc/push/retry.rs
Comment on lines
+1426
to
+1500
| async fn test_jwt_reason_403_evicts_and_asks_for_a_fresh_token_retry() { | ||
| // A provider-token 403 (ExpiredProviderToken) means the cached JWT is | ||
| // the problem: evict it (issue #145) and return AuthRejected so the | ||
| // retry engine retries once with a fresh JWT (issue #85). | ||
| let mock_server = MockServer::start().await; | ||
|
|
||
| Mock::given(method("POST")) | ||
| .and(path_regex(r"/3/device/[a-f0-9]+")) | ||
| .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({ | ||
| "reason": "BadJwtToken" | ||
| "reason": "ExpiredProviderToken" | ||
| }))) | ||
| .expect(1) | ||
| .mount(&mock_server) | ||
| .await; | ||
|
|
||
| let client = ApnsClient::mock(test_config(), false); | ||
| { | ||
| let mut cached = client.cached_token.write().await; | ||
| *cached = Some(CachedToken { | ||
| token: Zeroizing::new("poisoned-jwt".to_string()), | ||
| expires_at: SystemTime::now() + TOKEN_LIFETIME, | ||
| }); | ||
| } | ||
| client | ||
| .seed_token("poisoned-jwt", SystemTime::now() + TOKEN_LIFETIME) | ||
| .await; | ||
|
|
||
| let response = Client::new() | ||
| .post(format!("{}/3/device/{}", mock_server.uri(), "deadbeef1234")) | ||
| .send() | ||
| .await | ||
| .unwrap(); | ||
|
|
||
| let result = client | ||
| .handle_response(Duration::ZERO, response, "poisoned-jwt") | ||
| let result = client.handle_response(response, "poisoned-jwt").await; | ||
|
|
||
| assert!(matches!( | ||
| result, | ||
| SendAttemptResult::AuthRejected(Error::Apns(ref message)) | ||
| if message.contains("ExpiredProviderToken") | ||
| )); | ||
| assert_eq!(client.cached_token_value().await, None); | ||
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn test_non_jwt_403_does_not_evict_and_is_permanent() { | ||
| // A non-token 403 (BadCertificateEnvironment) is a static | ||
| // misconfiguration: the cached JWT must survive (no re-sign stampede, | ||
| // issue #145) and the result is a permanent error (not retriable). | ||
| let mock_server = MockServer::start().await; | ||
|
|
||
| Mock::given(method("POST")) | ||
| .and(path_regex(r"/3/device/[a-f0-9]+")) | ||
| .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({ | ||
| "reason": "BadCertificateEnvironment" | ||
| }))) | ||
| .expect(1) | ||
| .mount(&mock_server) | ||
| .await; | ||
|
|
||
| let client = ApnsClient::mock(test_config(), false); | ||
| client | ||
| .seed_token("valid-jwt", SystemTime::now() + TOKEN_LIFETIME) | ||
| .await; | ||
|
|
||
| let response = Client::new() | ||
| .post(format!("{}/3/device/{}", mock_server.uri(), "deadbeef1234")) | ||
| .send() | ||
| .await | ||
| .unwrap(); | ||
|
|
||
| let result = client.handle_response(response, "valid-jwt").await; | ||
|
|
||
| assert!(matches!( | ||
| result, | ||
| SendAttemptResult::Permanent(Error::Apns(ref message)) | ||
| if message.contains("BadJwtToken") | ||
| if message.contains("BadCertificateEnvironment") | ||
| )); | ||
| assert!(client.cached_token.read().await.is_none()); | ||
| // The valid JWT must NOT be evicted by a non-token 403. | ||
| assert_eq!( | ||
| client.cached_token_value().await.as_deref(), | ||
| Some("valid-jwt") | ||
| ); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add a full send() auth-retry test.
These tests validate handle_response, but the new behavior also depends on with_retry calling send_once again after AuthRejected. Please add an integration-style test where send() gets 403 ExpiredProviderToken, evicts the seeded JWT, mints/uses a fresh JWT, and succeeds on the second APNs response. As per coding guidelines, **/*.rs: Include tests for new functionality.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/push/apns.rs` around lines 1426 - 1500, The current tests cover only
handle_response, but the retry path in send() also needs coverage. Add an
integration-style test around ApnsClient::send that exercises with_retry and
send_once: start with a seeded JWT, return 403 ExpiredProviderToken on the first
APNs response, verify the cached token is evicted, then ensure send() retries
with a fresh JWT and succeeds on the second response. Use the existing test
helpers and symbols like ApnsClient::mock, seed_token, handle_response, send,
and with_retry so the new test directly validates the auth-retry flow end to
end.
Source: Coding guidelines
Comment on lines
+49
to
+53
| /// A second semaphore, acquired at task *spawn* and held for the task's whole | ||
| /// life (never released during backoff), caps the live-task count | ||
| /// independently. Sizing it at 2x the active-concurrency limit leaves generous | ||
| /// headroom for tasks legitimately sleeping in backoff while still bounding | ||
| /// decrypted-token residency to a small constant instead of the 10k queue. |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
Reword the live-task cap guarantee to avoid overstating token residency bounds.
The queue still stores Zeroizing<String> device tokens before they become send tasks, so this semaphore bounds spawned live tasks, not total decrypted-token residency.
Suggested wording
-/// independently. Sizing it at 2x the active-concurrency limit leaves generous
-/// headroom for tasks legitimately sleeping in backoff while still bounding
-/// decrypted-token residency to a small constant instead of the 10k queue.
+/// independently. Sizing it at 2x the active-concurrency limit leaves generous
+/// headroom for tasks legitimately sleeping in backoff while preventing a retry
+/// storm from draining the whole queue into spawned token-holding tasks.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| /// A second semaphore, acquired at task *spawn* and held for the task's whole | |
| /// life (never released during backoff), caps the live-task count | |
| /// independently. Sizing it at 2x the active-concurrency limit leaves generous | |
| /// headroom for tasks legitimately sleeping in backoff while still bounding | |
| /// decrypted-token residency to a small constant instead of the 10k queue. | |
| /// A second semaphore, acquired at task *spawn* and held for the task's whole | |
| /// life (never released during backoff), caps the live-task count | |
| /// independently. Sizing it at 2x the active-concurrency limit leaves generous | |
| /// headroom for tasks legitimately sleeping in backoff while preventing a retry | |
| /// storm from draining the whole queue into spawned token-holding tasks. |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/push/dispatcher.rs` around lines 49 - 53, The semaphore comment in
dispatcher should be reworded because it overstates the decrypted-token
residency guarantee; update the wording around the live-task cap in the
dispatcher logic to say it limits spawned live tasks rather than total token
residency, and keep the explanation aligned with the Zeroizing<String> device
tokens that remain in the queue before task spawn.
The signed assertion `aud` claim used `sa.token_uri.clone()` with no empty-fallback, while the POST target went through the empty-check helper. For a parseable-but-degenerate service account with `"token_uri": ""` this signed aud="" while POSTing to OAUTH_TOKEN_URL — the exact audience mismatch #153 exists to prevent. Extract signed_audience(sa) as the single source of truth for the token endpoint (token_uri, or the constant when empty). Both the aud claim and oauth_token_url's production branch derive from it, so they can never diverge. The test override still redirects only the request target (to a mock) while aud reflects the real token_uri, the intended test shape. Adds tests asserting aud == POST target for both a standard and an empty-token_uri service account. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
erskingardner
added a commit
that referenced
this pull request
Jul 5, 2026
Reconciliation regression for the merge with the push-provider rework (#233), which added DeliveryHealth streak state feeding /ready. A token dropped by the #177 pre-filter in process_inner never reaches the send path, so it is not a delivery attempt and must not touch a provider's DeliveryHealth streak. This test seeds a real FCM hard-failure streak just below the flagging threshold, processes an unconfigured-APNs drop, and asserts APNs stays delivering (no spurious streak increment) and FCM's seeded streak survives (one more failure flips it), proving the pre-filter neither increments nor resets the streak — no double-count, no missed update. Part of #218. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
erskingardner
added a commit
that referenced
this pull request
Jul 5, 2026
…apacity path (#234) * fix: shard rate limiter and add live cache-size gauge Two capacity-path fixes for the rate limiter (#123, #125). #123 — bounded/sharded capacity eviction. The admission path ran an O(CLEANUP_BATCH_SIZE=1000) eviction scan while holding the limiter's single global write lock, so a key-cardinality flood serialized every rate-limit check behind that scan. Fix on two axes: - Shard the limiter into up to MAX_SHARDS (32) independent RwLock<LruCache> stripes keyed by hash(key) % N, so checks for keys in different stripes never contend on the same lock. Tiny caches (< MIN_ENTRIES_PER_SHARD = 256) stay single-stripe so small-capacity LRU eviction semantics are byte-for-byte identical to before. Per-stripe capacities sum to exactly max_entries. - Use a small admission-path scan bound (ADMISSION_SCAN_LIMIT = 32), distinct from the periodic-cleanup CLEANUP_BATCH_SIZE (1000), so a full stripe costs bounded, small work per check. The #206 reservation identity is preserved: next_hit_id remains a single process-wide AtomicU64 shared across stripes, so reservation ids stay globally unique and a delayed rollback can never remove a newer hit. The #210 admission-eviction reporting is unchanged. cleanup() now sweeps every stripe (up to CLEANUP_BATCH_SIZE each). #125 — live cache-size gauge. check_and_increment now returns a sampled total entry count (RateLimitCheck::sampled_cache_len) on roughly one in GAUGE_SAMPLE_INTERVAL (64) stripe-mutating admissions, summed across stripes to keep the metric's per-cache_type label shape. The caller publishes it so transponder_rate_limit_cache_size tracks growth toward capacity instead of lagging until the 60s cleanup tick, without a metric write per check. Adds tests for single-vs-sharded stripe selection, capacity summing, cross-stripe reservation uniqueness, the bounded admission scan, the sampled gauge, and concurrent cross-stripe admissions. Two large-cache tests that asserted the old 1000-wide global scan were rewritten: one to a single-stripe variant preserving the exact semantic, plus a new sharding-aware sibling. Part of #218. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix: expose pre-charge dispatch admission checks Add PushDispatcher::accepts(platform) and PushDispatcher::token_is_encodable(payload) so the event processor can reject undispatchable tokens BEFORE charging the rate limiters (#177). accepts mirrors dispatch()'s platform-client presence filter; the token-encoding check mirrors its FCM UTF-8 filter. The existing filters inside dispatch() are kept as a defensive second layer (and their comment now says so) — in the normal event path they no longer fire because process_inner pre-filters, but dispatch still never enqueues an undeliverable token if a caller skips the pre-filter. Adds unit tests for both helpers. Part of #218. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix: transactional rate-limit admission lifecycle Make per-token rate-limit charging transactional in process_inner so no admission path can strand a spent increment (#170, #177), fold the redundant dedup lock on the success path (#197), and wire the live rate-limit cache-size gauge (#125). AdmissionGuard (RAII-style, explicit-consume). After the encrypted-token limiter charges a token, the charge is held by an AdmissionGuard that every exit path must resolve: - commit() — fully admitted; hands the encrypted+device charges to the dispatch-failure rollback ledger. - refund() — dropped after charging; rolls back every charge held. - keep_charge() — dropped but the charge is intentionally retained. The guard's Drop asserts it was resolved. Because the limiter refund is async (the stripe lock is a tokio RwLock), a blocking Drop-time refund isn't viable, so this is an explicit-consume guard rather than fire-and-forget RAII — the trade the tracker calls out as acceptable. It still centralizes the refund decision per exit path. #170 — device-limiter reject now refunds the encrypted charge. Previously a device reject `continue`d without rolling back the encrypted-token increment already spent for that token, so a legitimate transient redelivery could read as a replay. guard.refund() on the device-reject branch fixes this. The decrypt-failure path keeps its charge via guard.keep_charge() (documented replay defense). #177 — dispatch-level drops refund and are counted. Tokens targeting an unconfigured platform, or (FCM) carrying non-UTF-8 device bytes, were dropped silently inside dispatch() AFTER both limiters were charged, with no metric. process_inner now pre-filters via dispatcher.accepts() / token_is_encodable() right after decrypt (platform is only known then) and before the device charge: it refunds the encrypted charge and records record_push_failed(platform, "unconfigured" | "invalid_encoding"). These drops mark the event terminally dropped (not a retryable rate shed). #197 — single dedup write-lock on the success path. mark_seen refreshed the completion timestamp by re-put'ing the entry under a second lock, redundantly re-writing the dedup_cache_size gauge. SeenEventStore gains mark_terminal, which mutates the existing reservation entry in place (refreshing seen_at, flipping terminal, promoting LRU position) and reports whether the set size changed, so the gauge is written only when it actually moved. Completion-timestamp-refresh and release_reservation semantics are preserved. #125 — the sampled cross-stripe cache length from check_and_increment is published to transponder_rate_limit_cache_size on the admission path. Tests: device-reject refund, unconfigured-platform and non-UTF-8 FCM drop (refund + metric), decrypt-failure charge retention, a multi-thread concurrent device-reject interleaving test proving no encrypted charge is stranded, the live gauge updating before cleanup, and the single dedup-gauge update. The create_processor_with_rate_limits helper now wires a configured APNs mock so its device/encrypted rate-limit tests exercise dispatchable tokens (previously they relied on the #177 bug of charging budget for tokens dropped silently by an unconfigured dispatcher). Fixes #170 Fixes #177 Fixes #197 Part of #218. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test: assert pre-filter drop leaves delivery-health untouched Reconciliation regression for the merge with the push-provider rework (#233), which added DeliveryHealth streak state feeding /ready. A token dropped by the #177 pre-filter in process_inner never reaches the send path, so it is not a delivery attempt and must not touch a provider's DeliveryHealth streak. This test seeds a real FCM hard-failure streak just below the flagging threshold, processes an unconfigured-APNs drop, and asserts APNs stays delivering (no spurious streak increment) and FCM's seeded streak survives (one more failure flips it), proving the pre-filter neither increments nor resets the streak — no double-count, no missed update. Part of #218. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <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 join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
Summary
Unifies the APNs and FCM push-provider machinery — which had become ~95%-identical, already-divergent copy-pastes — behind shared credential and outcome handling, and bounds the retry policy. Implements trackers #214 (provider unification) and #220 (retry policy) as one PR since they share
src/push/retry.rs.A. Shared auth machinery — new
src/push/auth.rs(#158)trait AuthTokenGenerator { async fn mint(&self) -> Result<MintedToken, TokenAcquisitionError> }withMintedToken { token: Zeroizing<String>, expires_at }.TokenCache<G>: read-lock fast path; a single-flight refresh that runs outside the cache write lock ([LOW] FCM access-token refresh holds the cache write lock across the OAuth HTTP round-trip, stalling all concurrent senders during refresh #86). A dedicated refreshMutexcoalesces concurrent misses onto one mint via a double-check, and readers keep serving the still-valid token during a refresh — the mint (network IO for FCM) never holds thecachedwrite lock.invalidate_if_matches(failing_token)evicts only while the cached entry is still the failing token. Whether a rejection is credential-related is the caller's decision:reasonbefore invalidation and evicts only forExpiredProviderToken/InvalidProviderToken— non-JWT 403s (BadCertificateEnvironment,Forbidden, …) no longer trigger a per-notification ES256 re-sign stampede ([MEDIUM] APNs evicts the cached JWT on every HTTP 403 reason, causing a per-notification ES256 re-sign stampede under non-JWT 403s (e.g. BadCertificateEnvironment) #145).refresh_token_innerwholesale, keeping theTokenAcquisitionErrorretriable/permanent classification (PR fix: retry transient FCM OAuth token failures #209) andexpires_inTTL (PR fix: honor FCM OAuth token expiry #212); POSTs tosa.token_uri(falling back to the constant only when empty) so the signedaudand endpoint agree ([LOW] FCM OAuth JWTaudis signed withservice_account.token_uribut the request is always POSTed to the hardcodedOAUTH_TOKEN_URL— a non-standardtoken_uricauses a silent total FCM outage #153); reads the OAuth success body through the repo's bounded reader (parse_bounded_json_body, an 8 KiB cap, buffering through a zeroizingVec) instead of unboundedresponse.json()([LOW] FCM OAuth token success response body is read with unboundedresponse.json()#154).TokenCache; the duplicatedCachedToken/get_token/invalidatecode is deleted.B. Request templates (#198, #126)
HashMap/FcmRequest/device_token.to_string()rebuild is gone; the body template borrows the device token viaFcmMessage<'a>, and APNs hoists itsApnsRequestParts+ zeroizing device-token URL out of the closure.build_requestcall sites as the accepted [LOW] FCM/APNs copy the decrypted device token into non-zeroized String/JSON body inside the request builder, defeating the upstream Zeroizing wrapper #126 posture — it cannot be zeroized without reimplementing the client.C. Outcome / metrics polish (#168, #85, #199, #172)
transponder_push_request_duration_secondsis now recorded once per logical push insend()after retries resolve, so its count equals push volume rather than HTTP-attempt volume. Per-attempttransponder_push_response_status_totalstill increments once per attempt inhandle_response(that was already per-attempt; behavior unchanged). Dashboards keyed onpush_request_duration_*_countas an attempt counter will now see push counts.SendAttemptResult::AuthRejected— after an auth invalidation (APNs 403 JWT-reason / FCM 401) the in-hand notification is retried once with a freshly minted credential instead of being dropped. The auth-retry budget is one-shot and separate from the transient 429/5xx budget; a genuinely bad key re-fails and surfaces the error instead of looping.is_valid_device_tokenodd-length / non-hex / empty branches are unreachable on the production dispatch path — dead defensive code that reads as meaningful validation #199: removed the unreachableis_valid_device_tokenbranches; adebug_assert!documentsTokenPayload::from_decrypted+ the dispatcher UTF-8 gate as the single source of device-token well-formedness. (APNs no longer re-validates on the dispatch path — see the adaptedsend_push_records_invalid_token_metricstest.)reqwest::ErrorURL on transport failures #172: every provider/transport error that can carry a URL passes throughreqwest::Error::without_url()(Error::redact_transport_url) before conversion/logging — the APNs URL embeds the device token. Audited and covered:retry.rstransport-retrywarn!sites, the send-path conversion sites, and the dispatcherdebug!sites. Tests assert a mapped reqwest error'sDisplaycontains no token (error.rs,retry.rs,apns.rs).D. Retry policy (#220: #162, #160)
Retry-Afteris capped —server_delay.clamp(MIN_RETRY_BACKOFF, MAX_RETRY_AFTER)withMAX_RETRY_AFTER = 60s— so an untrusted header can no longer pin a token-holding send task (and block drain) arbitrarily long. The floor + jitter from PR fix: desynchronize push retry backoff #204 are preserved.MAX_LIVE_SEND_TASKS = 2× MAX_CONCURRENT_PUSHES = 200) is acquired at task spawn in the dispatcher recv loop and held for the task's whole life — not released during backoff. Since a retrying task releases its concurrency permit during backoff sleeps but stays alive holding a decrypted token, the old design let a provider 429/5xx storm drain the entire 10k queue into thousands of live token-holding tasks. The live-task semaphore caps live tasks (active + sleeping-in-backoff) at a small constant, applying backpressure to the queue instead.InFlightTrackerdrain semantics are unchanged.Metric semantic changes
transponder_push_request_duration_seconds: now one sample per logical push (was one per HTTP attempt). Count == notification volume.push_response_status_total,push_retries_total, andauth_token_refreshes_totalkeep their labels and per-attempt semantics.Testing / gates
cargo fmt --checkclean;cargo clippy --all-targets -- -D warnings(and--features tor) zero;cargo test --bin transponder590 passed / 0 failed;cargo check --all-targets --features torclean. Coverage: newauth.rsat 99.65% line / 98.92% region; all changed push files 96–99%. Tests that changed because semantics intentionally changed are called out in their commit messages (APNs 403 reason gating →AuthRejected, per-push duration, #199 removal).Fixes #158
Fixes #145
Fixes #86
Fixes #153
Fixes #154
Fixes #198
Fixes #168
Fixes #199
Fixes #85
Fixes #172
Fixes #126
Fixes #162
Fixes #160
Part of #214. Closes out #220.
🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Performance