···

       
10
       
10
       
 
       
    db::{Db, EventRecord},

     

       
11
       
11
       
 
       
    error::AppError,

     

       
12
       
12
       
 
       
    jetstream::JetstreamClient,

     

       
13
       
13
       
+
       
    utils::CLOCK,

     

       
13
       
14
       
 
       
};

     

       
14
       
15
       
 
       


     

       
15
       
16
       
 
       
mod api;

     
···

       
147
       
148
       
 
       
    let from = Arc::new(Db::new(".fjall_data_from").expect("couldnt create db"));

     

       
148
       
149
       
 
       
    let to = Arc::new(Db::new(".fjall_data_to").expect("couldnt create db"));

     

       
149
       
150
       
 
       


     

       
150
       
150
       
-
       
    let mut threads = Vec::new();

     

       
151
       
151
       
-
       
    for nsid in from.get_nsids() {

     

       
151
       
151
       
+
       
    let nsids = from.get_nsids().collect::<Vec<_>>();

     

       
152
       
152
       
+
       
    let mut threads = Vec::with_capacity(nsids.len());

     

       
153
       
153
       
+
       
    let start = CLOCK.now();

     

       
154
       
154
       
+
       
    for nsid in nsids {

     

       
152
       
155
       
 
       
        let from = from.clone();

     

       
153
       
156
       
 
       
        let to = to.clone();

     

       
154
       
157
       
 
       
        threads.push(std::thread::spawn(move || {

     
···

       
168
       
171
       
 
       
            count

     

       
169
       
172
       
 
       
        }));

     

       
170
       
173
       
 
       
    }

     

       
171
       
171
       
-
       


     

       
172
       
174
       
 
       
    let mut total_count = 0_u64;

     

       
173
       
175
       
 
       
    for thread in threads {

     

       
174
       
174
       
-
       
        total_count += thread.join().expect("thread panicked");

     

       
176
       
176
       
+
       
        let count = thread.join().expect("thread panicked");

     

       
177
       
177
       
+
       
        total_count += count;

     

       
175
       
178
       
 
       
    }

     

       
179
       
179
       
+
       
    let read_time = start.elapsed();

     

       
180
       
180
       
+
       
    let read_per_second = total_count as f64 / read_time.as_secs_f64();

     

       
176
       
181
       
 
       
    to.sync(true).expect("cant sync");

     

       
177
       
177
       
-
       
    tracing::info!("migrated {total_count} events!");

     

       
182
       
182
       
+
       
    let total_time = start.elapsed();

     

       
183
       
183
       
+
       
    let write_per_second = total_count as f64 / (total_time - read_time).as_secs_f64();

     

       
184
       
184
       
+
       
    tracing::info!(

     

       
185
       
185
       
+
       
        "migrated {total_count} events in {total_time:?} ({read_per_second:.2} rps, {write_per_second:.2} wps)"

     

       
186
       
186
       
+
       
    );

     

       
178
       
187
       
 
       
}