From bfa5132d1de61aa3da867b3280dfd310955c08c6 Mon Sep 17 00:00:00 2001 From: krypticmouse Date: Thu, 26 Mar 2026 21:29:50 +0000 Subject: [PATCH] feat: add SyncScheduler for periodic incremental sync Implements SyncScheduler that runs a daemon background thread to call SyncEngine.sync() on all registered connectors at a configurable interval. Provides run_once() as a synchronous helper for testing. Disconnected connectors are skipped each cycle; errors are logged without stopping the loop. 4 tests covering run_once, disconnected skip, start/stop, and per-connector chunk counts. Co-Authored-By: Claude Sonnet 4.6 --- src/openjarvis/connectors/scheduler.py | 149 +++++++++++++++++++++++++ tests/connectors/test_scheduler.py | 140 +++++++++++++++++++++++ 2 files changed, 289 insertions(+) create mode 100644 src/openjarvis/connectors/scheduler.py create mode 100644 tests/connectors/test_scheduler.py diff --git a/src/openjarvis/connectors/scheduler.py b/src/openjarvis/connectors/scheduler.py new file mode 100644 index 00000000..805a1fcb --- /dev/null +++ b/src/openjarvis/connectors/scheduler.py @@ -0,0 +1,149 @@ +"""SyncScheduler — background thread for periodic incremental connector syncs. + +Registers connectors for timed re-sync and runs them on a configurable +interval. Designed to be long-lived (daemon thread) inside a running +OpenJarvis server process. + +Typical usage:: + + store = KnowledgeStore(db_path=":memory:") + pipeline = IngestionPipeline(store) + engine = SyncEngine(pipeline) + + scheduler = SyncScheduler(engine, interval_seconds=3600) + scheduler.add(gmail_connector) + scheduler.add(slack_connector) + scheduler.start() # background thread syncs every hour + + # Later: + scheduler.stop() +""" + +from __future__ import annotations + +import logging +import threading +from typing import Dict, List, Optional + +from openjarvis.connectors._stubs import BaseConnector +from openjarvis.connectors.sync_engine import SyncEngine + +logger = logging.getLogger(__name__) + + +class SyncScheduler: + """Runs incremental sync for all registered connectors on a schedule. + + Parameters + ---------- + sync_engine: + The :class:`~openjarvis.connectors.sync_engine.SyncEngine` used to + drive each connector's sync (handles checkpointing). + interval_seconds: + How often (in seconds) to sync all connected connectors. + Defaults to ``3600`` (one hour). + """ + + def __init__(self, sync_engine: SyncEngine, interval_seconds: int = 3600) -> None: + self._engine = sync_engine + self._interval = interval_seconds + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + self._connectors: List[BaseConnector] = [] + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def add(self, connector: BaseConnector) -> None: + """Register a connector for scheduled sync. + + Parameters + ---------- + connector: + Any :class:`~openjarvis.connectors._stubs.BaseConnector` instance. + Only connected connectors are synced during each cycle. + """ + self._connectors.append(connector) + + def start(self) -> None: + """Start the background sync thread. + + The thread is a daemon so it does not prevent process exit. The + first sync run occurs after one full *interval_seconds* wait. + Calling :meth:`start` on an already-running scheduler is a no-op. + """ + if self._thread is not None and self._thread.is_alive(): + logger.debug("SyncScheduler already running; ignoring start()") + return + + self._stop.clear() + self._thread = threading.Thread( + target=self._loop, daemon=True, name="sync_scheduler" + ) + self._thread.start() + logger.info( + "SyncScheduler started (interval=%ds, connectors=%d)", + self._interval, + len(self._connectors), + ) + + def stop(self) -> None: + """Stop the background sync thread. + + Signals the thread to exit and waits up to 5 seconds for it to + finish the current sync cycle. Safe to call even when the scheduler + is not running. + """ + self._stop.set() + if self._thread is not None and self._thread.is_alive(): + self._thread.join(timeout=5) + self._thread = None + logger.info("SyncScheduler stopped") + + def run_once(self) -> Dict[str, int]: + """Sync all connected connectors once (synchronous, non-blocking test helper). + + Returns + ------- + dict[str, int] + Mapping of ``connector_id`` → number of new chunks ingested. + Only connectors that are currently connected are included. + """ + results: Dict[str, int] = {} + for conn in self._connectors: + if conn.is_connected(): + try: + count = self._engine.sync(conn) + results[conn.connector_id] = count + except Exception as exc: + logger.error( + "run_once sync failed for %s: %s", conn.connector_id, exc + ) + return results + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _loop(self) -> None: + """Background thread body: wait *interval* seconds then sync.""" + while not self._stop.wait(timeout=self._interval): + for conn in self._connectors: + if conn.is_connected(): + try: + count = self._engine.sync(conn) + logger.debug( + "Scheduled sync completed for %s (%d items)", + conn.connector_id, + count, + ) + except Exception as exc: + logger.error( + "Scheduled sync failed for %s: %s", + conn.connector_id, + exc, + ) + + +__all__ = ["SyncScheduler"] diff --git a/tests/connectors/test_scheduler.py b/tests/connectors/test_scheduler.py new file mode 100644 index 00000000..3b0ad1cd --- /dev/null +++ b/tests/connectors/test_scheduler.py @@ -0,0 +1,140 @@ +"""Tests for SyncScheduler — periodic incremental sync background thread.""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import Iterator, Optional + +import pytest + +from openjarvis.connectors._stubs import BaseConnector, Document, SyncStatus +from openjarvis.connectors.pipeline import IngestionPipeline +from openjarvis.connectors.scheduler import SyncScheduler +from openjarvis.connectors.store import KnowledgeStore +from openjarvis.connectors.sync_engine import SyncEngine + +# --------------------------------------------------------------------------- +# Fixtures and helpers +# --------------------------------------------------------------------------- + + +class _FakeConnector(BaseConnector): + """A minimal connector that yields a fixed number of documents.""" + + display_name = "Fake Scheduler Connector" + auth_type = "filesystem" + + def __init__( + self, + connector_id: str = "fake_sched", + *, + connected: bool = True, + doc_count: int = 2, + ) -> None: + self.connector_id = connector_id # type: ignore[misc] + self._connected = connected + self._doc_count = doc_count + + def is_connected(self) -> bool: + return self._connected + + def disconnect(self) -> None: + self._connected = False + + def sync( + self, *, since: Optional[datetime] = None, cursor: Optional[str] = None + ) -> Iterator[Document]: + for i in range(self._doc_count): + yield Document( + doc_id=f"{self.connector_id}:{i}", + source=self.connector_id, + doc_type="note", + content=f"Scheduled sync document {self.connector_id}:{i}.", + title=f"Doc {i}", + ) + + def sync_status(self) -> SyncStatus: + return SyncStatus(state="idle", items_synced=self._doc_count) + + +@pytest.fixture() +def engine(tmp_path: Path) -> SyncEngine: + """Return a SyncEngine backed by in-memory stores.""" + store = KnowledgeStore(db_path=":memory:") + pipeline = IngestionPipeline(store) + return SyncEngine(pipeline, state_db=str(tmp_path / "state.db")) + + +# --------------------------------------------------------------------------- +# Test 1: add + run_once syncs a connected connector +# --------------------------------------------------------------------------- + + +def test_run_once_syncs_connected_connector(engine: SyncEngine) -> None: + """run_once() calls engine.sync() for a connected connector.""" + conn = _FakeConnector("conn_single", connected=True, doc_count=3) + scheduler = SyncScheduler(engine, interval_seconds=3600) + scheduler.add(conn) + + results = scheduler.run_once() + + assert conn.connector_id in results + assert results[conn.connector_id] == 3 + + +# --------------------------------------------------------------------------- +# Test 2: run_once skips disconnected connectors +# --------------------------------------------------------------------------- + + +def test_run_once_skips_disconnected_connector(engine: SyncEngine) -> None: + """run_once() does not attempt to sync a disconnected connector.""" + connected = _FakeConnector("conn_yes", connected=True, doc_count=1) + disconnected = _FakeConnector("conn_no", connected=False, doc_count=5) + + scheduler = SyncScheduler(engine, interval_seconds=3600) + scheduler.add(connected) + scheduler.add(disconnected) + + results = scheduler.run_once() + + assert "conn_yes" in results + assert "conn_no" not in results + + +# --------------------------------------------------------------------------- +# Test 3: start/stop does not crash +# --------------------------------------------------------------------------- + + +def test_start_stop_does_not_crash(engine: SyncEngine) -> None: + """start() and stop() complete without error even with no connectors.""" + scheduler = SyncScheduler(engine, interval_seconds=60) + scheduler.start() + assert scheduler._thread is not None + assert scheduler._thread.is_alive() + scheduler.stop() + # After stop the internal thread reference is cleared + assert scheduler._thread is None + + +# --------------------------------------------------------------------------- +# Test 4: run_once returns chunk counts per connector +# --------------------------------------------------------------------------- + + +def test_run_once_returns_chunk_counts(engine: SyncEngine) -> None: + """run_once() returns a mapping of connector_id → chunks ingested.""" + conn_a = _FakeConnector("conn_a", connected=True, doc_count=2) + conn_b = _FakeConnector("conn_b", connected=True, doc_count=4) + + scheduler = SyncScheduler(engine, interval_seconds=3600) + scheduler.add(conn_a) + scheduler.add(conn_b) + + results = scheduler.run_once() + + assert results["conn_a"] == 2 + assert results["conn_b"] == 4 + assert len(results) == 2