async def run(
        self,
        sync: schemas.Sync,
        sync_job: schemas.SyncJob,
        dag: schemas.SyncDag,
        collection: schemas.Collection,
        source_connection: schemas.Connection,
        current_user: schemas.User,
        access_token: Optional[str] = None,
    ) -> schemas.Sync:
        """Run a sync.

        Args:
        ----
            sync (schemas.Sync): The sync to run.
            sync_job (schemas.SyncJob): The sync job to run.
            dag (schemas.SyncDag): The DAG to run.
            collection (schemas.Collection): The collection to sync.
            source_connection (schemas.Connection): The source connection to sync.
            current_user (schemas.User): The current user.
            access_token (Optional[str]): Optional access token to use
                instead of stored credentials.

        Returns:
        -------
            schemas.Sync: The sync.
        """
        try:
            async with get_db_context() as db:
                # Create dedicated orchestrator instance
                orchestrator = await SyncFactory.create_orchestrator(
                    db=db,
                    sync=sync,
                    sync_job=sync_job,
                    dag=dag,
                    collection=collection,
                    source_connection=source_connection,
                    current_user=current_user,
                    access_token=access_token,
                )
        except Exception as e:
            logger.error(f"Error during sync orchestrator creation: {e}")
            # Fail the sync job if orchestrator creation failed
            await sync_job_service.update_status(
                sync_job_id=sync_job.id,
                status=SyncJobStatus.FAILED,
                current_user=current_user,
                error=str(e),
                failed_at=datetime.now(),
            )
            raise e

        # Run the sync with the dedicated orchestrator instance
        return await orchestrator.run()
Run a sync.

        Args:
        ----
            sync (schemas.Sync): The sync to run.
            sync_job (schemas.SyncJob): The sync job to run.
            dag (schemas.SyncDag): The DAG to run.
            collection (schemas.Collection): The collection to sync.
            source_connection (schemas.Connection): The source connection to sync.
            current_user (schemas.User): The current user.
            access_token (Optional[str]): Optional access token to use
                instead of stored credentials.

        Returns:
        -------
            schemas.Sync: The sync.
        
backend/airweave/core/sync_service.py