diff --git a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs index bcb5d50a2..fce125dce 100644 --- a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs @@ -184,6 +184,7 @@ impl ComposioProvider for GmailProvider { } } -// Cap/date-floor math lives in the shared `super::super::helpers` module -// (`ItemCap`, `pages_for_max_items`, `epoch_floor_from_depth`) so every provider -// shares one implementation — see that module for the unit tests. +// The `max_items` cap math (`ItemCap`) lives in the orchestrator now that it is +// the sole consumer; the `sync_depth_days` date floor (`epoch_floor_from_depth`) +// stays in `super::super::helpers` because `gmail::source` builds an +// `after:` filter from it. diff --git a/src/openhuman/memory_sync/composio/providers/helpers.rs b/src/openhuman/memory_sync/composio/providers/helpers.rs index d326689e0..92c3ccf9b 100644 --- a/src/openhuman/memory_sync/composio/providers/helpers.rs +++ b/src/openhuman/memory_sync/composio/providers/helpers.rs @@ -46,21 +46,13 @@ pub(crate) fn merge_extra(args: &mut serde_json::Value, extra: &serde_json::Valu } } -// ── Cap enforcement helpers ────────────────────────────────────────────── - -/// Compute the number of API pages needed to cover `max_items` at `page_size` -/// items per page, rounding up. -/// -/// Returns `u32::MAX` when `page_size == 0` to avoid division by zero; -/// callers should treat this as "no page cap beyond the provider's own upper -/// bound". -pub(crate) fn pages_for_max_items(max_items: u32, page_size: u32) -> u32 { - if page_size == 0 { - return u32::MAX; - } - // Widen to u64 before the addition to prevent overflow for large cap values. - (((max_items as u64) + (page_size as u64) - 1) / (page_size as u64)).min(u32::MAX as u64) as u32 -} +// ── Window helper ──────────────────────────────────────────────────────── +// +// The per-sync `max_items` cap math (`ItemCap` + `pages_for_max_items`) lives +// in the orchestrator now that it is the sole consumer — see +// [`super::orchestrator`]. `epoch_floor_from_depth` stays here because it is a +// provider-facing window helper (gmail/source.rs builds an `after:` +// filter from it), not orchestrator-internal cap math. /// Compute the Unix epoch timestamp (seconds) for `sync_depth_days` days ago. /// Used to build after-date filters (e.g. Gmail `after:`) on first sync. @@ -70,86 +62,10 @@ pub(crate) fn epoch_floor_from_depth(sync_depth_days: u32) -> i64 { floor.timestamp() } -/// Single source of truth for the per-sync `max_items` cap. -/// -/// Every Composio provider used to open-code three near-identical blocks — a -/// page-count cap, a mid-page clamp, and a post-page hard stop — which is how -/// the same off-by-a-page bug ended up in five providers and was missed in a -/// sixth. Funnelling all of them through this one type keeps the rule in one -/// place: construct it from `ctx.max_items`, derive the page cap, clamp each -/// batch (or check per item), and stop once the cap is reached. -/// -/// `None` cap means "no item limit beyond the provider's own internal page -/// ceiling" (e.g. after the user clicks "All In"). -#[derive(Debug, Clone, Copy)] -pub(crate) struct ItemCap { - cap: Option, - persisted: usize, -} - -impl ItemCap { - /// Build from a source's `max_items` value (`None` = uncapped). - pub(crate) fn new(max_items: Option) -> Self { - Self { - cap: max_items.map(|n| n as usize), - persisted: 0, - } - } - - /// The page ceiling to actually fetch: the smaller of the provider's own - /// `fallback` (e.g. `MAX_PAGES_PER_SYNC`) and the pages needed to cover the - /// cap. Uncapped → `fallback` unchanged. - pub(crate) fn max_pages(&self, page_size: u32, fallback: u32) -> u32 { - match self.cap { - Some(cap) => pages_for_max_items(cap as u32, page_size).min(fallback), - None => fallback, - } - } - - /// How many more items may still be persisted. `None` = unlimited. - pub(crate) fn remaining(&self) -> Option { - self.cap.map(|cap| cap.saturating_sub(self.persisted)) - } - - /// True once the cap is set and reached — callers `break` their pagination. - pub(crate) fn is_reached(&self) -> bool { - matches!(self.remaining(), Some(0)) - } - - /// Record `n` newly-persisted items against the budget. - pub(crate) fn record(&mut self, n: usize) { - self.persisted = self.persisted.saturating_add(n); - } - - /// Truncate a to-ingest batch down to the remaining budget, so a single - /// page larger than the cap never over-persists. No-op when uncapped. - pub(crate) fn clamp_batch(&self, batch: &mut Vec) { - if let Some(remaining) = self.remaining() { - if batch.len() > remaining { - batch.truncate(remaining); - } - } - } -} - #[cfg(test)] -mod cap_helper_tests { +mod window_helper_tests { use super::*; - #[test] - fn pages_for_max_items_rounds_up() { - assert_eq!(pages_for_max_items(100, 25), 4); - assert_eq!(pages_for_max_items(101, 25), 5); - assert_eq!(pages_for_max_items(1, 25), 1); - assert_eq!(pages_for_max_items(50, 50), 1); - assert_eq!(pages_for_max_items(51, 50), 2); - } - - #[test] - fn pages_for_max_items_zero_page_size() { - assert_eq!(pages_for_max_items(100, 0), u32::MAX); - } - #[test] fn epoch_floor_from_depth_is_in_the_past() { let floor = epoch_floor_from_depth(30); @@ -161,61 +77,6 @@ mod cap_helper_tests { "expected ~30 days in past, got {diff_days}" ); } - - #[test] - fn item_cap_uncapped_is_never_reached() { - let mut cap = ItemCap::new(None); - assert_eq!(cap.remaining(), None); - assert!(!cap.is_reached()); - cap.record(1_000_000); - assert!(!cap.is_reached()); - assert_eq!( - cap.max_pages(25, 20), - 20, - "uncapped keeps the provider fallback" - ); - } - - #[test] - fn item_cap_tracks_remaining_and_reached() { - let mut cap = ItemCap::new(Some(3)); - assert_eq!(cap.remaining(), Some(3)); - assert!(!cap.is_reached()); - cap.record(2); - assert_eq!(cap.remaining(), Some(1)); - assert!(!cap.is_reached()); - cap.record(5); // saturates, never underflows - assert_eq!(cap.remaining(), Some(0)); - assert!(cap.is_reached()); - } - - #[test] - fn item_cap_max_pages_is_min_of_fallback_and_needed() { - // cap=2, page_size=50 → 1 page needed, well under the fallback. - assert_eq!(ItemCap::new(Some(2)).max_pages(50, 20), 1); - // cap=1000, page_size=25 → 40 pages needed, clamped to fallback 20. - assert_eq!(ItemCap::new(Some(1000)).max_pages(25, 20), 20); - } - - #[test] - fn item_cap_clamp_batch_truncates_to_remaining() { - let cap = ItemCap::new(Some(2)); - let mut batch = vec![1, 2, 3, 4, 5]; - cap.clamp_batch(&mut batch); - assert_eq!(batch, vec![1, 2]); - - // Uncapped leaves the batch untouched. - let mut full = vec![1, 2, 3]; - ItemCap::new(None).clamp_batch(&mut full); - assert_eq!(full, vec![1, 2, 3]); - - // After recording progress, clamp uses the reduced budget. - let mut cap2 = ItemCap::new(Some(5)); - cap2.record(3); - let mut batch2 = vec![1, 2, 3, 4]; - cap2.clamp_batch(&mut batch2); - assert_eq!(batch2, vec![1, 2], "only 2 of the 5 budget remained"); - } } /// Resolve the first array found among `array_paths` (dotted object diff --git a/src/openhuman/memory_sync/composio/providers/orchestrator.rs b/src/openhuman/memory_sync/composio/providers/orchestrator.rs index 312e31b36..227171bdc 100644 --- a/src/openhuman/memory_sync/composio/providers/orchestrator.rs +++ b/src/openhuman/memory_sync/composio/providers/orchestrator.rs @@ -8,11 +8,13 @@ //! [`SyncOutcome`]. The copy-paste is exactly how the page-granular cap bug //! (#3304) ended up in five providers and was missed in a sixth. //! -//! [`ItemCap`] (PR #3304) centralised the cap *math*; this module centralises -//! the *control flow*. A provider now implements only the slim -//! [`IncrementalSource`] primitives and rides [`run_sync`], inheriting the -//! `max_items` cap, the `sync_depth_days` window, dedup, cursor advance, and -//! budget enforcement for free. +//! [`ItemCap`] (PR #3304) centralised the cap *math*; this module owns it and +//! the *control flow*. Now that every provider rides [`run_sync`], `ItemCap` +//! lives here (orchestrator-private), not in the shared `helpers` module — the +//! cap rule exists in exactly one place. A provider implements only the slim +//! [`IncrementalSource`] primitives and inherits the `max_items` cap, the +//! `sync_depth_days` window, dedup, cursor advance, and budget enforcement for +//! free. //! //! ## Scopes: flat AND nested in one loop //! @@ -39,10 +41,86 @@ use async_trait::async_trait; use serde_json::{json, Value}; -use super::helpers::ItemCap; use super::sync_state::SyncState; use super::{ProviderContext, SyncOutcome, SyncReason}; +/// Compute the number of API pages needed to cover `max_items` at `page_size` +/// items per page, rounding up. +/// +/// Returns `u32::MAX` when `page_size == 0` to avoid division by zero; +/// callers should treat this as "no page cap beyond the provider's own upper +/// bound". +fn pages_for_max_items(max_items: u32, page_size: u32) -> u32 { + if page_size == 0 { + return u32::MAX; + } + // Widen to u64 before the addition to prevent overflow for large cap values. + (((max_items as u64) + (page_size as u64) - 1) / (page_size as u64)).min(u32::MAX as u64) as u32 +} + +/// Single source of truth for the per-sync `max_items` cap. +/// +/// Every Composio provider used to open-code three near-identical blocks — a +/// page-count cap, a mid-page clamp, and a post-page hard stop — which is how +/// the same off-by-a-page bug (#3304) ended up in five providers and was missed +/// in a sixth. Now that every provider rides [`run_sync`], this type and its +/// page math are **orchestrator-private**: the cap rule lives in exactly one +/// place — construct it from `ctx.max_items`, derive the page cap, clamp each +/// batch before ingest, and stop once the cap is reached. +/// +/// `None` cap means "no item limit beyond the provider's own internal page +/// ceiling" (e.g. after the user clicks "All In"). +#[derive(Debug, Clone, Copy)] +struct ItemCap { + cap: Option, + persisted: usize, +} + +impl ItemCap { + /// Build from a source's `max_items` value (`None` = uncapped). + fn new(max_items: Option) -> Self { + Self { + cap: max_items.map(|n| n as usize), + persisted: 0, + } + } + + /// The page ceiling to actually fetch: the smaller of the provider's own + /// `fallback` (e.g. `MAX_PAGES_PER_SYNC`) and the pages needed to cover the + /// cap. Uncapped → `fallback` unchanged. + fn max_pages(&self, page_size: u32, fallback: u32) -> u32 { + match self.cap { + Some(cap) => pages_for_max_items(cap as u32, page_size).min(fallback), + None => fallback, + } + } + + /// How many more items may still be persisted. `None` = unlimited. + fn remaining(&self) -> Option { + self.cap.map(|cap| cap.saturating_sub(self.persisted)) + } + + /// True once the cap is set and reached — callers `break` their pagination. + fn is_reached(&self) -> bool { + matches!(self.remaining(), Some(0)) + } + + /// Record `n` newly-persisted items against the budget. + fn record(&mut self, n: usize) { + self.persisted = self.persisted.saturating_add(n); + } + + /// Truncate a to-ingest batch down to the remaining budget, so a single + /// page larger than the cap never over-persists. No-op when uncapped. + fn clamp_batch(&self, batch: &mut Vec) { + if let Some(remaining) = self.remaining() { + if batch.len() > remaining { + batch.truncate(remaining); + } + } + } +} + /// A unit of work to iterate within one sync pass. /// /// Flat providers return a single [`SyncScope::flat`]; nested providers return @@ -1391,4 +1469,75 @@ mod tests { .expect_err("without tolerance a scope fetch error propagates"); assert!(err.contains("scope fetch failure"), "got: {err}"); } + + // ── ItemCap / pages_for_max_items (relocated from helpers.rs, #3902) ── + + #[test] + fn pages_for_max_items_rounds_up() { + assert_eq!(pages_for_max_items(100, 25), 4); + assert_eq!(pages_for_max_items(101, 25), 5); + assert_eq!(pages_for_max_items(1, 25), 1); + assert_eq!(pages_for_max_items(50, 50), 1); + assert_eq!(pages_for_max_items(51, 50), 2); + } + + #[test] + fn pages_for_max_items_zero_page_size() { + assert_eq!(pages_for_max_items(100, 0), u32::MAX); + } + + #[test] + fn item_cap_uncapped_is_never_reached() { + let mut cap = ItemCap::new(None); + assert_eq!(cap.remaining(), None); + assert!(!cap.is_reached()); + cap.record(1_000_000); + assert!(!cap.is_reached()); + assert_eq!( + cap.max_pages(25, 20), + 20, + "uncapped keeps the provider fallback" + ); + } + + #[test] + fn item_cap_tracks_remaining_and_reached() { + let mut cap = ItemCap::new(Some(3)); + assert_eq!(cap.remaining(), Some(3)); + assert!(!cap.is_reached()); + cap.record(2); + assert_eq!(cap.remaining(), Some(1)); + assert!(!cap.is_reached()); + cap.record(5); // saturates, never underflows + assert_eq!(cap.remaining(), Some(0)); + assert!(cap.is_reached()); + } + + #[test] + fn item_cap_max_pages_is_min_of_fallback_and_needed() { + // cap=2, page_size=50 → 1 page needed, well under the fallback. + assert_eq!(ItemCap::new(Some(2)).max_pages(50, 20), 1); + // cap=1000, page_size=25 → 40 pages needed, clamped to fallback 20. + assert_eq!(ItemCap::new(Some(1000)).max_pages(25, 20), 20); + } + + #[test] + fn item_cap_clamp_batch_truncates_to_remaining() { + let cap = ItemCap::new(Some(2)); + let mut batch = vec![1, 2, 3, 4, 5]; + cap.clamp_batch(&mut batch); + assert_eq!(batch, vec![1, 2]); + + // Uncapped leaves the batch untouched. + let mut full = vec![1, 2, 3]; + ItemCap::new(None).clamp_batch(&mut full); + assert_eq!(full, vec![1, 2, 3]); + + // After recording progress, clamp uses the reduced budget. + let mut cap2 = ItemCap::new(Some(5)); + cap2.record(3); + let mut batch2 = vec![1, 2, 3, 4]; + cap2.clamp_batch(&mut batch2); + assert_eq!(batch2, vec![1, 2], "only 2 of the 5 budget remained"); + } }