You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: cleanup.rs
    path of file: ./repos/cube/rust/cubestore/cubestore/src/remotefs
    
    the code of the file until where you have to start completion: use crate::config::ConfigObj;
use crate::metastore::MetaStore;
use crate::remotefs::RemoteFs;
use crate::{app_metrics, CubeError};
use chrono::Utc;
use datafusion::cube_ext;
use std::collections::HashSet;
use std::path::Path;
use std::sync::Arc;
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;

pub struct RemoteFsCleanup {
    config: Arc<dyn ConfigObj>,
    remote_fs: Arc<dyn RemoteFs>,
    metastore: Arc<dyn MetaStore>,
    stopped_token: CancellationToken,
}

crate::di_service!(RemoteFsCleanup, []);

impl RemoteFsCleanup {
    pub fn new(
        config: Arc<dyn ConfigObj>,
        remote_fs: Arc<dyn RemoteFs>,
        metastore: Arc<dyn MetaStore>,
    ) -> Self {
        Self {
            config,
            remote_fs,
            metastore,
            stopped_token: CancellationToken::new(),
        }
    }

    pub async fn wait_local_cleanup_loop(&self) {
        self.cleanup_local_files_loop().await;
    }

    #[allow(dead_code)]
    pub async fn wait_remote_fs_cleanup_loop(&self, stat_tags: Vec<String>) {
        self.cleanup_remotefs_loop(stat_tags).await;
    }

    pub fn stop(&self) {
        self.stopped_token.cancel()
    }

    async fn cleanup_remotefs_loop(&self, stat_tags: Vec<String>) -> () {
        let token = self.stopped_token.child_token();
        let cleanup_interval =
            Duration::from_secs(self.config.remote_files_cleanup_interval_secs());
        let cleanup_files_delay = self.config.remote_files_cleanup_delay_secs() as i64;
        let remote_fs = self.remote_fs.clone();
        let mut files_to_remove: HashSet<String> = HashSet::new(
    
    async fn queue_cleanup_local_files() {
        Config::test("cleanup_local_files")
            .update_config(|mut c| {
                c.local_files_cleanup_delay_secs = 2;
                c.local_files_cleanup_interval_secs = 1;
                c
            })
            .start_test(async move |services| {
                let service = services.sql_service;
                let meta_store = services.meta_store;
                let remote_fs = services.injector.get_service_typed::<dyn RemoteFs>().await;
                let _ = service.exec_query("CREATE SCHEMA test").await.unwrap();
                let _ = service
                    .exec_query("CREATE TABLE test.tst (a int, b int)")
                    .await
                    .unwrap();
                let _ = service
                    .exec_query("INSERT INTO test.tst (a, b) VALUES (10, 10), (20 , 20)")
                    .await
                    .unwrap();
                let _ = service
                    .exec_query("INSERT INTO test.tst (a, b) VALUES (20, 20), (40 , 40)")
                    .await
                    .unwrap();
                let files = remove_root_paritition(meta_store.get_all_filenames().await.unwrap());
                assert_eq!(files.len(), 2);
                for f in files.iter() {
                    let path = remote_fs.local_file(f.clone()).await.unwrap();
                    assert!(Path::new(&path).exists());
                }
                let path = remote_fs.local_file("metastore".to_string()).await.unwrap();
                assert!(Path::new(&path).exists());

                meta_store
                    .delete_chunks_without_checks(vec![1])
                    .await
                    .unwrap();

                assert_eq!(
                    remove_root_paritition(meta_store.get_all_filenames().await.unwrap()).len(),
                    1
                );
                for f in files.iter() {
                    let path = remote_fs.local_file(f.clone()).await.unwrap();
                    assert!(Path::new(&path).exists());
                }
                Delay::new(Duration::from_millis(4000)).await; // TODO logger init conflict

                let path = remote_fs.local_file(files[0].clone()).await.unwrap();
                assert!(!Path::new(&path).exists());

                let path = remote_fs.local_file(files[1].clone()).await.unwrap();
                assert!(Path::new(&path).exists());

                let path = remote_fs.local_file("metastore".to_string()).await.unwrap();
                assert!(Path::new(&path).exists());

                let _ = service.exec_query("SELECT * FROM test.tst").await.unwrap();
            })
            .await;
    }