tracks lexicons and how many times they appeared on the jetstream
feat(server): run all trees migrations in parallel
+26
-16
+2
-2
server/src/db/mod.rs
+2
-2
server/src/db/mod.rs
···
216
216
})
217
217
}
218
218
219
-
pub fn get_nsids(&self) -> impl Iterator<Item = impl Deref<Target = str>> {
219
+
pub fn get_nsids(&self) -> impl Iterator<Item = impl Deref<Target = str> + 'static> {
220
220
self.inner
221
221
.list_partitions()
222
222
.into_iter()
···
498
498
})
499
499
}
500
500
501
-
pub fn get_nsids(&self) -> impl Iterator<Item = impl Deref<Target = str>> {
501
+
pub fn get_nsids(&self) -> impl Iterator<Item = impl Deref<Target = str> + 'static> {
502
502
self.inner
503
503
.list_partitions()
504
504
.into_iter()
+24
-14
server/src/main.rs
+24
-14
server/src/main.rs
···
144
144
}
145
145
146
146
fn migrate() {
147
-
let from = DbOld::new(".fjall_data").expect("couldnt create db");
148
-
let to = Db::new(".fjall_data_migrated").expect("couldnt create db");
147
+
let from = Arc::new(DbOld::new(".fjall_data").expect("couldnt create db"));
148
+
let to = Arc::new(Db::new(".fjall_data_migrated").expect("couldnt create db"));
149
149
150
-
let mut total_count = 0_u64;
150
+
let mut threads = Vec::new();
151
151
for nsid in from.get_nsids() {
152
-
tracing::info!("migrating {} ...", nsid.deref());
153
-
for hit in from.get_hits(&nsid, ..) {
154
-
let (timestamp, data) = hit.expect("cant read event");
155
-
to.record_event(EventRecord {
156
-
nsid: nsid.to_smolstr(),
157
-
timestamp,
158
-
deleted: data.deleted,
159
-
})
160
-
.expect("cant record event");
161
-
total_count += 1;
162
-
}
152
+
let from = from.clone();
153
+
let to = to.clone();
154
+
threads.push(std::thread::spawn(move || {
155
+
tracing::info!("migrating {} ...", nsid.deref());
156
+
let mut count = 0_u64;
157
+
for hit in from.get_hits(&nsid, ..) {
158
+
let (timestamp, data) = hit.expect("cant read event");
159
+
to.record_event(EventRecord {
160
+
nsid: nsid.to_smolstr(),
161
+
timestamp,
162
+
deleted: data.deleted,
163
+
})
164
+
.expect("cant record event");
165
+
count += 1;
166
+
}
167
+
count
168
+
}));
163
169
}
164
170
171
+
let mut total_count = 0_u64;
172
+
for thread in threads {
173
+
total_count += thread.join().expect("thread panicked");
174
+
}
165
175
to.sync(true).expect("cant sync");
166
176
tracing::info!("migrated {total_count} events!");
167
177
}