From ffbeab105c3582a70cac3d8fb6762e48bf69bd6b Mon Sep 17 00:00:00 2001 From: mysma-9403 <64923976+mysma-9403@users.noreply.github.com> Date: Thu, 4 Jun 2026 07:08:56 +0200 Subject: [PATCH] perf(memory_sync): bounded-concurrency Notion/ClickUp/Linear ingest (#3322) --- .../composio/providers/clickup/provider.rs | 398 +++++++++++++---- .../composio/providers/linear/provider.rs | 392 ++++++++++++++--- .../composio/providers/notion/provider.rs | 401 ++++++++++++++---- 3 files changed, 977 insertions(+), 214 deletions(-) diff --git a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs index 5c3c3c5cb..548ed4bf2 100644 --- a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs @@ -25,10 +25,12 @@ //! and avoids accidentally ingesting other teammates' private tasks. use async_trait::async_trait; -use serde_json::json; +use futures::StreamExt; +use serde_json::{json, Value}; use super::{ingest::ingest_task_into_memory_tree, sync}; -use crate::openhuman::memory_sync::composio::providers::sync_state::SyncState; +use crate::openhuman::config::Config; +use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ first_array_str, merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter, TaskKind, @@ -57,6 +59,10 @@ const MAX_PAGES_PER_WORKSPACE: u32 = 20; /// the upstream payload under `data`, so we check both shapes. const TASK_ID_PATHS: &[&str] = &["id", "data.id", "task_id", "data.task_id"]; +/// Max in-flight ingests per page. DB writes serialize anyway and the +/// cloud embedder has rate limits, so keep this small. +const INGEST_CONCURRENCY: usize = 8; + pub struct ClickUpProvider; impl ClickUpProvider { @@ -319,80 +325,21 @@ impl ComposioProvider for ClickUpProvider { break; } - // ── Per-item dedup + persist ──────────────────────── - let mut hit_cursor_boundary = false; - for task in &tasks { - let Some(task_id) = - crate::openhuman::memory_sync::composio::providers::sync_state::extract_item_id( - task, - TASK_ID_PATHS, - ) - else { - tracing::debug!("[composio:clickup] task missing ID, skipping"); - continue; - }; + // ── Per-item dedup + bounded-concurrency ingest ───── + let (pending, hit_cursor_boundary) = + select_pending(&tasks, &state, &mut newest_updated); - let updated = sync::extract_task_updated(task); - - // Track newest `date_updated` for cursor advancement. - if let Some(ref ts) = updated { - if newest_updated.as_ref().is_none_or(|existing| ts > existing) { - newest_updated = Some(ts.clone()); - } - } - - // Use a composite (task_id, date_updated) key so that - // a task edited *after* its last sync is re-persisted. - // Same trick the Notion provider uses for - // `last_edited_time`. - let sync_key = match &updated { - Some(ts) => format!("{task_id}@{ts}"), - None => task_id.clone(), - }; - - // If `date_updated` is at or older than our cursor - // *and* we've already synced this composite key, the - // rest of the page is by definition older too — we - // can stop this workspace early. - if let (Some(ref cursor), Some(ref ts)) = (&state.cursor, &updated) { - if ts <= cursor && state.is_synced(&sync_key) { - hit_cursor_boundary = true; - continue; - } - } - - if state.is_synced(&sync_key) { - continue; - } - - let title_text = sync::extract_task_name(task) - .unwrap_or_else(|| format!("ClickUp task {task_id}")); - let title = format!("ClickUp: {title_text}"); - - match ingest_task_into_memory_tree( - &ctx.config, - &connection_id, - &task_id, - &title, - updated.as_deref(), - task, - ) - .await - { - Ok(_) => { - state.mark_synced(&sync_key); - total_persisted += 1; - } - Err(e) => { - tracing::warn!( - task_id = %task_id, - workspace_id = %workspace_id, - error = %e, - "[composio:clickup] ingest failed (continuing)" - ); - } - } + let ingestor = MemoryTreeIngestor { + config: ctx.config.as_ref(), + connection_id: &connection_id, + }; + let outcome = + ingest_pending_buffered(&ingestor, pending, workspace_id, INGEST_CONCURRENCY) + .await; + for key in &outcome.synced_keys { + state.mark_synced(key); } + total_persisted += outcome.persisted; if hit_cursor_boundary { tracing::debug!( @@ -668,3 +615,306 @@ impl ClickUpProvider { Ok(sync::extract_workspace_ids(&resp.data)) } } + +/// One task that passed dedupe in pass 1 and is queued for concurrent +/// ingest in pass 2. Borrows the raw task `Value` out of the current +/// page's `tasks` (same scope — no clone needed). +struct PendingIngest<'a> { + sync_key: String, + task_id: String, + title: String, + updated: Option, + task: &'a Value, +} + +/// Folded result of [`ingest_pending_buffered`]. Both fields are +/// order-independent, so the concurrent stage can accumulate into it +/// regardless of the order ingests complete. +#[derive(Default)] +struct BufferedIngestOutcome { + /// `sync_key`s whose ingest succeeded — the caller marks each synced. + synced_keys: Vec, + /// Number of tasks persisted (equals `synced_keys.len()`). + persisted: usize, +} + +/// Seam over "ingest one ClickUp task", so the bounded-concurrency driver +/// can be unit-tested with a fake that records peak in-flight calls +/// without a real memory tree or embedder. +#[async_trait] +trait TaskIngestor { + async fn ingest( + &self, + task_id: &str, + title: &str, + updated: Option<&str>, + task: &Value, + ) -> anyhow::Result; +} + +/// Production ingestor: routes into the memory-tree pipeline via +/// [`ingest_task_into_memory_tree`]. +struct MemoryTreeIngestor<'c> { + config: &'c Config, + connection_id: &'c str, +} + +#[async_trait] +impl TaskIngestor for MemoryTreeIngestor<'_> { + async fn ingest( + &self, + task_id: &str, + title: &str, + updated: Option<&str>, + task: &Value, + ) -> anyhow::Result { + ingest_task_into_memory_tree( + self.config, + self.connection_id, + task_id, + title, + updated, + task, + ) + .await + } +} + +/// Pass 1 (pure, no I/O): scan one page of `tasks`, advance +/// `newest_updated`, skip already-synced items, and collect the tasks +/// still needing ingest. Returns the queued items and whether we crossed +/// the persistent cursor boundary (the signal to stop paginating this +/// workspace). All order-dependent decisions (cursor/timestamp) live here +/// — never in the concurrent stage. +fn select_pending<'a>( + tasks: &'a [Value], + state: &SyncState, + newest_updated: &mut Option, +) -> (Vec>, bool) { + let mut hit_cursor_boundary = false; + let mut pending: Vec = Vec::new(); + for task in tasks { + let Some(task_id) = extract_item_id(task, TASK_ID_PATHS) else { + tracing::debug!("[composio:clickup] task missing ID, skipping"); + continue; + }; + + let updated = sync::extract_task_updated(task); + + // Track newest `date_updated` for cursor advancement. + if let Some(ref ts) = updated { + if newest_updated.as_ref().is_none_or(|existing| ts > existing) { + *newest_updated = Some(ts.clone()); + } + } + + // Composite (task_id, date_updated) key so a task edited *after* + // its last sync is re-persisted. Same trick Notion uses for + // `last_edited_time`. + let sync_key = match &updated { + Some(ts) => format!("{task_id}@{ts}"), + None => task_id.clone(), + }; + + // Older than cursor AND already synced → caught up. + if let (Some(ref cursor), Some(ref ts)) = (&state.cursor, &updated) { + if ts <= cursor && state.is_synced(&sync_key) { + hit_cursor_boundary = true; + continue; + } + } + + if state.is_synced(&sync_key) { + continue; + } + + let title_text = + sync::extract_task_name(task).unwrap_or_else(|| format!("ClickUp task {task_id}")); + let title = format!("ClickUp: {title_text}"); + + pending.push(PendingIngest { + sync_key, + task_id, + title, + updated, + task, + }); + } + (pending, hit_cursor_boundary) +} + +/// Pass 2: ingest the queued tasks with bounded concurrency. Overlaps the +/// per-item embedding RTT (`buffer_unordered`, up to `concurrency` in +/// flight) and folds results into an order-independent +/// [`BufferedIngestOutcome`]. Unordered is correct here: nothing +/// downstream depends on completion order — successes are keyed by +/// `sync_key`. A failed ingest is logged and skipped (parity with the +/// previous sequential `continue`); ClickUp advances its cursor regardless +/// of per-item failures. +async fn ingest_pending_buffered( + ingestor: &I, + pending: Vec>, + workspace_id: &str, + concurrency: usize, +) -> BufferedIngestOutcome { + // Materialize the per-item futures into a Vec before `buffer_unordered` + // so the spawned sync future keeps concrete lifetimes / `Send`. + let ingest_futs = pending + .into_iter() + .map(|p| async move { + let res = ingestor + .ingest(&p.task_id, &p.title, p.updated.as_deref(), p.task) + .await; + (p.sync_key, p.task_id, res) + }) + .collect::>(); + + let mut outcome = BufferedIngestOutcome::default(); + let mut ingest_stream = futures::stream::iter(ingest_futs).buffer_unordered(concurrency); + while let Some((sync_key, task_id, res)) = ingest_stream.next().await { + match res { + Ok(_chunks_written) => { + outcome.synced_keys.push(sync_key); + outcome.persisted += 1; + } + Err(e) => { + tracing::warn!( + task_id = %task_id, + workspace_id = %workspace_id, + error = %e, + "[composio:clickup] ingest failed (continuing)" + ); + } + } + } + outcome +} + +#[cfg(test)] +mod buffered_tests { + use super::*; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// Fake ingestor: records the peak number of concurrent in-flight + /// `ingest` calls and can be told to fail one specific `task_id`. No + /// memory tree or embedder involved — lets us assert the concurrency + /// bound and overlap deterministically. + struct CountingIngestor { + in_flight: AtomicUsize, + peak: AtomicUsize, + fail_task: Option, + } + + impl CountingIngestor { + fn new(fail_task: Option<&str>) -> Arc { + Arc::new(Self { + in_flight: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + fail_task: fail_task.map(str::to_string), + }) + } + } + + #[async_trait] + impl TaskIngestor for CountingIngestor { + async fn ingest( + &self, + task_id: &str, + _title: &str, + _updated: Option<&str>, + _task: &Value, + ) -> anyhow::Result { + let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(now, Ordering::SeqCst); + // Yield a few times so futures genuinely interleave and the + // peak counter reflects real overlap, not accidental serial run. + for _ in 0..4 { + tokio::task::yield_now().await; + } + self.in_flight.fetch_sub(1, Ordering::SeqCst); + if self.fail_task.as_deref() == Some(task_id) { + Err(anyhow::anyhow!("forced failure for {task_id}")) + } else { + Ok(1) + } + } + } + + fn make_tasks(n: usize) -> Vec { + (0..n).map(|i| json!({ "id": format!("t{i}") })).collect() + } + + fn make_pending(tasks: &[Value]) -> Vec> { + tasks + .iter() + .enumerate() + .map(|(i, task)| PendingIngest { + sync_key: format!("k{i}"), + task_id: format!("t{i}"), + title: format!("ClickUp: task {i}"), + updated: None, + task, + }) + .collect() + } + + #[tokio::test] + async fn ingest_pending_buffered_bounds_and_overlaps() { + let ingestor = CountingIngestor::new(None); + let tasks = make_tasks(20); + let pending = make_pending(&tasks); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, "ws1", 8).await; + + assert_eq!(outcome.persisted, 20, "all tasks persisted"); + assert_eq!(outcome.synced_keys.len(), 20); + + let peak = ingestor.peak.load(Ordering::SeqCst); + assert!(peak <= 8, "peak in-flight {peak} exceeded the bound of 8"); + assert!(peak >= 2, "peak in-flight {peak} shows no real overlap"); + } + + #[tokio::test] + async fn ingest_pending_buffered_skips_failures_order_independent() { + let ingestor = CountingIngestor::new(Some("t2")); + let tasks = make_tasks(5); + let pending = make_pending(&tasks); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, "ws1", 4).await; + + assert_eq!(outcome.persisted, 4, "the one failed ingest is not counted"); + assert_eq!(outcome.synced_keys.len(), 4); + assert!( + !outcome.synced_keys.contains(&"k2".to_string()), + "the failed task's sync_key must not be marked synced" + ); + } + + #[test] + fn select_pending_tracks_newest_skips_synced_and_detects_boundary() { + let mut state = SyncState::new("clickup", "conn1"); + state.cursor = Some("2026-04-15T00:00:00Z".to_string()); + // Task B is already synced and older than the cursor. + state.mark_synced("b@2026-04-01T00:00:00Z"); + + let tasks = vec![ + json!({ "id": "a", "date_updated": "2026-05-01T00:00:00Z" }), + json!({ "id": "b", "date_updated": "2026-04-01T00:00:00Z" }), + json!({ "date_updated": "2026-03-01T00:00:00Z" }), // no id → skipped + ]; + + let mut newest: Option = None; + let (pending, hit_boundary) = select_pending(&tasks, &state, &mut newest); + + assert_eq!(pending.len(), 1, "only the new task A is queued"); + assert_eq!(pending[0].task_id, "a"); + assert_eq!(pending[0].sync_key, "a@2026-05-01T00:00:00Z"); + assert!( + hit_boundary, + "older synced task B trips the cursor boundary" + ); + assert_eq!(newest.as_deref(), Some("2026-05-01T00:00:00Z")); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/linear/provider.rs b/src/openhuman/memory_sync/composio/providers/linear/provider.rs index b43e1c60e..292c12838 100644 --- a/src/openhuman/memory_sync/composio/providers/linear/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/linear/provider.rs @@ -19,9 +19,11 @@ //! and avoids accidentally ingesting other teammates' private issues. use async_trait::async_trait; -use serde_json::json; +use futures::StreamExt; +use serde_json::{json, Value}; use super::{ingest::ingest_issue_into_memory_tree, sync}; +use crate::openhuman::config::Config; use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, @@ -45,6 +47,10 @@ const MAX_PAGES_PER_SYNC: u32 = 20; /// Paths for extracting a Linear issue's unique ID. const ISSUE_ID_PATHS: &[&str] = &["id", "data.id", "identifier", "data.identifier"]; +/// Max in-flight ingests per page. DB writes serialize anyway and the +/// cloud embedder has rate limits, so keep this small. +const INGEST_CONCURRENCY: usize = 8; + pub struct LinearProvider; impl LinearProvider { @@ -248,70 +254,23 @@ impl ComposioProvider for LinearProvider { break; } - // ── Per-item dedup + persist ───────────────────────────── - for issue in &issues { - let Some(issue_id) = extract_item_id(issue, ISSUE_ID_PATHS) else { - tracing::debug!("[composio:linear] issue missing ID, skipping"); - continue; - }; + // ── Per-item dedup + bounded-concurrency ingest ────────── + let (pending, page_hit_boundary) = select_pending(&issues, &state, &mut newest_updated); + if page_hit_boundary { + hit_cursor_boundary = true; + } - let updated = sync::extract_issue_updated(issue); - - // Track newest `updatedAt` for cursor advancement. - if let Some(ref ts) = updated { - if newest_updated.as_ref().is_none_or(|existing| ts > existing) { - newest_updated = Some(ts.clone()); - } - } - - // Composite (issue_id, updatedAt) key so re-edited - // issues are re-persisted on the next sync. - let sync_key = match &updated { - Some(ts) => format!("{issue_id}@{ts}"), - None => issue_id.clone(), - }; - - // If `updatedAt` is at or older than our cursor *and* - // we already synced this key, the rest of the page is - // by definition older — stop early. - if let (Some(ref cursor), Some(ref ts)) = (&state.cursor, &updated) { - if ts <= cursor && state.is_synced(&sync_key) { - hit_cursor_boundary = true; - continue; - } - } - - if state.is_synced(&sync_key) { - continue; - } - - let title_text = sync::extract_issue_title(issue) - .unwrap_or_else(|| format!("Linear issue {issue_id}")); - let title = format!("Linear: {title_text}"); - - match ingest_issue_into_memory_tree( - &ctx.config, - &connection_id, - &issue_id, - &title, - updated.as_deref(), - issue, - ) - .await - { - Ok(_) => { - state.mark_synced(&sync_key); - total_persisted += 1; - } - Err(e) => { - had_persist_failures = true; - tracing::warn!( - issue_id = %issue_id, - error = %e, - "[composio:linear] failed to ingest issue into memory_tree (continuing)" - ); - } - } + let ingestor = MemoryTreeIngestor { + config: ctx.config.as_ref(), + connection_id: &connection_id, + }; + let outcome = ingest_pending_buffered(&ingestor, pending, INGEST_CONCURRENCY).await; + for key in &outcome.synced_keys { + state.mark_synced(key); + } + total_persisted += outcome.persisted; + if outcome.had_failures { + had_persist_failures = true; } if hit_cursor_boundary { @@ -541,3 +500,308 @@ impl LinearProvider { }) } } + +/// One issue that passed dedupe in pass 1 and is queued for concurrent +/// ingest in pass 2. Borrows the raw issue `Value` out of the current +/// page's `issues` (same scope — no clone needed). +struct PendingIngest<'a> { + sync_key: String, + issue_id: String, + title: String, + updated: Option, + issue: &'a Value, +} + +/// Folded result of [`ingest_pending_buffered`]. Every field is +/// order-independent, so the concurrent stage can accumulate into it +/// regardless of the order ingests complete. +#[derive(Default)] +struct BufferedIngestOutcome { + /// `sync_key`s whose ingest succeeded — the caller marks each synced. + synced_keys: Vec, + /// Number of issues persisted (equals `synced_keys.len()`). + persisted: usize, + /// Whether any per-item ingest failed (the caller holds the cursor). + had_failures: bool, +} + +/// Seam over "ingest one Linear issue", so the bounded-concurrency driver +/// can be unit-tested with a fake that records peak in-flight calls +/// without a real memory tree or embedder. +#[async_trait] +trait IssueIngestor { + async fn ingest( + &self, + issue_id: &str, + title: &str, + updated: Option<&str>, + issue: &Value, + ) -> anyhow::Result; +} + +/// Production ingestor: routes into the memory-tree pipeline via +/// [`ingest_issue_into_memory_tree`]. +struct MemoryTreeIngestor<'c> { + config: &'c Config, + connection_id: &'c str, +} + +#[async_trait] +impl IssueIngestor for MemoryTreeIngestor<'_> { + async fn ingest( + &self, + issue_id: &str, + title: &str, + updated: Option<&str>, + issue: &Value, + ) -> anyhow::Result { + ingest_issue_into_memory_tree( + self.config, + self.connection_id, + issue_id, + title, + updated, + issue, + ) + .await + } +} + +/// Pass 1 (pure, no I/O): scan one page of `issues`, advance +/// `newest_updated`, skip already-synced items, and collect the issues +/// still needing ingest. Returns the queued items and whether we crossed +/// the persistent cursor boundary (the signal to stop paginating). All +/// order-dependent decisions (cursor/timestamp) live here — never in the +/// concurrent stage. +fn select_pending<'a>( + issues: &'a [Value], + state: &SyncState, + newest_updated: &mut Option, +) -> (Vec>, bool) { + let mut hit_cursor_boundary = false; + let mut pending: Vec = Vec::new(); + for issue in issues { + let Some(issue_id) = extract_item_id(issue, ISSUE_ID_PATHS) else { + tracing::debug!("[composio:linear] issue missing ID, skipping"); + continue; + }; + + let updated = sync::extract_issue_updated(issue); + + // Track newest `updatedAt` for cursor advancement. + if let Some(ref ts) = updated { + if newest_updated.as_ref().is_none_or(|existing| ts > existing) { + *newest_updated = Some(ts.clone()); + } + } + + // Composite (issue_id, updatedAt) key so re-edited issues are + // re-persisted on the next sync. + let sync_key = match &updated { + Some(ts) => format!("{issue_id}@{ts}"), + None => issue_id.clone(), + }; + + // Older than cursor AND already synced → caught up. + if let (Some(ref cursor), Some(ref ts)) = (&state.cursor, &updated) { + if ts <= cursor && state.is_synced(&sync_key) { + hit_cursor_boundary = true; + continue; + } + } + + if state.is_synced(&sync_key) { + continue; + } + + let title_text = + sync::extract_issue_title(issue).unwrap_or_else(|| format!("Linear issue {issue_id}")); + let title = format!("Linear: {title_text}"); + + pending.push(PendingIngest { + sync_key, + issue_id, + title, + updated, + issue, + }); + } + (pending, hit_cursor_boundary) +} + +/// Pass 2: ingest the queued issues with bounded concurrency. Overlaps +/// the per-item embedding RTT (`buffer_unordered`, up to `concurrency` in +/// flight) and folds results into an order-independent +/// [`BufferedIngestOutcome`]. Unordered is correct here: nothing +/// downstream depends on completion order — successes are keyed by +/// `sync_key`. A failed ingest is logged and skipped, tripping +/// `had_failures` so the caller holds the cursor (parity with the +/// previous sequential path). +async fn ingest_pending_buffered( + ingestor: &I, + pending: Vec>, + concurrency: usize, +) -> BufferedIngestOutcome { + // Materialize the per-item futures into a Vec before `buffer_unordered` + // so the spawned sync future keeps concrete lifetimes / `Send`. + let ingest_futs = pending + .into_iter() + .map(|p| async move { + let res = ingestor + .ingest(&p.issue_id, &p.title, p.updated.as_deref(), p.issue) + .await; + (p.sync_key, p.issue_id, res) + }) + .collect::>(); + + let mut outcome = BufferedIngestOutcome::default(); + let mut ingest_stream = futures::stream::iter(ingest_futs).buffer_unordered(concurrency); + while let Some((sync_key, issue_id, res)) = ingest_stream.next().await { + match res { + Ok(_chunks_written) => { + outcome.synced_keys.push(sync_key); + outcome.persisted += 1; + } + Err(e) => { + outcome.had_failures = true; + tracing::warn!( + issue_id = %issue_id, + error = %e, + "[composio:linear] failed to ingest issue into memory_tree (continuing)" + ); + } + } + } + outcome +} + +#[cfg(test)] +mod buffered_tests { + use super::*; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// Fake ingestor: records the peak number of concurrent in-flight + /// `ingest` calls and can be told to fail one specific `issue_id`. No + /// memory tree or embedder involved — lets us assert the concurrency + /// bound and overlap deterministically. + struct CountingIngestor { + in_flight: AtomicUsize, + peak: AtomicUsize, + fail_issue: Option, + } + + impl CountingIngestor { + fn new(fail_issue: Option<&str>) -> Arc { + Arc::new(Self { + in_flight: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + fail_issue: fail_issue.map(str::to_string), + }) + } + } + + #[async_trait] + impl IssueIngestor for CountingIngestor { + async fn ingest( + &self, + issue_id: &str, + _title: &str, + _updated: Option<&str>, + _issue: &Value, + ) -> anyhow::Result { + let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(now, Ordering::SeqCst); + // Yield a few times so futures genuinely interleave and the + // peak counter reflects real overlap, not accidental serial run. + for _ in 0..4 { + tokio::task::yield_now().await; + } + self.in_flight.fetch_sub(1, Ordering::SeqCst); + if self.fail_issue.as_deref() == Some(issue_id) { + Err(anyhow::anyhow!("forced failure for {issue_id}")) + } else { + Ok(1) + } + } + } + + fn make_issues(n: usize) -> Vec { + (0..n).map(|i| json!({ "id": format!("i{i}") })).collect() + } + + fn make_pending(issues: &[Value]) -> Vec> { + issues + .iter() + .enumerate() + .map(|(i, issue)| PendingIngest { + sync_key: format!("k{i}"), + issue_id: format!("i{i}"), + title: format!("Linear: issue {i}"), + updated: None, + issue, + }) + .collect() + } + + #[tokio::test] + async fn ingest_pending_buffered_bounds_and_overlaps() { + let ingestor = CountingIngestor::new(None); + let issues = make_issues(20); + let pending = make_pending(&issues); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 8).await; + + assert_eq!(outcome.persisted, 20, "all issues persisted"); + assert_eq!(outcome.synced_keys.len(), 20); + assert!(!outcome.had_failures); + + let peak = ingestor.peak.load(Ordering::SeqCst); + assert!(peak <= 8, "peak in-flight {peak} exceeded the bound of 8"); + assert!(peak >= 2, "peak in-flight {peak} shows no real overlap"); + } + + #[tokio::test] + async fn ingest_pending_buffered_skips_failures_order_independent() { + let ingestor = CountingIngestor::new(Some("i2")); + let issues = make_issues(5); + let pending = make_pending(&issues); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 4).await; + + assert_eq!(outcome.persisted, 4, "the one failed ingest is not counted"); + assert!(outcome.had_failures); + assert_eq!(outcome.synced_keys.len(), 4); + assert!( + !outcome.synced_keys.contains(&"k2".to_string()), + "the failed issue's sync_key must not be marked synced" + ); + } + + #[test] + fn select_pending_tracks_newest_skips_synced_and_detects_boundary() { + let mut state = SyncState::new("linear", "conn1"); + state.cursor = Some("2026-04-15T00:00:00Z".to_string()); + // Issue B is already synced and older than the cursor. + state.mark_synced("b@2026-04-01T00:00:00Z"); + + let issues = vec![ + json!({ "id": "a", "updatedAt": "2026-05-01T00:00:00Z" }), + json!({ "id": "b", "updatedAt": "2026-04-01T00:00:00Z" }), + json!({ "updatedAt": "2026-03-01T00:00:00Z" }), // no id → skipped + ]; + + let mut newest: Option = None; + let (pending, hit_boundary) = select_pending(&issues, &state, &mut newest); + + assert_eq!(pending.len(), 1, "only the new issue A is queued"); + assert_eq!(pending[0].issue_id, "a"); + assert_eq!(pending[0].sync_key, "a@2026-05-01T00:00:00Z"); + assert!( + hit_boundary, + "older synced issue B trips the cursor boundary" + ); + assert_eq!(newest.as_deref(), Some("2026-05-01T00:00:00Z")); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/notion/provider.rs b/src/openhuman/memory_sync/composio/providers/notion/provider.rs index 6acd21705..d44832166 100644 --- a/src/openhuman/memory_sync/composio/providers/notion/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/notion/provider.rs @@ -20,12 +20,14 @@ use serde_json::{json, Value}; use super::ingest::ingest_page_into_memory_tree; use super::sync; +use crate::openhuman::config::Config; use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ first_array_str, merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskContainer, TaskFetchFilter, TaskKind, }; +use futures::StreamExt; pub(crate) const ACTION_GET_ABOUT_ME: &str = "NOTION_GET_ABOUT_ME"; pub(crate) const ACTION_FETCH_DATA: &str = "NOTION_FETCH_DATA"; @@ -52,6 +54,10 @@ const PAGE_EDITED_PATHS: &[&str] = &[ "data.lastEditedTime", ]; +/// Max in-flight ingests per page. DB writes serialize anyway and the +/// cloud embedder has rate limits, so keep this small. +const INGEST_CONCURRENCY: usize = 8; + pub struct NotionProvider; impl NotionProvider { @@ -242,83 +248,22 @@ impl ComposioProvider for NotionProvider { break; } - // ── Step 4: deduplicate and persist per-item ──────────── - let mut hit_cursor_boundary = false; - for page in &results { - let Some(page_id) = extract_item_id(page, PAGE_ID_PATHS) else { - tracing::debug!("[composio:notion] page missing ID, skipping"); - continue; - }; + // ── Step 4a: dedupe + decide which pages to ingest ────── + let (pending, hit_cursor_boundary) = + select_pending(&results, &state, &mut newest_edited_time); - let edited_time = extract_item_id(page, PAGE_EDITED_PATHS); - - // Track the newest edited time for cursor advancement. - if let Some(ref et) = edited_time { - if newest_edited_time - .as_ref() - .is_none_or(|existing| et > existing) - { - newest_edited_time = Some(et.clone()); - } - } - - // For Notion, a page can be *edited* after we last synced - // it. We use a composite key of page_id + edited_time to - // detect this: if the page_id is in synced_ids but the - // edited_time is newer than the cursor, we re-sync it. - let sync_key = match &edited_time { - Some(et) => format!("{page_id}@{et}"), - None => page_id.clone(), - }; - - // If the page's edited time is older than our cursor, - // we've caught up — everything beyond is already synced. - if let (Some(ref cursor), Some(ref et)) = (&state.cursor, &edited_time) { - if et <= cursor && state.is_synced(&sync_key) { - hit_cursor_boundary = true; - continue; - } - } - - if state.is_synced(&sync_key) { - continue; - } - - // Build a title from the page's properties. - let title_text = sync::extract_page_title(page) - .unwrap_or_else(|| format!("Notion page {page_id}")); - let title = format!("Notion: {title_text}"); - - // Route into the memory-tree pipeline (#2885). The prior - // implementation called `persist_single_item` → - // `MemoryClient::store_skill_sync` → UnifiedMemory - // `memory_docs`, which the modern retrieval surfaces - // (`memory.search`, `tree.read_chunk`, `tree.browse`, - // summary trees, MCP tools) don't read from — the data - // was invisible to every agent recall path. - match ingest_page_into_memory_tree( - &ctx.config, - &connection_id, - &page_id, - &title, - edited_time.as_deref(), - page, - ) - .await - { - Ok(_chunks_written) => { - state.mark_synced(&sync_key); - total_persisted += 1; - } - Err(e) => { - had_ingest_failures = true; - tracing::warn!( - page_id = %page_id, - error = %e, - "[composio:notion] failed to ingest page into memory_tree (continuing)" - ); - } - } + // ── Step 4b: ingest queued pages (bounded concurrency) ── + let ingestor = MemoryTreeIngestor { + config: ctx.config.as_ref(), + connection_id: &connection_id, + }; + let outcome = ingest_pending_buffered(&ingestor, pending, INGEST_CONCURRENCY).await; + for key in &outcome.synced_keys { + state.mark_synced(key); + } + total_persisted += outcome.persisted; + if outcome.had_failures { + had_ingest_failures = true; } if hit_cursor_boundary { @@ -656,3 +601,307 @@ fn extract_database_title(db: &serde_json::Value) -> Option { let text = text.trim(); (!text.is_empty()).then(|| text.to_string()) } + +/// One page that passed dedupe in pass 1 and is queued for concurrent +/// ingest in pass 2. Borrows the raw page `Value` out of the current +/// page's `results` (same scope — no clone needed). +struct PendingIngest<'a> { + sync_key: String, + page_id: String, + title: String, + edited_time: Option, + page: &'a Value, +} + +/// Folded result of [`ingest_pending_buffered`]. Every field is +/// order-independent, so the concurrent stage can accumulate into it +/// regardless of the order ingests complete. +#[derive(Default)] +struct BufferedIngestOutcome { + /// `sync_key`s whose ingest succeeded — the caller marks each synced. + synced_keys: Vec, + /// Number of pages persisted (equals `synced_keys.len()`). + persisted: usize, + /// Whether any per-item ingest failed (the caller holds the cursor). + had_failures: bool, +} + +/// Seam over "ingest one Notion page", so the bounded-concurrency driver +/// can be unit-tested with a fake that records peak in-flight calls +/// without a real memory tree or embedder. +#[async_trait] +trait PageIngestor { + async fn ingest( + &self, + page_id: &str, + title: &str, + edited_time: Option<&str>, + page: &Value, + ) -> anyhow::Result; +} + +/// Production ingestor: routes into the memory-tree pipeline (#2885) via +/// [`ingest_page_into_memory_tree`]. +struct MemoryTreeIngestor<'c> { + config: &'c Config, + connection_id: &'c str, +} + +#[async_trait] +impl PageIngestor for MemoryTreeIngestor<'_> { + async fn ingest( + &self, + page_id: &str, + title: &str, + edited_time: Option<&str>, + page: &Value, + ) -> anyhow::Result { + ingest_page_into_memory_tree( + self.config, + self.connection_id, + page_id, + title, + edited_time, + page, + ) + .await + } +} + +/// Pass 1 (pure, no I/O): scan one page of `results`, advance +/// `newest_edited_time`, skip already-synced items, and collect the +/// pages still needing ingest. Returns the queued items and whether we +/// crossed the persistent cursor boundary (the signal to stop +/// paginating). All order-dependent decisions (cursor/timestamp) live +/// here — never in the concurrent stage. +fn select_pending<'a>( + results: &'a [Value], + state: &SyncState, + newest_edited_time: &mut Option, +) -> (Vec>, bool) { + let mut hit_cursor_boundary = false; + let mut pending: Vec = Vec::new(); + for page in results { + let Some(page_id) = extract_item_id(page, PAGE_ID_PATHS) else { + tracing::debug!("[composio:notion] page missing ID, skipping"); + continue; + }; + + let edited_time = extract_item_id(page, PAGE_EDITED_PATHS); + + // Track the newest edited time for cursor advancement. + if let Some(ref et) = edited_time { + if newest_edited_time + .as_ref() + .is_none_or(|existing| et > existing) + { + *newest_edited_time = Some(et.clone()); + } + } + + let sync_key = match &edited_time { + Some(et) => format!("{page_id}@{et}"), + None => page_id.clone(), + }; + + // Older than cursor AND already synced → caught up. + if let (Some(ref cursor), Some(ref et)) = (&state.cursor, &edited_time) { + if et <= cursor && state.is_synced(&sync_key) { + hit_cursor_boundary = true; + continue; + } + } + + if state.is_synced(&sync_key) { + continue; + } + + let title_text = + sync::extract_page_title(page).unwrap_or_else(|| format!("Notion page {page_id}")); + let title = format!("Notion: {title_text}"); + + pending.push(PendingIngest { + sync_key, + page_id, + title, + edited_time, + page, + }); + } + (pending, hit_cursor_boundary) +} + +/// Pass 2: ingest the queued pages with bounded concurrency. Overlaps +/// the per-item embedding RTT (`buffer_unordered`, up to `concurrency` +/// in flight) and folds results into an order-independent +/// [`BufferedIngestOutcome`]. Unordered is correct here (unlike the +/// order-aligned GitHub reader): nothing downstream depends on +/// completion order — successes are keyed by `sync_key`. +async fn ingest_pending_buffered( + ingestor: &I, + pending: Vec>, + concurrency: usize, +) -> BufferedIngestOutcome { + // Materialize the per-item futures into a Vec before `buffer_unordered` + // so the spawned sync future keeps concrete lifetimes / `Send`. + let ingest_futs = pending + .into_iter() + .map(|p| async move { + let res = ingestor + .ingest(&p.page_id, &p.title, p.edited_time.as_deref(), p.page) + .await; + (p.sync_key, p.page_id, res) + }) + .collect::>(); + + let mut outcome = BufferedIngestOutcome::default(); + let mut ingest_stream = futures::stream::iter(ingest_futs).buffer_unordered(concurrency); + while let Some((sync_key, page_id, res)) = ingest_stream.next().await { + match res { + Ok(_chunks_written) => { + outcome.synced_keys.push(sync_key); + outcome.persisted += 1; + } + Err(e) => { + outcome.had_failures = true; + tracing::warn!( + page_id = %page_id, + error = %e, + "[composio:notion] failed to ingest page into memory_tree (continuing)" + ); + } + } + } + outcome +} + +#[cfg(test)] +mod buffered_tests { + use super::*; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// Fake ingestor: records the peak number of concurrent in-flight + /// `ingest` calls and can be told to fail one specific `page_id`. No + /// memory tree or embedder involved — lets us assert the concurrency + /// bound and overlap deterministically. + struct CountingIngestor { + in_flight: AtomicUsize, + peak: AtomicUsize, + fail_page: Option, + } + + impl CountingIngestor { + fn new(fail_page: Option<&str>) -> Arc { + Arc::new(Self { + in_flight: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + fail_page: fail_page.map(str::to_string), + }) + } + } + + #[async_trait] + impl PageIngestor for CountingIngestor { + async fn ingest( + &self, + page_id: &str, + _title: &str, + _edited_time: Option<&str>, + _page: &Value, + ) -> anyhow::Result { + let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(now, Ordering::SeqCst); + // Yield a few times so futures genuinely interleave and the + // peak counter reflects real overlap, not accidental serial run. + for _ in 0..4 { + tokio::task::yield_now().await; + } + self.in_flight.fetch_sub(1, Ordering::SeqCst); + if self.fail_page.as_deref() == Some(page_id) { + Err(anyhow::anyhow!("forced failure for {page_id}")) + } else { + Ok(1) + } + } + } + + fn make_pages(n: usize) -> Vec { + (0..n).map(|i| json!({ "id": format!("p{i}") })).collect() + } + + fn make_pending<'a>(pages: &'a [Value]) -> Vec> { + pages + .iter() + .enumerate() + .map(|(i, page)| PendingIngest { + sync_key: format!("k{i}"), + page_id: format!("p{i}"), + title: format!("Notion: page {i}"), + edited_time: None, + page, + }) + .collect() + } + + #[tokio::test] + async fn ingest_pending_buffered_bounds_and_overlaps() { + let ingestor = CountingIngestor::new(None); + let pages = make_pages(20); + let pending = make_pending(&pages); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 8).await; + + assert_eq!(outcome.persisted, 20, "all pages persisted"); + assert_eq!(outcome.synced_keys.len(), 20); + assert!(!outcome.had_failures); + + let peak = ingestor.peak.load(Ordering::SeqCst); + assert!(peak <= 8, "peak in-flight {peak} exceeded the bound of 8"); + assert!(peak >= 2, "peak in-flight {peak} shows no real overlap"); + } + + #[tokio::test] + async fn ingest_pending_buffered_skips_failures_order_independent() { + let ingestor = CountingIngestor::new(Some("p2")); + let pages = make_pages(5); + let pending = make_pending(&pages); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 4).await; + + assert_eq!(outcome.persisted, 4, "the one failed ingest is not counted"); + assert!(outcome.had_failures); + assert_eq!(outcome.synced_keys.len(), 4); + assert!( + !outcome.synced_keys.contains(&"k2".to_string()), + "the failed page's sync_key must not be marked synced" + ); + } + + #[test] + fn select_pending_tracks_newest_skips_synced_and_detects_boundary() { + let mut state = SyncState::new("notion", "conn1"); + state.cursor = Some("2026-04-15T00:00:00Z".to_string()); + // Page B is already synced and older than the cursor. + state.mark_synced("b@2026-04-01T00:00:00Z"); + + let pages = vec![ + json!({ "id": "a", "last_edited_time": "2026-05-01T00:00:00Z" }), + json!({ "id": "b", "last_edited_time": "2026-04-01T00:00:00Z" }), + json!({ "last_edited_time": "2026-03-01T00:00:00Z" }), // no id → skipped + ]; + + let mut newest: Option = None; + let (pending, hit_boundary) = select_pending(&pages, &state, &mut newest); + + assert_eq!(pending.len(), 1, "only the new page A is queued"); + assert_eq!(pending[0].page_id, "a"); + assert_eq!(pending[0].sync_key, "a@2026-05-01T00:00:00Z"); + assert!( + hit_boundary, + "older synced page B trips the cursor boundary" + ); + assert_eq!(newest.as_deref(), Some("2026-05-01T00:00:00Z")); + } +}