Repository: libsql/libsql
File: libsql/src/replication/mod.rs (Rust)

Code before:

    pub async fn get_sync_usage_stats(&self) -> Result<SyncUsageStats> {
        let mut replicator = self.replicator.lock().await;
        match replicator.client_mut() {
            Either::Right(_) => {
                Err(crate::errors::Error::Misuse(
                    "Trying to get sync usage stats, but this is a local replicator".into(),
                ))
            }
            Either::Left(c) => {
                let stats = c.sync_stats();
                let result = SyncUsageStats {
                    prefetched_bytes: stats.prefetched_bytes.load(std::sync::atomic::Ordering::SeqCst),
                    prefetched_bytes_discarded_due_to_new_session: stats
                        .prefetched_bytes_discarded_due_to_new_session.load(std::sync::atomic::Ordering::SeqCst),
                    prefetched_bytes_discarded_due_to_consecutive_handshake: stats
                        .prefetched_bytes_discarded_due_to_consecutive_handshake.load(std::sync::atomic::Ordering::SeqCst),
                    prefetched_bytes_discarded_due_to_invalid_frame_header: stats
                        .prefetched_bytes_discarded_due_to_invalid_frame_header.load(std::sync::atomic::Ordering::SeqCst),
                    synced_bytes_discarded_due_to_invalid_frame_header: stats
                        .synced_bytes_discarded_due_to_invalid_frame_header.load(std::sync::atomic::Ordering::SeqCst),
                    prefetched_bytes_used: stats.prefetched_bytes_used.load(std::sync::atomic::Ordering::SeqCst),
                    synced_bytes_used: stats.synced_bytes_used.load(std::sync::atomic::Ordering::SeqCst),
                    snapshot_bytes: stats.snapshot_bytes.load(std::sync::atomic::Ordering::SeqCst),
                };
                tracing::info!("Sync usage stats: {:?}", result);
                Ok(result)
            }
        }

    }

    pub async fn sync_oneshot(&self) -> Result<Replicated> {
        use libsql_replication::replicator::ReplicatorClient;

        let mut replicator = self.replicator.lock().await;
        if !matches!(replicator.client_mut(), Either::Left(_)) {
            return Err(crate::errors::Error::Misuse(
                "Trying to replicate from HTTP, but this is a local replicator".into(),
            ));
        }

        // we force a handshake to get the most up to date replication index from the primary.
        replicator.force_handshake();

        loop {
            match replicator.replicate().await {
                Err(libsql_replication::replicator::Error::Meta(
                    libsql_replication::meta::Error::LogIncompatible,
                )) => {
                    // The meta must have been marked as dirty, replicate again from scratch

Review comment (bug):
I don't think we want this to be info? I think this should at least be debug or else it will be spammed in the logs often? I think we should let users choose if they want this to be info or not by having their own log statement.

Code after:
            let jh = tokio::spawn(
                async move {
                    loop {
                        if let Err(e) = replicator2.sync_oneshot().await {
                            tracing::error!("replicator sync error: {}", e);
                        }

                        tokio::time::sleep(sync_duration).await;
                    }
                }
                .instrument(tracing::info_span!("sync_interval")),
            );

            replicator.bg_abort = Some(Arc::new(DropAbort(jh.abort_handle())));
        }

        Ok(replicator)
    }

    pub async fn with_local(
        client: LocalClient,
        db_path: PathBuf,
        auto_checkpoint: u32,
        encryption_config: Option<EncryptionConfig>,
    ) -> Result<Self> {
        let replicator = Arc::new(Mutex::new(
            Replicator::new_sqlite(
                Either::Right(client),
                db_path,
                auto_checkpoint,
                encryption_config,
            )
            .await?,
        ));

        Ok(Self {
            replicator,
            bg_abort: None,
            last_frames_synced: Arc::new(AtomicUsize::new(0)),
        })
    }

    pub async fn get_sync_usage_stats(&self) -> Result<SyncUsageStats> {
        let mut replicator = self.replicator.lock().await;
        match replicator.client_mut() {
            Either::Right(_) => {
                Err(crate::errors::Error::Misuse(
                    "Trying to get sync usage stats, but this is a local replicator".into(),
                ))
            }
            Either::Left(c) => {