sync

Copies changed records from source to target, batch by batch.

Source code in vecparity/sync/engine.py

class SyncEngine:
    """Copies changed records from `source` to `target`, batch by batch."""

    def __init__(
        self,
        source: VectorDBAdapter,
        target: VectorDBAdapter,
        batch_size: int = 500,
        cursor: float | None = None,
    ) -> None:
        self.source = source
        self.target = target
        self.batch_size = batch_size
        self.cursor = cursor
        self.stats = SyncStats(last_cursor=cursor)

    def run_once(self) -> int:
        """Replicate everything changed since the current cursor.
        Returns the number of records synced, and advances the cursor."""
        batch: list[VectorRecord] = []
        synced = 0
        max_seen = self.cursor

        for record in self.source.list_changed_since(self.cursor):
            batch.append(record)
            if record.updated_at is not None:
                max_seen = max(max_seen or 0, record.updated_at)
            if len(batch) >= self.batch_size:
                self.target.upsert(batch)
                synced += len(batch)
                self.stats.batches += 1
                batch = []

        if batch:
            self.target.upsert(batch)
            synced += len(batch)
            self.stats.batches += 1

        self.cursor = max_seen
        self.stats.last_cursor = max_seen
        self.stats.records_synced += synced
        return synced

    def run_until_caught_up(self, poll_interval: float = 5.0, idle_passes: int = 2) -> None:
        """Poll `run_once` until N consecutive passes sync nothing."""
        consecutive_idle = 0
        while consecutive_idle < idle_passes:
            synced = self.run_once()
            consecutive_idle = consecutive_idle + 1 if synced == 0 else 0
            if consecutive_idle < idle_passes:
                time.sleep(poll_interval)

run_once()

Replicate everything changed since the current cursor. Returns the number of records synced, and advances the cursor.

Source code in vecparity/sync/engine.py

def run_once(self) -> int:
    """Replicate everything changed since the current cursor.
    Returns the number of records synced, and advances the cursor."""
    batch: list[VectorRecord] = []
    synced = 0
    max_seen = self.cursor

    for record in self.source.list_changed_since(self.cursor):
        batch.append(record)
        if record.updated_at is not None:
            max_seen = max(max_seen or 0, record.updated_at)
        if len(batch) >= self.batch_size:
            self.target.upsert(batch)
            synced += len(batch)
            self.stats.batches += 1
            batch = []

    if batch:
        self.target.upsert(batch)
        synced += len(batch)
        self.stats.batches += 1

    self.cursor = max_seen
    self.stats.last_cursor = max_seen
    self.stats.records_synced += synced
    return synced

run_until_caught_up(poll_interval=5.0, idle_passes=2)

Poll run_once until N consecutive passes sync nothing.

Source code in vecparity/sync/engine.py

def run_until_caught_up(self, poll_interval: float = 5.0, idle_passes: int = 2) -> None:
    """Poll `run_once` until N consecutive passes sync nothing."""
    consecutive_idle = 0
    while consecutive_idle < idle_passes:
        synced = self.run_once()
        consecutive_idle = consecutive_idle + 1 if synced == 0 else 0
        if consecutive_idle < idle_passes:
            time.sleep(poll_interval)

Source code in vecparity/sync/engine.py

@dataclass
class SyncStats:
    batches: int = 0
    records_synced: int = 0
    last_cursor: float | None = None
    started_at: float = field(default_factory=time.time)