diff --git a/.sampo/changesets/capture-retry-defaults.md b/.sampo/changesets/capture-retry-defaults.md new file mode 100644 index 00000000..f8f6ceb1 --- /dev/null +++ b/.sampo/changesets/capture-retry-defaults.md @@ -0,0 +1,5 @@ +--- +cargo/posthog-rs: minor +--- + +Capture now makes 4 attempts by default instead of 3, and the first retry waits 100 ms instead of 200 ms. Remote `/flags` retries no longer use the capture retry options. They wait 300 ms before the first retry and double up to 30 s. diff --git a/docs/migration-0.x-to-1.0.md b/docs/migration-0.x-to-1.0.md index 03a0646b..5759268c 100644 --- a/docs/migration-0.x-to-1.0.md +++ b/docs/migration-0.x-to-1.0.md @@ -92,6 +92,8 @@ One behavior differs from 0.x without `capture-v1`, which used the V0 endpoint: The V1 endpoint returns a result for each event. The SDK retries transient request failures and only the events with retryable results from a partial response. +Capture now makes 4 attempts by default (`max_capture_attempts`, was 3). The first retry waits 100 ms (`retry_initial_backoff_ms`, was 200 ms), and each later wait doubles up to `retry_max_backoff_ms`. These options apply to `capture` and `capture_ai`. Remote `/flags` retries no longer use them. They wait 300 ms before the first retry (was 200 ms by default) and double up to 30 s. `feature_flags_request_max_retries` still sets how many `/flags` retries the SDK makes. + `CaptureSummary::not_persisted()` and `CaptureSummary::all_persisted()` now use those per-event results. In the V0 path they reported a successful `2xx` as fully persisted without per-event confirmation. Applications that advance durable state after `capture_immediate` should check both that `submitted()` equals the number of intended events and that `all_persisted()` is true, because disabled clients and fully `before_send`-filtered batches submit no events but still report `all_persisted()` as true. HTTP 429 is not a retryable V1 capture status. The V1 service uses HTTP 402 for billing limits and per-event `drop` or `warning` results in successful responses. `Retry-After` is still honored for retryable failures and retry results. diff --git a/src/client/async_client.rs b/src/client/async_client.rs index 10b4efa9..a39e65c7 100644 --- a/src/client/async_client.rs +++ b/src/client/async_client.rs @@ -691,9 +691,11 @@ impl Client { /// `capture_ai_compression`, and the batch is split into requests of about 5 MiB /// each, sent in order. Each request retries transient failures and /// per-event `retry` verdicts per the client's retry configuration; the - /// summary merges every request's verdicts. Unlike the background lane, no - /// event is dropped locally for size: an oversize event is sent and comes - /// back as a per-event `drop` (`ai_event_too_big`) in the summary. + /// summary merges every request's verdicts. An event whose serialized + /// `properties` are over 8 MiB is not sent, as the backend would refuse it. + /// It gets a local `drop` verdict with `details` `ai_event_too_big` in the + /// summary, and counts in [`CaptureSummary::submitted`] and + /// [`CaptureSummary::not_persisted`]. /// /// # Errors /// @@ -778,12 +780,11 @@ impl Client { use super::capture::{self, Step}; let compression = lane.compression(&self.options); - let mut submitted = 0; - let mut final_results = HashMap::new(); + let plan = capture::prepare_immediate(&self.options, lane, events, historical_migration); + let mut submitted = plan.local_results.len(); + let mut final_results = plan.local_results; - for mut prep in - capture::prepare_immediate(&self.options, lane, events, historical_migration) - { + for mut prep in plan.requests { let mut attempt: u32 = 1; loop { let (headers, body) = capture::build_attempt_parts( diff --git a/src/client/blocking.rs b/src/client/blocking.rs index 12673f1f..47ee42db 100644 --- a/src/client/blocking.rs +++ b/src/client/blocking.rs @@ -623,9 +623,11 @@ impl Client { /// `capture_ai_compression`, and the batch is split into requests of about 5 MiB /// each, sent in order. Each request retries transient failures and /// per-event `retry` verdicts per the client's retry configuration; the - /// summary merges every request's verdicts. Unlike the background lane, no - /// event is dropped locally for size: an oversize event is sent and comes - /// back as a per-event `drop` (`ai_event_too_big`) in the summary. + /// summary merges every request's verdicts. An event whose serialized + /// `properties` are over 8 MiB is not sent, as the backend would refuse it. + /// It gets a local `drop` verdict with `details` `ai_event_too_big` in the + /// summary, and counts in [`CaptureSummary::submitted`] and + /// [`CaptureSummary::not_persisted`]. /// /// # Errors /// @@ -708,12 +710,11 @@ impl Client { use super::capture::{self, Step}; let compression = lane.compression(&self.options); - let mut submitted = 0; - let mut final_results = HashMap::new(); + let plan = capture::prepare_immediate(&self.options, lane, events, historical_migration); + let mut submitted = plan.local_results.len(); + let mut final_results = plan.local_results; - for mut prep in - capture::prepare_immediate(&self.options, lane, events, historical_migration) - { + for mut prep in plan.requests { let mut attempt: u32 = 1; loop { let (headers, body) = capture::build_attempt_parts( diff --git a/src/client/capture.rs b/src/client/capture.rs index dc7bc261..63dd36d0 100644 --- a/src/client/capture.rs +++ b/src/client/capture.rs @@ -212,6 +212,14 @@ pub(crate) struct Prepared { pub(crate) submitted: usize, } +/// The requests of an inline immediate capture, plus the verdicts the SDK +/// decided locally for events it does not send. +#[derive(Default)] +pub(crate) struct ImmediatePlan { + pub(crate) requests: Vec, + pub(crate) local_results: HashMap, +} + /// Per-lane settings for an inline immediate capture. The analytics lane sends /// one request; the AI lane splits the batch by bytes. Each reads its own /// compression option. @@ -220,6 +228,9 @@ pub(crate) struct ImmediateLane { pub(crate) endpoint: Endpoint, /// `None`: everything goes in one request. pub(crate) batch_bytes_target: Option, + /// Ceiling on an event's serialized `properties`. A larger event is not + /// sent and gets a local `drop` verdict. `None`: no ceiling. + pub(crate) max_event_bytes: Option, } impl ImmediateLane { @@ -227,6 +238,7 @@ impl ImmediateLane { Self { endpoint: Endpoint::Capture, batch_bytes_target: None, + max_event_bytes: None, } } @@ -234,6 +246,7 @@ impl ImmediateLane { Self { endpoint: Endpoint::CaptureAi, batch_bytes_target: Some(super::transport::AI_BATCH_BYTES_TARGET), + max_event_bytes: Some(super::transport::AI_MAX_EVENT_BYTES), } } @@ -246,46 +259,80 @@ impl ImmediateLane { } } +/// The `details` of the local verdict for an AI event over the size ceiling, +/// the same code the backend uses when it refuses one. +const AI_EVENT_TOO_BIG: &str = "ai_event_too_big"; + +/// Measure each event, and keep only those within `max`. An event whose +/// serialized `properties` are strictly larger than `max` gets a local `drop` +/// verdict instead, as the backend would refuse it. +fn measure_within_limit( + events: Vec, + max: Option, + local_results: &mut HashMap, +) -> Vec<(CaptureEvent, usize)> { + events + .into_iter() + .filter_map(|event| { + let size = measure_event(&event); + if max.is_some_and(|max| size.properties > max) { + local_results.insert( + event.uuid, + EventResult { + result: EventStatus::Drop, + details: Some(AI_EVENT_TOO_BIG.to_string()), + }, + ); + return None; + } + Some((event, size.total)) + }) + .collect() +} + /// Prepare an inline immediate capture: apply client defaults + `before_send`, /// then build the wire events and the per-request identity (request id, -/// `created_at`, URL) for each request. Returns an empty `Vec` when nothing +/// `created_at`, URL) for each request. Returns no requests when nothing /// survives filtering (an empty or fully `before_send`-dropped batch), so the /// caller returns a default summary without sending. The analytics lane always /// yields exactly one `Prepared`; the AI lane yields one per byte-sized chunk, -/// in order. +/// in order. An event over the lane's `max_event_bytes` is not sent; its local +/// `drop` verdict is in `local_results`. pub(crate) fn prepare_immediate( opts: &ClientOptions, lane: ImmediateLane, events: Vec, historical_migration: bool, -) -> Vec { +) -> ImmediatePlan { let defaults = opts.capture_defaults(); let events: Vec = events .into_iter() .filter_map(|event| preprocess_capture_event(event, &defaults, &opts.before_send)) .collect(); if events.is_empty() { - return Vec::new(); + return ImmediatePlan::default(); } + let mut local_results = HashMap::new(); let historical_migration = historical_migration.then_some(true); let url = opts.endpoints().build_url(lane.endpoint); let built = build_events_at(&events, Utc::now()); let chunks = match lane.batch_bytes_target { Some(target) => chunk_by_bytes( - built - .into_iter() - .map(|event| { - let bytes = measure_event(&event).total; - (event, bytes) - }) - .collect(), + measure_within_limit(built, lane.max_event_bytes, &mut local_results), target, ), - None => vec![built], + None if lane.max_event_bytes.is_none() => vec![built], + None => vec![ + measure_within_limit(built, lane.max_event_bytes, &mut local_results) + .into_iter() + .map(|(event, _)| event) + .collect(), + ], }; - chunks + let requests = chunks .into_iter() + .filter(|pending| !pending.is_empty()) .map(|pending| Prepared { url: url.clone(), request_id: Uuid::now_v7(), @@ -294,7 +341,11 @@ pub(crate) fn prepare_immediate( submitted: pending.len(), pending, }) - .collect() + .collect(); + ImmediatePlan { + requests, + local_results, + } } /// Build the headers and (optionally compressed) body for one immediate V1 @@ -1147,7 +1198,7 @@ mod tests { e }) .collect(); - let prepared = prepare_immediate(&opts, analytics, events.clone(), false); + let prepared = prepare_immediate(&opts, analytics, events.clone(), false).requests; assert_eq!(prepared.len(), 1); assert_eq!(prepared[0].submitted, 3); assert!(prepared[0].url.ends_with("/i/v1/analytics/events")); @@ -1164,7 +1215,7 @@ mod tests { batch_bytes_target: Some(2 * size + size / 2), ..ai }; - let prepared = prepare_immediate(&opts, small_target, events.clone(), true); + let prepared = prepare_immediate(&opts, small_target, events.clone(), true).requests; assert_eq!(prepared.len(), 2); assert_eq!(prepared[0].pending.len(), 2); assert_eq!(prepared[1].pending.len(), 1); @@ -1181,7 +1232,57 @@ mod tests { .api_key("phc_test".to_string()) .before_send(|_| None); let filtered = filtered.build().unwrap(); - assert!(prepare_immediate(&filtered, ai, events, false).is_empty()); + let plan = prepare_immediate(&filtered, ai, events, false); + assert!(plan.requests.is_empty()); + assert!(plan.local_results.is_empty()); + } + + #[test] + fn prepare_immediate_drops_events_over_the_lane_ceiling_locally() { + let mut builder = ClientOptionsBuilder::default(); + builder.api_key("phc_test".to_string()); + let opts = builder.build().unwrap(); + let event = |blob: usize| { + let mut e = Event::new("$ai_generation".to_string(), "user-1".to_string()); + e.insert_prop("blob", "x".repeat(blob)).unwrap(); + e + }; + let unlimited = ImmediateLane { + max_event_bytes: None, + ..ImmediateLane::ai() + }; + let sent = prepare_immediate(&opts, unlimited, vec![event(64)], false); + let limit = measure_event(&sent.requests[0].pending[0]).properties; + + // An event exactly at the ceiling is sent; one byte over is a local drop. + // The analytics lane covers a ceiling without byte chunking. + for lane in [ImmediateLane::ai(), ImmediateLane::analytics()] { + let lane = ImmediateLane { + max_event_bytes: Some(limit), + ..lane + }; + let (mut at_limit, mut over) = (event(64), event(65)); + let (at_limit_uuid, over_uuid) = (Uuid::now_v7(), Uuid::now_v7()); + at_limit.set_uuid(at_limit_uuid); + over.set_uuid(over_uuid); + let plan = prepare_immediate(&opts, lane, vec![at_limit, over], false); + assert_eq!(plan.requests.len(), 1, "{:?}", lane.endpoint); + assert_eq!(plan.requests[0].submitted, 1); + assert_eq!(plan.requests[0].pending[0].uuid, at_limit_uuid); + assert_eq!(plan.local_results.len(), 1); + let verdict = &plan.local_results[&over_uuid]; + assert_eq!(verdict.result, EventStatus::Drop); + assert_eq!(verdict.details.as_deref(), Some("ai_event_too_big")); + } + + // Every event over the ceiling: no request, only local verdicts. + let lane = ImmediateLane { + max_event_bytes: Some(limit), + ..ImmediateLane::ai() + }; + let plan = prepare_immediate(&opts, lane, vec![event(65), event(66)], false); + assert!(plan.requests.is_empty()); + assert_eq!(plan.local_results.len(), 2); } #[test] diff --git a/src/client/mod.rs b/src/client/mod.rs index 28139849..3bc69c34 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -236,16 +236,17 @@ pub struct ClientOptions { #[builder(default = "false")] local_evaluation_only: bool, - /// Maximum number of attempts for capture requests (default: 3). - /// Includes the initial attempt, so `3` means 1 initial + 2 retries. - #[builder(default = "3")] + /// Maximum number of attempts for capture requests (default: 4). + /// Includes the initial attempt, so `4` means 1 initial + 3 retries. + #[builder(default = "4")] pub(crate) max_capture_attempts: u32, - /// Initial retry backoff duration in milliseconds (default: 200) - #[builder(default = "200")] + /// Initial capture retry backoff duration in milliseconds (default: 100). + /// Remote `/flags` retries use their own fixed backoff. + #[builder(default = "100")] pub(crate) retry_initial_backoff_ms: u64, - /// Maximum retry backoff duration in milliseconds (default: 30000) + /// Maximum capture retry backoff duration in milliseconds (default: 30000) #[builder(default = "30000")] pub(crate) retry_max_backoff_ms: u64, diff --git a/src/client/retry.rs b/src/client/retry.rs index 2e9cb862..22bb25bd 100644 --- a/src/client/retry.rs +++ b/src/client/retry.rs @@ -119,15 +119,33 @@ pub(crate) fn backoff_duration( attempt: u32, retry_after: Option, ) -> Duration { - let base_ms = opts.retry_initial_backoff_ms; let max_ms = opts.retry_max_backoff_ms; - let backoff_ms = base_ms.saturating_mul(2u64.saturating_pow(attempt.saturating_sub(1))); - let configured_delay = Duration::from_millis(backoff_ms.min(max_ms)); + let configured_delay = exponential_backoff(opts.retry_initial_backoff_ms, max_ms, attempt); let max_backoff = Duration::from_millis(max_ms); let clamped_retry_after = retry_after.map(|d| d.min(max_backoff)); clamped_retry_after.map_or(configured_delay, |d| configured_delay.max(d)) } +/// `base_ms` doubled for each attempt after the first, clamped to `max_ms`. +fn exponential_backoff(base_ms: u64, max_ms: u64, attempt: u32) -> Duration { + let backoff_ms = base_ms.saturating_mul(2u64.saturating_pow(attempt.saturating_sub(1))); + Duration::from_millis(backoff_ms.min(max_ms)) +} + +/// Base of the `/flags` retry backoff. It is separate from the capture retry +/// options and matches the `/flags` backoff of posthog-python and posthog-go. +const FEATURE_FLAGS_BACKOFF_BASE_MS: u64 = 300; +/// Ceiling of the `/flags` retry backoff. +const FEATURE_FLAGS_BACKOFF_MAX_MS: u64 = 30_000; + +fn feature_flags_backoff(attempt: u32) -> Duration { + exponential_backoff( + FEATURE_FLAGS_BACKOFF_BASE_MS, + FEATURE_FLAGS_BACKOFF_MAX_MS, + attempt, + ) +} + fn is_retryable_feature_flags_status(status: u16) -> bool { matches!(status, 502 | 504) } @@ -142,7 +160,7 @@ pub(crate) fn feature_flags_after_response( if is_retryable_feature_flags_status(status) && attempt <= opts.feature_flags_request_max_retries { - return FeatureFlagsResponseStep::Backoff(backoff_duration(opts, attempt, None)); + return FeatureFlagsResponseStep::Backoff(feature_flags_backoff(attempt)); } FeatureFlagsResponseStep::Done } @@ -159,7 +177,7 @@ pub(crate) fn feature_flags_after_transport_error( if !retryable || attempt > opts.feature_flags_request_max_retries { return FeatureFlagsTransportStep::Fail(Error::Connection(err_msg)); } - FeatureFlagsTransportStep::Backoff(backoff_duration(opts, attempt, None)) + FeatureFlagsTransportStep::Backoff(feature_flags_backoff(attempt)) } #[cfg(test)] @@ -368,8 +386,12 @@ mod tests { } #[test] - fn backoff_exponential_growth() { - let opts = test_opts(); + fn backoff_exponential_growth_with_default_options() { + let opts = ClientOptionsBuilder::default() + .api_key("k".to_string()) + .build() + .unwrap(); + assert_eq!(opts.max_capture_attempts, 4); assert_eq!(backoff_duration(&opts, 1, None), Duration::from_millis(100)); assert_eq!(backoff_duration(&opts, 2, None), Duration::from_millis(200)); assert_eq!(backoff_duration(&opts, 3, None), Duration::from_millis(400)); @@ -408,6 +430,21 @@ mod tests { /// off-by-one: the first retry must wait exactly `retry_initial_backoff_ms` /// (100ms here), not double it. + #[test] + fn feature_flags_backoff_doubles_from_300ms_up_to_30s() { + for (attempt, expected_ms) in [(1, 300), (2, 600), (3, 1_200), (7, 19_200), (8, 30_000)] { + assert_eq!( + feature_flags_backoff(attempt), + Duration::from_millis(expected_ms), + "attempt={attempt}" + ); + } + assert_eq!( + feature_flags_backoff(u32::MAX), + Duration::from_millis(30_000) + ); + } + #[test] fn feature_flags_http_status_uses_retry_budget_for_502_and_504_only() { let opts = ClientOptionsBuilder::default() @@ -421,7 +458,7 @@ mod tests { for status in [502, 504] { assert!(matches!( feature_flags_after_response(&opts, 1, status), - FeatureFlagsResponseStep::Backoff(_) + FeatureFlagsResponseStep::Backoff(d) if d == Duration::from_millis(300) )); assert!(matches!( feature_flags_after_response(&opts, 2, status), @@ -448,7 +485,7 @@ mod tests { assert!(matches!( feature_flags_after_transport_error(&opts, 1, true, "reset".into()), - FeatureFlagsTransportStep::Backoff(_) + FeatureFlagsTransportStep::Backoff(d) if d == Duration::from_millis(300) )); assert!(matches!( feature_flags_after_transport_error(&opts, 2, true, "reset".into()), diff --git a/src/client/summary.rs b/src/client/summary.rs index ccd0cd47..41081940 100644 --- a/src/client/summary.rs +++ b/src/client/summary.rs @@ -35,7 +35,10 @@ impl CaptureSummary { Self { submitted, results } } - /// Number of events sent on the wire (after `before_send` filtering). + /// Number of events that reached a verdict (after `before_send` filtering): + /// the events sent on the wire, plus any AI event the SDK did not send + /// because it was over the size limit. Such an event has a local `drop` + /// verdict in [`event_results`](Self::event_results). pub fn submitted(&self) -> usize { self.submitted } diff --git a/tests/test_capture_ai.rs b/tests/test_capture_ai.rs index 9dcc7f40..4c8d800c 100644 --- a/tests/test_capture_ai.rs +++ b/tests/test_capture_ai.rs @@ -13,7 +13,8 @@ use std::sync::{Arc, Mutex}; use httpmock::prelude::*; use posthog_rs::{ - CaptureCompression, ClientOptionsBuilder, Endpoint, Event, EventStatus, PostHogError, + CaptureCompression, CaptureSummary, ClientOptionsBuilder, Endpoint, Event, EventStatus, + PostHogError, }; use serde_json::json; use uuid::Uuid; @@ -139,6 +140,40 @@ fn three_two_mib_events() -> (Vec, Vec) { (uuids, events) } +/// One AI event over the 8 MiB `properties` ceiling and one normal AI event, +/// as `(oversize uuid, normal uuid, batch)`. +fn oversize_and_normal_events() -> (Uuid, Uuid, Vec) { + let (too_big, normal) = (Uuid::now_v7(), Uuid::now_v7()); + let mut oversize = sized_ai_event(8 * MIB + 1); + oversize.set_uuid(too_big); + let events = vec![oversize, ai_event_with_uuid("$ai_generation", normal)]; + (too_big, normal, events) +} + +/// The AI path accepting `normal` only from a body that leaves out `too_big`. +fn ai_mock_without<'a>(server: &'a MockServer, too_big: Uuid, normal: Uuid) -> httpmock::Mock<'a> { + server.mock(move |when, then| { + when.method(POST).path(AI_PATH).is_true(move |req| { + let body = zstd_body(req); + body.contains(&normal.to_string()) && !body.contains(&too_big.to_string()) + }); + then.status(200) + .header("content-type", "application/json") + .json_body(json!({ "results": { normal.to_string(): { "result": "ok" } } })); + }) +} + +/// The oversize event has a local `drop` verdict and counts as submitted but +/// not persisted, next to the normal event's `ok`. +fn assert_local_too_big_drop(summary: &CaptureSummary, too_big: Uuid, normal: Uuid) { + assert_eq!(summary.submitted(), 2); + assert_eq!(summary.not_persisted(), 1); + let verdict = &summary.event_results()[&too_big]; + assert_eq!(verdict.result, EventStatus::Drop); + assert_eq!(verdict.details.as_deref(), Some("ai_event_too_big")); + assert_eq!(summary.event_results()[&normal].result, EventStatus::Ok); +} + fn all_ok(uuids: &[Uuid]) -> serde_json::Value { json!(uuids .iter() @@ -337,32 +372,17 @@ mod async_client { } #[tokio::test] - async fn capture_ai_immediate_returns_server_verdicts_including_too_big() { + async fn capture_ai_batch_immediate_drops_an_oversize_event_locally() { let server = MockServer::start(); - let too_big = Uuid::now_v7(); - let ai = server.mock(|when, then| { - when.method(POST) - .path(AI_PATH) - .header("content-encoding", "zstd"); - then.status(200) - .header("content-type", "application/json") - .json_body(json!({ "results": { - too_big.to_string(): { "result": "drop", "details": "ai_event_too_big" } - }})); - }); + let (too_big, normal, events) = oversize_and_normal_events(); + let ai = ai_mock_without(&server, too_big, normal); let client = client(&mut options(server.base_url())).await; - // Over the background lane's local ceiling, yet sent: the immediate path - // leaves the size verdict to the backend and returns it. - let mut event = sized_ai_event(8 * MIB + 1); - event.set_uuid(too_big); - let summary = client.capture_ai_immediate(event).await.unwrap(); + let summary = client + .capture_ai_batch_immediate(events, false) + .await + .unwrap(); ai.assert_calls(1); - assert_eq!(summary.submitted(), 1); - assert_eq!(summary.not_persisted(), 1); - assert_eq!( - summary.event_results()[&too_big].details.as_deref(), - Some("ai_event_too_big") - ); + assert_local_too_big_drop(&summary, too_big, normal); } #[tokio::test] @@ -566,30 +586,14 @@ mod blocking { } #[test] - fn capture_ai_immediate_returns_server_verdicts_including_too_big() { + fn capture_ai_batch_immediate_drops_an_oversize_event_locally() { let server = MockServer::start(); - let too_big = Uuid::now_v7(); - let ai = server.mock(|when, then| { - when.method(POST) - .path(AI_PATH) - .header("content-encoding", "zstd"); - then.status(200) - .header("content-type", "application/json") - .json_body(json!({ "results": { - too_big.to_string(): { "result": "drop", "details": "ai_event_too_big" } - }})); - }); + let (too_big, normal, events) = oversize_and_normal_events(); + let ai = ai_mock_without(&server, too_big, normal); let client = client(&mut options(server.base_url())); - let mut event = sized_ai_event(8 * MIB + 1); - event.set_uuid(too_big); - let summary = client.capture_ai_immediate(event).unwrap(); + let summary = client.capture_ai_batch_immediate(events, false).unwrap(); ai.assert_calls(1); - assert_eq!(summary.submitted(), 1); - assert_eq!(summary.not_persisted(), 1); - assert_eq!( - summary.event_results()[&too_big].details.as_deref(), - Some("ai_event_too_big") - ); + assert_local_too_big_drop(&summary, too_big, normal); } #[test]