mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
perf(memory_sync): bounded-concurrency Notion/ClickUp/Linear ingest (#3322)
This commit is contained in:
mysma-9403
@@ -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<String>,
|
||||
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<String>,
|
||||
/// 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<usize>;
|
||||
}
|
||||
|
||||
/// 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<usize> {
|
||||
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<String>,
|
||||
) -> (Vec<PendingIngest<'a>>, bool) {
|
||||
let mut hit_cursor_boundary = false;
|
||||
let mut pending: Vec<PendingIngest> = 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<I: TaskIngestor + Sync>(
|
||||
ingestor: &I,
|
||||
pending: Vec<PendingIngest<'_>>,
|
||||
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::<Vec<_>>();
|
||||
|
||||
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<String>,
|
||||
}
|
||||
|
||||
impl CountingIngestor {
|
||||
fn new(fail_task: Option<&str>) -> Arc<Self> {
|
||||
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<usize> {
|
||||
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<Value> {
|
||||
(0..n).map(|i| json!({ "id": format!("t{i}") })).collect()
|
||||
}
|
||||
|
||||
fn make_pending(tasks: &[Value]) -> Vec<PendingIngest<'_>> {
|
||||
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<String> = 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"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String>,
|
||||
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<String>,
|
||||
/// 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<usize>;
|
||||
}
|
||||
|
||||
/// 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<usize> {
|
||||
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<String>,
|
||||
) -> (Vec<PendingIngest<'a>>, bool) {
|
||||
let mut hit_cursor_boundary = false;
|
||||
let mut pending: Vec<PendingIngest> = 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<I: IssueIngestor + Sync>(
|
||||
ingestor: &I,
|
||||
pending: Vec<PendingIngest<'_>>,
|
||||
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::<Vec<_>>();
|
||||
|
||||
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<String>,
|
||||
}
|
||||
|
||||
impl CountingIngestor {
|
||||
fn new(fail_issue: Option<&str>) -> Arc<Self> {
|
||||
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<usize> {
|
||||
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<Value> {
|
||||
(0..n).map(|i| json!({ "id": format!("i{i}") })).collect()
|
||||
}
|
||||
|
||||
fn make_pending(issues: &[Value]) -> Vec<PendingIngest<'_>> {
|
||||
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<String> = 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"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String> {
|
||||
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<String>,
|
||||
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<String>,
|
||||
/// 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<usize>;
|
||||
}
|
||||
|
||||
/// 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<usize> {
|
||||
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<String>,
|
||||
) -> (Vec<PendingIngest<'a>>, bool) {
|
||||
let mut hit_cursor_boundary = false;
|
||||
let mut pending: Vec<PendingIngest> = 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<I: PageIngestor + Sync>(
|
||||
ingestor: &I,
|
||||
pending: Vec<PendingIngest<'_>>,
|
||||
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::<Vec<_>>();
|
||||
|
||||
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<String>,
|
||||
}
|
||||
|
||||
impl CountingIngestor {
|
||||
fn new(fail_page: Option<&str>) -> Arc<Self> {
|
||||
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<usize> {
|
||||
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<Value> {
|
||||
(0..n).map(|i| json!({ "id": format!("p{i}") })).collect()
|
||||
}
|
||||
|
||||
fn make_pending<'a>(pages: &'a [Value]) -> Vec<PendingIngest<'a>> {
|
||||
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<String> = 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"));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user