diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c306e53..f801a97 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -152,8 +152,16 @@ jobs: # deliberately can still be built by someone else's dependency. `--depth 2` # covers every pair without the combinatorial blow-up of the full set. # - # Check-only: this is about whether the combinations *compile*, and the - # behaviour of each is the main job's business. + # Check-only: this is about whether the combinations *compile*, and + # whether they link and run is the `feature-configs` job's business. + # + # This is also where §E2's `contacts` row is covered, without a step of + # its own: `contacts` is a feature of `tinymemory-core`, and the powerset + # enumerates it as a subset of size one on this same runner. What it does + # not cover is *executing* it, which needs `macos-latest` — the feature's + # only behaviour is a CNContactStore reader whose dependencies sit behind + # a `cfg(target_os = "macos")` table, so on ubuntu there is nothing to run + # but the empty stub. That runner is deliberately not spent. - name: Feature powerset compiles run: cargo hack --feature-powerset --depth 2 --workspace check --all-targets @@ -166,6 +174,63 @@ jobs: cargo llvm-cov --all-features --workspace --summary-only \ | tee "$GITHUB_STEP_SUMMARY" + # §E2's first half: build **and test** each engine configuration on its own. + # + # What this adds over the powerset pass, precisely: `cargo check` never + # links, and it never runs a test binary. A feature set that type-checks can + # still fail to link — the root `Cargo.toml` documents one such hazard, where + # a second crate claiming `links = "git2"` becomes a hard cargo error — and + # that failure is invisible to a check. So these rows are worth their minutes + # for linking and running, not for behaviour that varies by feature: the + # facade's own suite is the same set of tests in every configuration, because + # `DriverRegistry` admission is a static policy table rather than a function + # of which adapters were compiled in. + # + # Two of §E2's nine configurations name features the facade does not have. + # `--features contacts` belongs to `tinymemory-core` and is covered by the + # powerset job above. `--features sync-composio` names a feature that exists + # nowhere in the workspace: the Composio sync is unconditional in + # `tinymemory-core`, so there is nothing to select and nothing to isolate. + # Recorded here rather than quietly dropped, because a missing row in a + # matrix reads as covered. + feature-configs: + name: Test ${{ matrix.name }} + runs-on: ubuntu-latest + strategy: + # Every configuration is independent, and knowing that three of them + # broke is worth more than stopping at the first. + fail-fast: false + matrix: + include: + - name: no default features + features: --no-default-features + - name: tinycortex + features: --features tinycortex + - name: tinycortex and memory-git + features: --features tinycortex,memory-git + - name: mem0 + features: --features mem0 + - name: supermemory + features: --features supermemory + - name: cognee + features: --features cognee + - name: all features + features: --all-features + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + submodules: recursive + + - uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + + # Scoped to the facade: these are *its* features, and it is what a host + # compiles against. An engine's own suite runs in the main job. + - name: Test + run: cargo test -p tinymemory ${{ matrix.features }} + # The module crate is its own workspace root (see the `exclude` note in the # root Cargo.toml), so NONE of the steps above touch it: `--all-targets`, # `--all-features` and `--workspace` all stop at the workspace boundary and diff --git a/core/src/diff/source.rs b/core/src/diff/source.rs index 149ef34..7f99147 100644 --- a/core/src/diff/source.rs +++ b/core/src/diff/source.rs @@ -14,10 +14,14 @@ //! //! The crate calls `items_for_source(source_id)` with the *logical* source id, //! but the host chunk `source_id LIKE` prefix is kind-dependent — Composio -//! sources key their chunks by `:%`, not `mem_src::%`, and the -//! toolkit is not derivable from the logical id alone. The adapter is therefore -//! built from the full [`MemorySourceEntry`] list (which carries `toolkit`) and -//! resolves each id → prefix up front. +//! sources key their chunks by `::%`, not +//! `mem_src::%`, and neither is derivable from the logical id alone. The +//! adapter is therefore built from the full [`MemorySourceEntry`] list (which +//! carries both) and resolves each id → prefix up front, through +//! `sources::status::source_id_prefix` — the one definition of the scheme, +//! shared so a snapshot and a status can never disagree about which chunks +//! belong to a source. Named rather than linked: it is `pub(crate)`, and a +//! link to it from this module's public documentation is a rustdoc error. use std::collections::HashMap; use std::sync::Arc; @@ -27,7 +31,8 @@ use crate::engine::backend::diff::{extract_item_id, SnapshotItem, SnapshotItemSo #[cfg(test)] use tinymemory_api::host::test_support::TestHostConfig; -use crate::sources::types::{MemorySourceEntry, SourceKind}; +use crate::sources::status::source_id_prefix; +use crate::sources::types::MemorySourceEntry; use crate::Config; /// Host [`SnapshotItemSource`] backed by `mem_tree_chunks`. @@ -126,22 +131,10 @@ impl SnapshotItemSource for ChunkStoreItemSource { } } -/// Build the `source_id LIKE` prefix that matches chunks belonging to a source. -/// Mirrors `memory_sources::status::source_id_prefix`. -pub(crate) fn source_id_prefix(source: &MemorySourceEntry) -> String { - match source.kind { - SourceKind::Composio => source - .toolkit - .as_deref() - .map(|t| format!("{t}:%")) - .unwrap_or_else(|| "__no_toolkit__:%".to_string()), - _ => format!("mem_src:{}:%", source.id), - } -} - #[cfg(test)] mod tests { use super::*; + use crate::sources::types::SourceKind; fn folder_source(id: &str) -> MemorySourceEntry { MemorySourceEntry { @@ -169,30 +162,21 @@ mod tests { } } + /// The prefix scheme itself is covered where it is defined + /// (`sources::status::source_id_prefix_dispatch`). What matters here is + /// that the adapter resolves through *that* definition, so a snapshot and + /// a status agree on which chunks belong to a source. #[test] - fn source_id_prefix_folder() { + fn the_adapter_resolves_prefixes_through_the_shared_definition() { + let source = folder_source("src_abc"); + let adapter = + ChunkStoreItemSource::single(std::sync::Arc::new(TestHostConfig::default()), &source); assert_eq!( - source_id_prefix(&folder_source("src_abc")), - "mem_src:src_abc:%" + adapter.prefixes.get("src_abc").map(String::as_str), + Some(crate::sources::status::source_id_prefix(&source).as_str()) ); } - #[test] - fn source_id_prefix_composio() { - let mut entry = folder_source("src_cmp"); - entry.kind = SourceKind::Composio; - entry.toolkit = Some("gmail".into()); - assert_eq!(source_id_prefix(&entry), "gmail:%"); - } - - #[test] - fn source_id_prefix_composio_without_toolkit() { - let mut entry = folder_source("src_cmp"); - entry.kind = SourceKind::Composio; - entry.toolkit = None; - assert_eq!(source_id_prefix(&entry), "__no_toolkit__:%"); - } - #[test] fn read_only_adapter_never_yields_items() { let source = ChunkStoreItemSource::read_only( diff --git a/core/src/sources/status.rs b/core/src/sources/status.rs index f2568f5..6621d98 100644 --- a/core/src/sources/status.rs +++ b/core/src/sources/status.rs @@ -3,9 +3,23 @@ //! Queries `mem_tree_chunks` filtered by source-id prefix: //! - Reader-backed kinds (folder/github/rss/web/twitter) tag chunks //! with `mem_src:{source.id}:%`, so we count those directly. -//! - Composio sources tag chunks with the toolkit-specific id -//! (e.g. `gmail:user@example.com:msg_xxx`), so we match by toolkit +//! - Composio sources tag chunks with the connector id +//! (`{toolkit}:{connection_id}:{document_id}`), so we match by that //! prefix instead. +//! +//! # Where "pending" lives +//! +//! Not on `mem_tree_chunks`. That table carries a legacy `embedding` column, +//! added by an idempotent migration and written by nothing — counting +//! `embedding IS NULL` reports every chunk as pending forever, so a healthy +//! source shows `chunks_pending == chunks_synced` and the memory-sources UI +//! shows eternal work in flight. +//! +//! Embeddings live in the `mem_tree_chunk_embeddings` sidecar, one row per +//! `(chunk, model signature)`. A chunk with no row there is still not +//! necessarily pending: the lifecycle may have dropped it, or it may be +//! recorded in `mem_tree_chunk_reembed_skipped`. Both are terminal, and both +//! count as resolved. use serde::Serialize; @@ -13,31 +27,14 @@ use crate::sources::types::{MemorySourceEntry, SourceKind}; use crate::store::chunks::store::with_connection; use crate::Config; -#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum FreshnessLabel { - Active, - Recent, - Idle, -} - -impl FreshnessLabel { - pub fn from_age_ms(last_ms: Option, now_ms: i64) -> Self { - match last_ms { - None => Self::Idle, - Some(ts) => { - let age = now_ms.saturating_sub(ts); - if age <= 30_000 { - Self::Active - } else if age <= 5 * 60_000 { - Self::Recent - } else { - Self::Idle - } - } - } - } -} +/// Freshness is one vocabulary, owned by [`crate::sync::sync_status`]. +/// +/// It was declared a second time here, with the same variants, the same +/// snake_case wire strings and the same thresholds — two definitions that had +/// to be kept in step by hand and nothing checking that they were. Re-exported +/// rather than merely imported, so `sources::status::FreshnessLabel` stays a +/// working path for callers that already name it. +pub use crate::sync::sync_status::FreshnessLabel; #[derive(Clone, Debug, Serialize)] pub struct SourceStatus { @@ -62,13 +59,27 @@ pub async fn source_status( // Surface real query errors so status telemetry doesn't lie about // a healthy zero-row state when the DB is actually broken. + // + // "Pending" is "not resolved", and a chunk resolves three ways: + // it has an embedding, it was dropped by the lifecycle, or it was + // deliberately skipped for re-embedding. This is the engine's own + // predicate from `list_sync_statuses`, kept identical so the + // per-source view and the per-provider one cannot disagree about + // the same chunk. let (synced, pending, last_ts): (i64, i64, Option) = conn.query_row( "SELECT \ COUNT(*), \ - SUM(CASE WHEN embedding IS NULL THEN 1 ELSE 0 END), \ - MAX(timestamp_ms) \ - FROM mem_tree_chunks \ - WHERE source_id LIKE ?1", + SUM(CASE WHEN EXISTS ( \ + SELECT 1 FROM mem_tree_chunk_embeddings e \ + WHERE e.chunk_id = c.id) \ + OR c.lifecycle_status = 'dropped' \ + OR EXISTS ( \ + SELECT 1 FROM mem_tree_chunk_reembed_skipped s \ + WHERE s.chunk_id = c.id) \ + THEN 0 ELSE 1 END), \ + MAX(c.timestamp_ms) \ + FROM mem_tree_chunks c \ + WHERE c.source_id LIKE ?1", [&prefix], |r| { Ok(( @@ -121,16 +132,30 @@ pub async fn status_list(config: &Config) -> Result, String> { } /// Build the `source_id LIKE` prefix that matches chunks belonging to a source. -fn source_id_prefix(source: &MemorySourceEntry) -> String { +/// +/// The scheme is set by the ingest paths, not chosen here: reader-backed kinds +/// key chunks `mem_src:{source.id}:{item}`, and the Composio sync keys them +/// `{toolkit}:{connection_id}:{document_id}`. +/// +/// Matching a Composio source on its toolkit alone would sweep in every *other* +/// connection of that toolkit — two Gmail accounts would each report the +/// other's chunks as their own — so the connection narrows it. A Composio entry +/// without a connection id does not pass validation; the toolkit-only fallback +/// is there so a malformed row degrades to a wide match rather than to no +/// match at all. +/// +/// Shared with [`crate::diff::source`], which builds its snapshot item source +/// from the same prefixes. It held a second copy of this function whose comment +/// said it mirrored this one, which is a mirror only for as long as someone +/// remembers it is. +pub(crate) fn source_id_prefix(source: &MemorySourceEntry) -> String { match source.kind { SourceKind::Composio => { - // Composio providers write chunks with source_id = `{toolkit}:%` - // (e.g. `gmail:user@example.com:msg_xxx`). Match by toolkit only. - source - .toolkit - .as_deref() - .map(|t| format!("{t}:%")) - .unwrap_or_else(|| "__no_toolkit__:%".to_string()) + match (source.toolkit.as_deref(), source.connection_id.as_deref()) { + (Some(toolkit), Some(connection_id)) => format!("{toolkit}:{connection_id}:%"), + (Some(toolkit), None) => format!("{toolkit}:%"), + (None, _) => "__no_toolkit__:%".to_string(), + } } _ => format!("mem_src:{}:%", source.id), } @@ -140,28 +165,10 @@ fn source_id_prefix(source: &MemorySourceEntry) -> String { mod tests { use super::*; - #[test] - fn freshness_thresholds() { - let now = 1_000_000_000_000; - assert_eq!( - FreshnessLabel::from_age_ms(Some(now - 1_000), now), - FreshnessLabel::Active - ); - assert_eq!( - FreshnessLabel::from_age_ms(Some(now - 60_000), now), - FreshnessLabel::Recent - ); - assert_eq!( - FreshnessLabel::from_age_ms(Some(now - 600_000), now), - FreshnessLabel::Idle - ); - assert_eq!(FreshnessLabel::from_age_ms(None, now), FreshnessLabel::Idle); - } - - #[test] - fn source_id_prefix_dispatch() { - let mut entry = MemorySourceEntry { - id: "src_abc".into(), + /// A folder source, the shape the prefix and status tests both start from. + fn folder_entry(id: &str) -> MemorySourceEntry { + MemorySourceEntry { + id: id.into(), kind: SourceKind::Folder, label: "x".into(), enabled: true, @@ -182,11 +189,136 @@ mod tests { max_tokens_per_sync: None, max_cost_per_sync_usd: None, sync_depth_days: None, - }; + } + } + + #[test] + fn source_id_prefix_dispatch() { + let mut entry = folder_entry("src_abc"); assert_eq!(source_id_prefix(&entry), "mem_src:src_abc:%"); + // A Composio source is matched on its connection, not just its + // toolkit: a second Gmail account must not count the first's chunks. entry.kind = SourceKind::Composio; entry.toolkit = Some("gmail".into()); + entry.connection_id = Some("conn-1".into()); + assert_eq!(source_id_prefix(&entry), "gmail:conn-1:%"); + + entry.connection_id = None; assert_eq!(source_id_prefix(&entry), "gmail:%"); + + entry.toolkit = None; + assert_eq!(source_id_prefix(&entry), "__no_toolkit__:%"); + } + + /// A chunk under `source_id`, with a deterministic id the test can address. + fn chunk(id: &str, source_id: &str) -> crate::store::chunks::types::Chunk { + use crate::store::chunks::types::{Chunk, Metadata, SourceKind as ChunkSourceKind}; + + let at = chrono::Utc::now(); + Chunk { + id: id.into(), + content: "content".into(), + metadata: Metadata::point_in_time(ChunkSourceKind::Document, source_id, "owner", at), + token_count: 1, + seq_in_source: 0, + created_at: at, + partial_message: false, + } + } + + /// The status query counted pending as `embedding IS NULL` over + /// `mem_tree_chunks`. That column is a legacy migration artefact nothing + /// writes, so every chunk read as pending and a healthy source reported + /// `chunks_pending == chunks_synced` forever. + /// + /// Pending is "not resolved", and a chunk resolves by carrying an + /// embedding, by being dropped, or by being recorded as skipped for + /// re-embedding. Only the first of these four is genuinely still in + /// flight. + #[tokio::test] + async fn pending_counts_unresolved_chunks_not_the_dead_embedding_column() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let mut host = TestHostConfig::default(); + host.workspace_dir = workspace.path().join("workspace"); + let config = host.to_arc(); + + let source = folder_entry("src_status"); + let chunks = [ + chunk("chunk-embedded", "mem_src:src_status:item-1"), + chunk("chunk-pending", "mem_src:src_status:item-2"), + chunk("chunk-dropped", "mem_src:src_status:item-3"), + chunk("chunk-skipped", "mem_src:src_status:item-4"), + ]; + crate::store::chunks::store::upsert_chunks(&*config, &chunks).expect("upsert chunks"); + + crate::store::chunks::store::set_chunk_embedding(&*config, "chunk-embedded", &[0.1, 0.2]) + .expect("set embedding"); + crate::store::chunks::store::set_chunk_lifecycle_status( + &*config, + "chunk-dropped", + crate::store::chunks::store::CHUNK_STATUS_DROPPED, + ) + .expect("set lifecycle status"); + crate::store::chunks::store::mark_chunk_reembed_skipped( + &*config, + "chunk-skipped", + "test-signature", + "too long", + ) + .expect("mark reembed skipped"); + + // Guard against a vacuous test: the legacy column must still be NULL + // for every row, so a pending count of 1 is attributable to the new + // predicate rather than to the old one happening to agree. + let legacy_nulls: i64 = crate::store::chunks::store::with_connection(&*config, |conn| { + Ok(conn.query_row( + "SELECT COUNT(*) FROM mem_tree_chunks \ + WHERE embedding IS NULL AND source_id LIKE 'mem_src:src_status:%'", + [], + |row| row.get(0), + )?) + }) + .expect("count legacy nulls"); + assert_eq!( + legacy_nulls, 4, + "nothing writes the legacy column, so counting it would report all four pending" + ); + + let status = source_status(&*config, &source) + .await + .expect("source status"); + assert_eq!(status.chunks_synced, 4); + assert_eq!( + status.chunks_pending, 1, + "only the chunk with no embedding, no drop and no skip is still in flight" + ); + assert!(status.last_chunk_at_ms.is_some()); + } + + /// A source with no chunks reports zeroes rather than failing on the + /// `NULL` a `SUM` over no rows produces. + #[tokio::test] + async fn a_source_with_no_chunks_reports_zeroes() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let mut host = TestHostConfig::default(); + host.workspace_dir = workspace.path().join("workspace"); + let config = host.to_arc(); + + let status = source_status(&*config, &folder_entry("src_empty")) + .await + .expect("source status"); + assert_eq!(status.chunks_synced, 0); + assert_eq!(status.chunks_pending, 0); + assert_eq!(status.last_chunk_at_ms, None); + assert_eq!(status.freshness, FreshnessLabel::Idle); } } diff --git a/core/src/sync/pipelines/host.rs b/core/src/sync/pipelines/host.rs index 1b8a403..56c013a 100644 --- a/core/src/sync/pipelines/host.rs +++ b/core/src/sync/pipelines/host.rs @@ -8,9 +8,11 @@ //! [`MemoryClient`](crate::store::MemoryClient), so whatever driver the host //! bound serves the sync. -use std::sync::Arc; +use std::collections::HashMap; +use std::sync::{Arc, Mutex, OnceLock}; use async_trait::async_trait; +use tokio::sync::OwnedMutexGuard; use crate::store::MemoryClientRef; use crate::sync::composio::providers::sync_state::SyncStateStore; @@ -145,25 +147,64 @@ impl SkillDocSink for PipelineHost { } } -/// Mirror of the engine adapter's tree reconnect: route the stored document -/// through core's ingest funnel under the same source id scheme. +/// Mirror of the engine adapter's tree reconnect (`engine::sync`'s +/// `ingest_document_into_memory_tree`): route the stored document through +/// core's ingest funnel under the same addressing scheme. +/// +/// # The scheme is the contract, not an implementation detail +/// +/// The tree scope a chunk seals under is `path_scope`, falling back to +/// `source_id`. Retrieval selects source trees by that scope and classifies +/// them by their **platform prefix** — `gmail:` is email, `slack:` is chat. +/// So the scope has to be `"{toolkit}:{connection_id}"`: one tree per +/// connection, named by a prefix retrieval knows. +/// +/// Passing no `path_scope` is not a smaller version of that. It makes each +/// item's own `source_id` the scope, which is a *tree per document*, named by +/// a prefix that matches no platform — the items are stored and then +/// unreachable, which is the #5473 defect this reconnect exists to fix. The +/// `source_id LIKE` prefix the memory-source status and diff snapshots query +/// by is keyed on this same scheme — see `sources::status::source_id_prefix`. +/// +/// Tags are a deliberate superset of the engine adapter's: it tags the toolkit +/// alone, this also tags `composio_sync`. Tags feed scoring and filtering, not +/// addressing, so the extra one costs nothing and marks the ingest path. async fn ingest_into_tree(config: &Config, document: &SkillDocument) -> anyhow::Result<()> { - let source_id = format!( - "composio:{}:{}:{}", - document.toolkit, document.connection_id, document.document_id - ); + let toolkit = document.toolkit.trim().to_ascii_lowercase(); + let connection_id = document.connection_id.trim(); + // A blank toolkit or connection would yield a scope with no platform + // prefix (`":conn"`, `"gmail:"`), which no retrieval kind matches; skip + // rather than write an unreachable tree. The skill store still holds the + // item. + if toolkit.is_empty() || connection_id.is_empty() { + tracing::debug!( + document_id = %document.document_id, + "[memory_sync] skipping memory-tree ingest: item has no toolkit/connection scope" + ); + return Ok(()); + } + let tree_scope = format!("{toolkit}:{connection_id}"); + let source_id = format!("{tree_scope}:{}", document.document_id); + let owner = format!("{toolkit}-sync:{connection_id}"); let doc = crate::ingest_pipeline::IngestDocumentInput { - provider: document.toolkit.clone(), + provider: format!("composio:{toolkit}"), title: document.title.clone(), body: document.content.clone(), modified_at: chrono::Utc::now(), - source_ref: Some(source_id.clone()), + source_ref: Some(document.document_id.clone()), }; - let tags = vec!["composio_sync".to_string(), document.toolkit.clone()]; - crate::ingest_pipeline::ingest_document(config, &source_id, "", tags, doc) - .await - .map(|_| ()) - .map_err(|error| anyhow::anyhow!("{error}")) + let tags = vec!["composio_sync".to_string(), toolkit]; + crate::ingest_pipeline::ingest_document_with_scope( + config, + &source_id, + &owner, + tags, + doc, + Some(tree_scope), + ) + .await + .map(|_| ()) + .map_err(|error| anyhow::anyhow!("memory-tree ingest failed for source `{source_id}`: {error}")) } #[async_trait] @@ -333,7 +374,14 @@ pub async fn run_composio_connection_with_caps( max_cost_per_sync_usd: caps.max_cost_per_sync_usd, }; let host = Arc::new(PipelineHost::new(memory, config.to_arc())); - run_pipeline(pipeline, &pipeline_config, &host.context()).await + run_pipeline( + pipeline, + toolkit, + connection_id, + &pipeline_config, + &host.context(), + ) + .await } /// Run a bounded Gmail backfill through the engine-free pipelines. @@ -353,7 +401,17 @@ pub async fn run_gmail_backfill( .with_query(query), ); let host = Arc::new(PipelineHost::new(memory, config.to_arc())); - run_pipeline(pipeline, &PipelineConfig::default(), &host.context()).await + // The backfill drives the Gmail pipeline, which keys its `SyncState` on + // `"gmail"`; naming the same toolkit here puts it behind the same guard as + // a periodic or RPC Gmail sync of this connection. + run_pipeline( + pipeline, + "gmail", + connection_id, + &PipelineConfig::default(), + &host.context(), + ) + .await } /// Run the Slack search backfill through the engine-free pipelines. @@ -372,14 +430,105 @@ pub async fn run_slack_search_backfill( backfill_days, )); let host = Arc::new(PipelineHost::new(memory, config.to_arc())); - run_pipeline(pipeline, &PipelineConfig::default(), &host.context()).await + // `SlackSearchBackfillPipeline` loads and saves the same + // `("slack", connection_id)` state the Slack sync pipeline does, so the two + // must share one guard or they clobber each other's cursor and budget. + run_pipeline( + pipeline, + "slack", + connection_id, + &PipelineConfig::default(), + &host.context(), + ) + .await +} + +/// The note a run carries when another run already holds its connection. +/// +/// Callers that distinguish "nothing to sync" from "did not sync" match on +/// this rather than on a message they would have to keep in step by hand. +pub const SYNC_ALREADY_RUNNING: &str = "sync already running for this connection"; + +/// One guard per connection, so two runs cannot clobber each other's state. +type ConnectionLock = Arc>; + +/// The process-wide guard table. +/// +/// `run_incremental_sync` loads the connection's `SyncState` once, mutates it +/// in memory for the whole run, and saves at the end; the Slack search +/// backfill does the same over the same `("slack", connection_id)` record. Two +/// runs of one connection therefore race on the cursor, the dedup set and the +/// daily budget, and whichever saves last wins — losing either the dedup set +/// (re-fetch, re-spend) or the budget count (overspend past the cap). The +/// periodic loop, the sync RPC and a trigger can each fire the same +/// connection, so the race is reachable as the code stands. +/// +/// This is the single-process answer, which is how the loop and the RPC paths +/// actually run. An optimistic version stamp on the KV record is what a +/// multi-process host would need instead. +/// +/// The table only ever grows, bounded by the number of connections the host +/// has seen — the same shape, and the same bound, as the periodic scheduler's +/// last-fired map. An entry is one `Arc` and an unlocked mutex. +fn connection_locks() -> &'static Mutex> { + static LOCKS: OnceLock>> = OnceLock::new(); + LOCKS.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// The guard key for a connection. +/// +/// Normalised exactly as [`build_composio_pipeline`] normalises the toolkit +/// gate, so `" Gmail "` and `gmail` name one connection rather than two — and +/// so the Slack sync pipeline and the Slack search backfill, which share one +/// `SyncState` record, share one guard. +fn connection_key(toolkit: &str, connection_id: &str) -> (String, String) { + ( + toolkit.trim().to_ascii_lowercase(), + connection_id.trim().to_owned(), + ) +} + +/// Take the guard for one connection, or `None` if a run already holds it. +/// +/// Deliberately non-blocking. Queueing behind the running sync would stall the +/// periodic loop's whole tick — it walks connections sequentially — and then +/// run a second sync of a connection that has just been synced, which is the +/// Composio spend this guard exists to avoid. +fn try_hold_connection(toolkit: &str, connection_id: &str) -> Option> { + let lock = { + // A panic inside a run cannot corrupt the table: it holds `Arc`s, and + // the async guard is released by its own `Drop`. Recovering from the + // poison keeps one panicking sync from disabling every later one. + let mut locks = connection_locks() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + Arc::clone( + locks + .entry(connection_key(toolkit, connection_id)) + .or_default(), + ) + }; + lock.try_lock_owned().ok() } async fn run_pipeline( pipeline: Arc, + toolkit: &str, + connection_id: &str, config: &PipelineConfig, context: &SyncContext, ) -> Result { + let Some(_connection) = try_hold_connection(toolkit, connection_id) else { + tracing::debug!( + toolkit, + connection_id, + "[memory_sync] a sync of this connection is already running; skipping" + ); + return Ok(SyncOutcome { + note: Some(SYNC_ALREADY_RUNNING.to_owned()), + ..SyncOutcome::default() + }); + }; let pipeline_id = pipeline.id().to_owned(); let mut dispatcher = SyncDispatcher::new(); dispatcher @@ -439,4 +588,269 @@ mod tests { ); } } + + /// The guard table is process-global and shared by every test in this + /// binary, so each test names connections nothing else touches. + #[test] + fn one_connection_admits_one_run_at_a_time() { + let held = + try_hold_connection("gmail", "guard-single").expect("the first run takes the guard"); + assert!( + try_hold_connection("gmail", "guard-single").is_none(), + "a second run of the same connection must be refused, not queued" + ); + drop(held); + assert!( + try_hold_connection("gmail", "guard-single").is_some(), + "the guard must be released when the run ends" + ); + } + + /// The guard is per connection, not per toolkit: one slow Gmail sync must + /// not stop every other Gmail connection from syncing. + #[test] + fn different_connections_hold_independent_guards() { + let first = try_hold_connection("gmail", "guard-independent-a") + .expect("the first connection takes its guard"); + let second = try_hold_connection("gmail", "guard-independent-b") + .expect("a different connection has its own guard"); + drop((first, second)); + } + + /// `build_composio_pipeline` accepts `" Gmail "` by normalising it. The + /// guard key must normalise identically, or a padded toolkit syncs the + /// same connection concurrently with an unpadded one and they clobber each + /// other's state — the defect the guard exists to prevent. + #[test] + fn the_guard_key_normalises_the_toolkit_like_the_gate() { + assert_eq!( + connection_key(" Gmail ", " conn-1 "), + connection_key("gmail", "conn-1") + ); + let held = try_hold_connection("gmail", "guard-normalised") + .expect("the first run takes the guard"); + assert!( + try_hold_connection(" GMAIL\t", "guard-normalised").is_none(), + "a padded, mixed-case toolkit names the same connection" + ); + drop(held); + } + + /// The Slack sync pipeline and the Slack search backfill load and save the + /// same `("slack", connection_id)` state, so they must contend. + #[test] + fn the_slack_backfill_shares_the_slack_sync_guard() { + assert_eq!( + connection_key("slack", "guard-slack"), + connection_key("Slack", "guard-slack") + ); + let held = try_hold_connection("slack", "guard-slack").expect("the sync takes the guard"); + assert!( + try_hold_connection("slack", "guard-slack").is_none(), + "the backfill must not run while a Slack sync of this connection is running" + ); + drop(held); + } + + /// The engine adapter's tree reconnect has this test + /// (`engine::sync`'s `composio_sync_document_reaches_memory_tree`); the + /// engine-free host that replaced it on the live path did not, and drifted + /// — it wrote a `composio:`-prefixed source id and no `path_scope`, so + /// every synced item became its own tree under a scope no platform prefix + /// matches. Chunks existed, recall could not reach them. Asserting the + /// addressing, not merely the row count, is what catches that. + #[tokio::test] + async fn a_synced_document_is_keyed_by_its_connection_scope() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let workspace_dir = workspace.path().join("workspace"); + let mut host_config = TestHostConfig::default(); + host_config.workspace_dir = workspace_dir.clone(); + let config = host_config.to_arc(); + let client: MemoryClientRef = Arc::new( + crate::store::MemoryClient::from_workspace_dir(workspace_dir) + .expect("memory client initialises against a fresh workspace"), + ); + let host = PipelineHost::new(client, config.clone()); + + // A fresh tree is empty, so a non-zero count after the store is + // attributable to this sync rather than to pre-existing state. + assert_eq!( + crate::store::chunks::store::count_chunks(&*config).expect("count chunks"), + 0, + "fresh workspace must start with an empty memory tree" + ); + + host.store(SkillDocument { + namespace_skill_id: "gmail".into(), + connection_id: "conn-1".into(), + document_id: "gmail:msg-1".into(), + title: "Quarterly planning".into(), + content: "Let's finalise the Q3 roadmap and align on the launch date.".into(), + toolkit: "gmail".into(), + metadata: serde_json::json!({ "source": "composio-provider-incremental" }), + }) + .await + .expect("storing a synced document must also ingest it into the memory tree"); + + let scoped = crate::store::chunks::store::list_chunks( + &*config, + &crate::store::chunks::store::ListChunksQuery { + source_id: Some("gmail:conn-1:gmail:msg-1".into()), + limit: Some(8), + ..Default::default() + }, + ) + .expect("list chunks by source id"); + assert!( + !scoped.is_empty(), + "ingested chunks must be keyed by `{{toolkit}}:{{connection_id}}:{{document_id}}` — \ + the scheme the memory-source status and diff snapshots query by" + ); + assert!( + scoped + .iter() + .all(|chunk| chunk.metadata.path_scope.as_deref() == Some("gmail:conn-1")), + "connector chunks must carry the `{{toolkit}}:{{connection_id}}` tree scope so \ + query_source resolves them (gmail → email)" + ); + assert!( + scoped + .iter() + .all(|chunk| chunk.metadata.owner == "gmail-sync:conn-1"), + "connector chunks must be owned by the connection that synced them" + ); + } + + /// A blank toolkit or connection cannot produce a scope any retrieval kind + /// matches, so the tree half is skipped rather than writing an unreachable + /// tree. The skill store, which committed first, still holds the item. + #[tokio::test] + async fn an_item_without_a_connection_scope_skips_the_tree_but_not_the_store() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let workspace_dir = workspace.path().join("workspace"); + let mut host_config = TestHostConfig::default(); + host_config.workspace_dir = workspace_dir.clone(); + let config = host_config.to_arc(); + let client: MemoryClientRef = Arc::new( + crate::store::MemoryClient::from_workspace_dir(workspace_dir) + .expect("memory client initialises against a fresh workspace"), + ); + let host = PipelineHost::new(client.clone(), config.clone()); + + host.store(SkillDocument { + namespace_skill_id: "gmail".into(), + connection_id: " ".into(), + document_id: "gmail:msg-2".into(), + title: "No connection".into(), + content: "This item has no connection scope.".into(), + toolkit: "gmail".into(), + metadata: serde_json::Value::Null, + }) + .await + .expect("a scopeless item must not fail the sync"); + + assert_eq!( + crate::store::chunks::store::count_chunks(&*config).expect("count chunks"), + 0, + "a scopeless item must not write a tree no retrieval can reach" + ); + let stored = client + .list_documents(Some("skill-gmail")) + .await + .expect("list skill documents"); + let documents = stored + .get("documents") + .and_then(|value| value.as_array()) + .cloned() + .unwrap_or_default(); + assert_eq!( + documents.len(), + 1, + "the skill store is the source of truth and must still hold the item" + ); + } + + /// A pipeline that records whether it was ticked, so the refusal path can + /// be shown to skip the run rather than to run and discard the result. + struct RecordingPipeline(Arc); + + #[async_trait] + impl SyncPipeline for RecordingPipeline { + fn id(&self) -> &str { + "test:recording" + } + + fn kind(&self) -> crate::sync::pipelines::traits::SyncPipelineKind { + crate::sync::pipelines::traits::SyncPipelineKind::Composio + } + + async fn init(&self, _: &PipelineConfig, _: &SyncContext) -> anyhow::Result<()> { + Ok(()) + } + + async fn tick(&self, _: &PipelineConfig, _: &SyncContext) -> anyhow::Result { + self.0.store(true, std::sync::atomic::Ordering::SeqCst); + Ok(SyncOutcome { + records_ingested: 7, + ..SyncOutcome::default() + }) + } + } + + /// End to end: with the connection held, `run_pipeline` returns the note + /// without ticking the pipeline — no fetch, no Composio spend, and no + /// second writer of the connection's `SyncState`. + #[tokio::test] + async fn a_held_connection_short_circuits_the_run() { + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let client: MemoryClientRef = Arc::new( + crate::store::MemoryClient::from_workspace_dir(workspace.path().join("store")) + .expect("memory client initialises against a fresh workspace"), + ); + let host = Arc::new(PipelineHost::without_tree_ingest(client)); + let ticked = Arc::new(std::sync::atomic::AtomicBool::new(false)); + + let held = try_hold_connection("gmail", "guard-short-circuit") + .expect("the first run takes the guard"); + let outcome = run_pipeline( + Arc::new(RecordingPipeline(ticked.clone())), + "gmail", + "guard-short-circuit", + &PipelineConfig::default(), + &host.context(), + ) + .await + .expect("a refused run is not a failure"); + + assert_eq!(outcome.note.as_deref(), Some(SYNC_ALREADY_RUNNING)); + assert_eq!(outcome.records_ingested, 0); + assert!( + !ticked.load(std::sync::atomic::Ordering::SeqCst), + "the refused run must not tick the pipeline" + ); + + // Released, the same call runs normally — the guard skips a concurrent + // run, it does not disable the connection. + drop(held); + let outcome = run_pipeline( + Arc::new(RecordingPipeline(ticked.clone())), + "gmail", + "guard-short-circuit", + &PipelineConfig::default(), + &host.context(), + ) + .await + .expect("the run succeeds once the guard is free"); + assert_eq!(outcome.records_ingested, 7); + assert!(ticked.load(std::sync::atomic::Ordering::SeqCst)); + } } diff --git a/sources/src/types_tests.rs b/sources/src/types_tests.rs index ed9f5ec..53771f3 100644 --- a/sources/src/types_tests.rs +++ b/sources/src/types_tests.rs @@ -259,3 +259,145 @@ fn max_items_is_applicable_to_composio_and_rss_but_not_other_kinds() { assert!(patch().validate_for_kind(SourceKind::GithubRepo).is_err()); assert!(patch().validate_for_kind(SourceKind::WebPage).is_err()); } + +/// The engine keeps its own copy of these types in `memory/sources/types.rs`, +/// and the two are joined by a live wire: `tinymemory-core`'s engine seam +/// converts between them with `serde_json::to_value` / `from_value` for the +/// tree-coupled source kinds, in both directions. Nothing but the serialised +/// shape holds that seam together — the copies are distinct Rust types in +/// distinct crates and neither compiles against the other. +/// +/// So a renamed field or a new `SourceKind` variant on either side is not a +/// compile error. It is a runtime failure on the first external-source sync +/// after the engine pin moves, at the point of conversion, far from the edit +/// that caused it. +/// +/// These pin the full serialised shape of each type that crosses. A failure +/// here means the copies have diverged and the change needs coordinating +/// across both crates, never a local edit to the expectation. +#[test] +fn source_entry_wire_format_is_pinned() { + let entry = MemorySourceEntry { + id: "src_pinned".into(), + kind: SourceKind::GithubRepo, + label: "Pinned".into(), + enabled: false, + toolkit: Some("gmail".into()), + connection_id: Some("conn-1".into()), + path: Some("/notes".into()), + glob: Some("**/*.md".into()), + url: Some("https://github.com/tinyhumansai/tinymemory".into()), + branch: Some("main".into()), + paths: vec!["core/src".into()], + max_commits: Some(10), + max_issues: Some(20), + max_prs: Some(30), + query: Some("from:me".into()), + since_days: Some(7), + max_items: Some(40), + selector: Some("article".into()), + max_tokens_per_sync: Some(50_000), + max_cost_per_sync_usd: Some(1.5), + sync_depth_days: Some(90), + }; + + assert_eq!( + serde_json::to_value(&entry).unwrap(), + serde_json::json!({ + "id": "src_pinned", + "kind": "github_repo", + "label": "Pinned", + "enabled": false, + "toolkit": "gmail", + "connection_id": "conn-1", + "path": "/notes", + "glob": "**/*.md", + "url": "https://github.com/tinyhumansai/tinymemory", + "branch": "main", + "paths": ["core/src"], + "max_commits": 10, + "max_issues": 20, + "max_prs": 30, + "query": "from:me", + "since_days": 7, + "max_items": 40, + "selector": "article", + "max_tokens_per_sync": 50000, + "max_cost_per_sync_usd": 1.5, + "sync_depth_days": 90 + }) + ); +} + +/// Every optional field is skipped when absent, so an entry carrying only its +/// required fields is a four-key object. A `skip_serializing_if` dropped from +/// one copy and not the other changes what the seam sends without changing +/// what either side compiles. +#[test] +fn an_empty_source_entry_serialises_to_its_required_fields_only() { + let entry = MemorySourceEntry { + id: "src_min".into(), + label: "Minimal".into(), + ..default_entry() + }; + + assert_eq!( + serde_json::to_value(&entry).unwrap(), + serde_json::json!({ + "id": "src_min", + "kind": "folder", + "label": "Minimal", + "enabled": true + }) + ); +} + +/// `SourceItem` crosses the same seam, on the `list_items` direction. +#[test] +fn source_item_wire_format_is_pinned() { + assert_eq!( + serde_json::to_value(SourceItem { + id: "item-1".into(), + title: "Quarterly planning".into(), + updated_at_ms: Some(1_777_000_000_000), + }) + .unwrap(), + serde_json::json!({ + "id": "item-1", + "title": "Quarterly planning", + "updated_at_ms": 1_777_000_000_000i64 + }) + ); + + assert_eq!( + serde_json::to_value(SourceItem { + id: "item-2".into(), + title: "No timestamp".into(), + updated_at_ms: None, + }) + .unwrap(), + serde_json::json!({ "id": "item-2", "title": "No timestamp" }) + ); +} + +/// `SourceContent` crosses the same seam, on the `read_item` direction. +#[test] +fn source_content_wire_format_is_pinned() { + assert_eq!( + serde_json::to_value(SourceContent { + id: "item-1".into(), + title: "Quarterly planning".into(), + body: "# Roadmap".into(), + content_type: ContentType::Markdown, + metadata: serde_json::json!({ "author": "shanu" }), + }) + .unwrap(), + serde_json::json!({ + "id": "item-1", + "title": "Quarterly planning", + "body": "# Roadmap", + "content_type": "markdown", + "metadata": { "author": "shanu" } + }) + ); +}