ptr.pet
tracks lexicons and how many times they appeared on the jetstream
feat(server): actually make the block inserts start in parallel
+13
-16
server/src/db/mod.rs
+13
-16
server/src/db/mod.rs
···
175
175
drop(_guard);
176
176
177
177
// process the blocks
178
-
let mut blocks = Vec::with_capacity(data.len());
179
178
data.into_par_iter()
180
179
.map(|chunk| {
181
180
chunk
···
194
193
})
195
194
.collect::<Result<Vec<_>, _>>()
196
195
})
197
-
.collect_into_vec(&mut blocks);
198
-
199
-
// execute into db
200
-
for chunk in blocks.into_iter() {
201
-
let chunk = chunk?;
202
-
for (i, block, handle) in chunk {
203
-
self.sync_pool
204
-
.execute(move || match handle.insert(block.key, block.data) {
205
-
Ok(_) => {
206
-
tracing::info!("{}: [{i}] synced {}", block.written, handle.nsid())
207
-
}
208
-
Err(err) => tracing::error!("failed to sync block: {}", err),
209
-
});
210
-
}
211
-
}
196
+
.try_for_each(|chunk| {
197
+
let chunk = chunk?;
198
+
for (i, block, handle) in chunk {
199
+
self.sync_pool
200
+
.execute(move || match handle.insert(block.key, block.data) {
201
+
Ok(_) => {
202
+
tracing::info!("{}: [{i}] synced {}", block.written, handle.nsid())
203
+
}
204
+
Err(err) => tracing::error!("failed to sync block: {}", err),
205
+
});
206
+
}
207
+
AppResult::Ok(())
208
+
})?;
212
209
self.sync_pool.join();
213
210
214
211
Ok(())