From 3084e9baa40a81f84617bb0f79964d7cae5f5a9f Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Wed, 15 Jul 2026 16:24:29 +0200 Subject: [PATCH 1/3] SQL: Prevent duplicated binds --- opsqueue/src/common/submission.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 30c41f7e..462ef5b2 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -532,12 +532,10 @@ pub mod db { r#" SELECT id AS "id: SubmissionId" FROM submissions WHERE prefix = $1 UNION ALL - SELECT id AS "id: SubmissionId" FROM submissions_completed WHERE prefix = $2 + SELECT id AS "id: SubmissionId" FROM submissions_completed WHERE prefix = $1 UNION ALL - SELECT id AS "id: SubmissionId" FROM submissions_failed WHERE prefix = $3 + SELECT id AS "id: SubmissionId" FROM submissions_failed WHERE prefix = $1 "#, - prefix, - prefix, prefix ) .fetch_optional(conn.get_inner()) From ae678e9684bab1143f1802df405be79139eccefe Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Wed, 15 Jul 2026 17:02:48 +0200 Subject: [PATCH 2/3] SQL: Filter already reserved chunks from `Chunk` selection --- Cargo.lock | 1 + opsqueue/Cargo.toml | 2 + opsqueue/src/common/chunk.rs | 12 + opsqueue/src/consumer/dispatcher/mod.rs | 158 ++++++++- opsqueue/src/consumer/dispatcher/reserver.rs | 5 + opsqueue/src/consumer/strategy.rs | 337 +++++++++++-------- 6 files changed, 373 insertions(+), 142 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2b318677..90f9fcf1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2274,6 +2274,7 @@ dependencies = [ "humantime", "insta", "itertools 0.14.0", + "libsqlite3-sys", "moka", "moro-local", "object_store", diff --git a/opsqueue/Cargo.toml b/opsqueue/Cargo.toml index 486ea652..e75d457a 100644 --- a/opsqueue/Cargo.toml +++ b/opsqueue/Cargo.toml @@ -33,6 +33,7 @@ ux = "0.1.6" anyhow = "1.0.102" # Database: sqlx = { version = "0.8.2", features = ["sqlite", "runtime-tokio", "chrono"], optional = true } +libsqlite3-sys = { version = "0.30.1", optional = true } # Serialization: serde = { version = "1.0.203", features = ["derive"] } serde_json = "1.0.149" @@ -96,6 +97,7 @@ assert_matches = { version = "1.5.0" } # Dependencies only in use by the server-logic: server-logic = [ "dep:sqlx", + "dep:libsqlite3-sys", "dep:opentelemetry-otlp", "dep:opentelemetry-semantic-conventions", "dep:moka", diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index b5b84414..6b25b21e 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -124,6 +124,18 @@ impl From for i64 { } } +impl TryFrom for ChunkIndex { + type Error = crate::common::errors::TryFromIntError; + + fn try_from(value: i64) -> Result { + if value < 0 { + return Err(crate::common::errors::TryFromIntError(())); + } + + Ok(Self(u63::new(value as u64))) + } +} + impl TryFrom for ChunkIndex { type Error = crate::common::errors::TryFromIntError; fn try_from(value: u64) -> Result { diff --git a/opsqueue/src/consumer/dispatcher/mod.rs b/opsqueue/src/consumer/dispatcher/mod.rs index 26514b68..41b8b6d3 100644 --- a/opsqueue/src/consumer/dispatcher/mod.rs +++ b/opsqueue/src/consumer/dispatcher/mod.rs @@ -3,12 +3,13 @@ pub mod reserver; use crate::{ common::{ - chunk::{Chunk, ChunkId}, - submission::Submission, + chunk::{Chunk, ChunkId, ChunkIndex}, + submission::{Submission, SubmissionId}, }, - db::{magic::Bool, Connection, Pool, ReaderPool}, + db::{Connection, Pool, ReaderPool, magic::Bool}, }; use futures::stream::{StreamExt as _, TryStreamExt as _}; +use libsqlite3_sys as ffi; use metastate::MetaState; use reserver::Reserver; use sqlx::QueryBuilder; @@ -21,6 +22,63 @@ use std::sync::Arc; use super::strategy; use crate::common::StrategicMetadataMap; +unsafe extern "C" fn sqlite_reserved_chunk_lookup( + context: *mut ffi::sqlite3_context, + n_args: i32, + args: *mut *mut ffi::sqlite3_value, +) { + if n_args != 2 { + tracing::error!( + n_args, + "opsqueue_is_reserved called with unexpected argument count" + ); + // Fail open: this callback is an optimization only. + ffi::sqlite3_result_int(context, 0); + return; + } + + let user_data = ffi::sqlite3_user_data(context) as *const Reserver; + if user_data.is_null() { + tracing::error!("opsqueue_is_reserved called without registered reserver user_data"); + // Fail open: this callback is an optimization only. + ffi::sqlite3_result_int(context, 0); + return; + } + + let submission_id_raw = ffi::sqlite3_value_int64(*args.add(0)); + let chunk_index_raw = ffi::sqlite3_value_int64(*args.add(1)); + + let Ok(submission_id) = SubmissionId::try_from(submission_id_raw) else { + tracing::error!( + submission_id_raw, + "opsqueue_is_reserved got invalid submission_id" + ); + // Fail open: this callback is an optimization only. + ffi::sqlite3_result_int(context, 0); + return; + }; + let Ok(chunk_index) = ChunkIndex::try_from(chunk_index_raw) else { + tracing::error!( + chunk_index_raw, + "opsqueue_is_reserved got invalid chunk_index" + ); + // Fail open: this callback is an optimization only. + ffi::sqlite3_result_int(context, 0); + return; + }; + + let chunk_id = ChunkId::from((submission_id, chunk_index)); + let is_reserved = (*user_data).is_reserved(&chunk_id); + ffi::sqlite3_result_int(context, i32::from(is_reserved)); +} + +unsafe extern "C" fn sqlite_reserved_chunk_lookup_destructor(ptr: *mut std::ffi::c_void) { + if ptr.is_null() { + return; + } + let _boxed: Box> = Box::from_raw(ptr.cast()); +} + #[derive(Debug, Clone)] pub struct Dispatcher { reserver: Reserver, @@ -66,6 +124,8 @@ impl Dispatcher { stale_chunks_notifier: &UnboundedSender, ) -> Result, sqlx::Error> { let mut conn = pool.reader_conn().await?; + self.register_reserved_chunk_lookup(conn.get_inner()) + .await?; let mut query_builder = QueryBuilder::new(""); let stream = strategy .build_query(&mut query_builder, &self.metastate) @@ -79,6 +139,43 @@ impl Dispatcher { .await } + async fn register_reserved_chunk_lookup( + &self, + conn: &mut sqlx::SqliteConnection, + ) -> Result<(), sqlx::Error> { + let mut handle = conn.lock_handle().await?; + let sqlite = handle.as_raw_handle().as_ptr(); + let function_name = b"opsqueue_is_reserved\0"; + + // Register the current reserver state on this connection. + // Re-registering replaces any previous callback on this handle. + let user_data = Box::new(self.reserver.clone()); + let user_data = Box::into_raw(user_data).cast::(); + + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + function_name.as_ptr().cast(), + 2, + ffi::SQLITE_UTF8, + user_data, + Some(sqlite_reserved_chunk_lookup), + None, + None, + Some(sqlite_reserved_chunk_lookup_destructor), + ) + }; + + if rc != ffi::SQLITE_OK { + unsafe { sqlite_reserved_chunk_lookup_destructor(user_data) }; + return Err(sqlx::Error::Protocol(format!( + "sqlite3_create_function_v2 failed with rc={rc}" + ))); + } + + Ok(()) + } + async fn reserve_chunk( &self, chunk: Chunk, @@ -144,3 +241,58 @@ impl Dispatcher { .run_pending_tasks_periodically(cancellation_token); } } + +#[cfg(test)] +#[cfg(feature = "server-logic")] +mod test { + use super::*; + use crate::common::chunk::ChunkId; + use crate::common::chunk::ChunkSize; + use crate::db::DBPools; + use tokio::sync::mpsc::unbounded_channel; + use ux::u63; + + #[sqlx::test] + async fn fetch_and_reserve_chunks_excludes_already_reserved(db: sqlx::SqlitePool) { + let pools = DBPools::from_test_pool(&db); + let dispatcher = Dispatcher::new(Duration::from_secs(60)); + let (stale_chunks_notifier, mut _stale_chunks_receiver) = unbounded_channel::(); + + let mut writer_conn = pools.writer_conn().await.unwrap(); + let submission_id = crate::common::submission::db::insert_submission_from_chunks( + None, + vec![Some("a".into()), Some("b".into()), Some("c".into())], + None, + StrategicMetadataMap::default(), + ChunkSize::default(), + &mut writer_conn, + ) + .await + .unwrap(); + + let pre_reserved_chunk = ChunkId::from((submission_id, u63::new(0).into())); + dispatcher + .reserver() + .try_reserve( + pre_reserved_chunk, + pre_reserved_chunk, + &stale_chunks_notifier, + ) + .expect("precondition: pre-reserving chunk should succeed"); + + let reserved = dispatcher + .fetch_and_reserve_chunks( + pools.reader_pool(), + strategy::Strategy::Oldest, + 10, + &stale_chunks_notifier, + ) + .await + .unwrap(); + + assert_eq!(reserved.len(), 2); + assert!(reserved.iter().all(|(chunk, _submission)| { + ChunkId::from((chunk.submission_id, chunk.chunk_index)) != pre_reserved_chunk + })); + } +} diff --git a/opsqueue/src/consumer/dispatcher/reserver.rs b/opsqueue/src/consumer/dispatcher/reserver.rs index 24d7ab1d..0dc6e740 100644 --- a/opsqueue/src/consumer/dispatcher/reserver.rs +++ b/opsqueue/src/consumer/dispatcher/reserver.rs @@ -82,6 +82,11 @@ where } } + /// Returns whether a key currently has an active reservation. + pub fn is_reserved(&self, key: &K) -> bool { + self.reservations.contains_key(key) + } + /// Removes a particular key-val from the reserver. /// Afterwards, it is possible to reserve it again. /// diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index 68f0493b..1b3c5815 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -40,14 +40,22 @@ impl Strategy { ) -> &'a mut QueryBuilder<'a, Sqlite> { use Strategy::*; match self { - Oldest => qb.push("SELECT * FROM chunks ORDER BY submission_id ASC"), - Newest => qb.push("SELECT * FROM chunks ORDER BY submission_id DESC"), + Oldest => qb + .push("SELECT * FROM chunks") + .push(" WHERE opsqueue_is_reserved(submission_id, chunk_index) = 0") + .push(" ORDER BY submission_id ASC"), + Newest => qb + .push("SELECT * FROM chunks") + .push(" WHERE opsqueue_is_reserved(submission_id, chunk_index) = 0") + .push(" ORDER BY submission_id DESC"), Random => { let random_offset: u16 = rand::random(); qb.push("SELECT * FROM chunks WHERE random_order >= ") .push_bind(random_offset) + .push(" AND opsqueue_is_reserved(submission_id, chunk_index) = 0") .push(" UNION ALL SELECT * FROM chunks WHERE random_order < ") .push_bind(random_offset) + .push(" AND opsqueue_is_reserved(submission_id, chunk_index) = 0") } PreferDistinct { @@ -96,15 +104,44 @@ pub type ChunkStream<'a> = BoxStream<'a, Result>; #[cfg(test)] #[cfg(feature = "server-logic")] pub mod test { - use crate::common::chunk::ChunkSize; - use crate::common::StrategicMetadataMap; - - use super::*; use itertools::Itertools; - use sqlformat::{format, FormatOptions, QueryParams}; + use libsqlite3_sys as ffi; + use sqlformat::{FormatOptions, QueryParams, format}; use sqlx::Row; use sqlx::{QueryBuilder, Sqlite, SqliteConnection}; + use super::*; + use crate::common::StrategicMetadataMap; + use crate::common::chunk::ChunkSize; + + unsafe extern "C" fn sqlite_reserved_chunk_lookup_noop( + context: *mut ffi::sqlite3_context, + _n_args: i32, + _args: *mut *mut ffi::sqlite3_value, + ) { + ffi::sqlite3_result_int(context, 0); + } + + async fn register_reserved_lookup_noop(conn: &mut SqliteConnection) { + let mut handle = conn.lock_handle().await.unwrap(); + let sqlite = handle.as_raw_handle().as_ptr(); + let function_name = b"opsqueue_is_reserved\0"; + let rc = unsafe { + ffi::sqlite3_create_function_v2( + sqlite, + function_name.as_ptr().cast(), + 2, + ffi::SQLITE_UTF8, + std::ptr::null_mut(), + Some(sqlite_reserved_chunk_lookup_noop), + None, + None, + None, + ) + }; + assert_eq!(rc, ffi::SQLITE_OK, "register opsqueue_is_reserved failed"); + } + async fn explain( qb: &mut sqlx::QueryBuilder<'_, Sqlite>, conn: &mut SqliteConnection, @@ -134,6 +171,7 @@ pub mod test { #[sqlx::test] pub async fn test_query_plan_oldest(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); + register_reserved_lookup_noop(&mut conn).await; let mut qb = QueryBuilder::new(""); let metastate = MetaState::default(); @@ -145,6 +183,8 @@ pub mod test { * FROM chunks + WHERE + opsqueue_is_reserved(submission_id, chunk_index) = 0 ORDER BY submission_id ASC "); @@ -157,6 +197,7 @@ pub mod test { #[sqlx::test] pub async fn test_query_plan_newest(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); + register_reserved_lookup_noop(&mut conn).await; let mut qb = QueryBuilder::new(""); let metastate = MetaState::default(); @@ -168,6 +209,8 @@ pub mod test { * FROM chunks + WHERE + opsqueue_is_reserved(submission_id, chunk_index) = 0 ORDER BY submission_id DESC "); @@ -180,6 +223,7 @@ pub mod test { #[sqlx::test] pub async fn test_query_plan_random(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); + register_reserved_lookup_noop(&mut conn).await; let metastate = MetaState::default(); let mut qb = QueryBuilder::new(""); @@ -193,6 +237,7 @@ pub mod test { chunks WHERE random_order >= ? + AND opsqueue_is_reserved(submission_id, chunk_index) = 0 UNION ALL SELECT * @@ -200,6 +245,7 @@ pub mod test { chunks WHERE random_order < ? + AND opsqueue_is_reserved(submission_id, chunk_index) = 0 "); let explained = explain(qb, &mut conn).await; @@ -208,8 +254,8 @@ pub mod test { 1, 0, COMPOUND QUERY 2, 1, LEFT-MOST SUBQUERY 5, 2, SEARCH chunks USING INDEX random_chunks_order (random_order>?) - 22, 1, UNION ALL - 25, 22, SEARCH chunks USING INDEX random_chunks_order (random_order= ? + AND opsqueue_is_reserved(submission_id, chunk_index) = 0 UNION ALL SELECT * @@ -415,6 +469,7 @@ pub mod test { chunks WHERE random_order < ? + AND opsqueue_is_reserved(submission_id, chunk_index) = 0 ), taken_company_id AS ( SELECT @@ -467,30 +522,30 @@ pub mod test { 3, 2, COMPOUND QUERY 4, 3, LEFT-MOST SUBQUERY 7, 4, SEARCH chunks USING INDEX random_chunks_order (random_order>?) - 16, 4, CORRELATED SCALAR SUBQUERY 5 - 20, 16, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 26, 16, LIST SUBQUERY 3 - 28, 26, SCAN json_each VIRTUAL TABLE INDEX 0: - 63, 3, UNION ALL - 66, 63, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 137, 125, CORRELATED SCALAR SUBQUERY 7 - 141, 137, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 147, 137, LIST SUBQUERY 3 - 149, 147, SCAN json_each VIRTUAL TABLE INDEX 0: - 184, 124, UNION ALL - 187, 184, SEARCH chunks USING INDEX random_chunks_order (random_order?) + 149, 133, CORRELATED SCALAR SUBQUERY 7 + 153, 149, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) + 159, 149, LIST SUBQUERY 3 + 161, 159, SCAN json_each VIRTUAL TABLE INDEX 0: + 196, 132, UNION ALL + 199, 196, SEARCH chunks USING INDEX random_chunks_order (random_order= ? + AND opsqueue_is_reserved(submission_id, chunk_index) = 0 UNION ALL SELECT * @@ -530,6 +587,7 @@ pub mod test { chunks WHERE random_order < ? + AND opsqueue_is_reserved(submission_id, chunk_index) = 0 ), taken_priority AS ( SELECT @@ -626,92 +684,92 @@ pub mod test { 5, 4, COMPOUND QUERY 6, 5, LEFT-MOST SUBQUERY 9, 6, SEARCH chunks USING INDEX random_chunks_order (random_order>?) - 18, 6, CORRELATED SCALAR SUBQUERY 5 - 22, 18, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 28, 18, LIST SUBQUERY 3 - 30, 28, SCAN json_each VIRTUAL TABLE INDEX 0: - 57, 6, CORRELATED SCALAR SUBQUERY 11 - 61, 57, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 67, 57, LIST SUBQUERY 9 - 69, 67, SCAN json_each VIRTUAL TABLE INDEX 0: - 104, 5, UNION ALL - 107, 104, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 217, 205, CORRELATED SCALAR SUBQUERY 7 - 221, 217, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 227, 217, LIST SUBQUERY 3 - 229, 227, SCAN json_each VIRTUAL TABLE INDEX 0: - 256, 205, CORRELATED SCALAR SUBQUERY 11 - 260, 256, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 266, 256, LIST SUBQUERY 9 - 268, 266, SCAN json_each VIRTUAL TABLE INDEX 0: - 303, 204, UNION ALL - 306, 303, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 418, 406, CORRELATED SCALAR SUBQUERY 5 - 422, 418, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 428, 418, LIST SUBQUERY 3 - 430, 428, SCAN json_each VIRTUAL TABLE INDEX 0: - 457, 406, CORRELATED SCALAR SUBQUERY 13 - 461, 457, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 467, 457, LIST SUBQUERY 9 - 469, 467, SCAN json_each VIRTUAL TABLE INDEX 0: - 504, 405, UNION ALL - 507, 504, SEARCH chunks USING INDEX random_chunks_order (random_order?) - 617, 605, CORRELATED SCALAR SUBQUERY 7 - 621, 617, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 627, 617, LIST SUBQUERY 3 - 629, 627, SCAN json_each VIRTUAL TABLE INDEX 0: - 656, 605, CORRELATED SCALAR SUBQUERY 13 - 660, 656, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) - 666, 656, LIST SUBQUERY 9 - 668, 666, SCAN json_each VIRTUAL TABLE INDEX 0: - 703, 604, UNION ALL - 706, 703, SEARCH chunks USING INDEX random_chunks_order (random_order?) + 229, 213, CORRELATED SCALAR SUBQUERY 7 + 233, 229, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) + 239, 229, LIST SUBQUERY 3 + 241, 239, SCAN json_each VIRTUAL TABLE INDEX 0: + 268, 213, CORRELATED SCALAR SUBQUERY 11 + 272, 268, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) + 278, 268, LIST SUBQUERY 9 + 280, 278, SCAN json_each VIRTUAL TABLE INDEX 0: + 315, 212, UNION ALL + 318, 315, SEARCH chunks USING INDEX random_chunks_order (random_order?) + 438, 422, CORRELATED SCALAR SUBQUERY 5 + 442, 438, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) + 448, 438, LIST SUBQUERY 3 + 450, 448, SCAN json_each VIRTUAL TABLE INDEX 0: + 477, 422, CORRELATED SCALAR SUBQUERY 13 + 481, 477, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) + 487, 477, LIST SUBQUERY 9 + 489, 487, SCAN json_each VIRTUAL TABLE INDEX 0: + 524, 421, UNION ALL + 527, 524, SEARCH chunks USING INDEX random_chunks_order (random_order?) + 645, 629, CORRELATED SCALAR SUBQUERY 7 + 649, 645, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) + 655, 645, LIST SUBQUERY 3 + 657, 655, SCAN json_each VIRTUAL TABLE INDEX 0: + 684, 629, CORRELATED SCALAR SUBQUERY 13 + 688, 684, SEARCH submissions_metadata USING COVERING INDEX lookup_submission_by_metadata (metadata_key=? AND metadata_value=? AND submission_id=?) + 694, 684, LIST SUBQUERY 9 + 696, 694, SCAN json_each VIRTUAL TABLE INDEX 0: + 731, 628, UNION ALL + 734, 731, SEARCH chunks USING INDEX random_chunks_order (random_order = Strategy::Random .build_query(&mut query_builder, &Default::default()) From 2addea092f009c29997b370b3cb899d0d24beee8 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Thu, 16 Jul 2026 17:01:45 +0200 Subject: [PATCH 3/3] fixup! SQL: Filter already reserved chunks from `Chunk` selection --- opsqueue/src/consumer/dispatcher/mod.rs | 2 +- opsqueue/src/consumer/strategy.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/opsqueue/src/consumer/dispatcher/mod.rs b/opsqueue/src/consumer/dispatcher/mod.rs index 41b8b6d3..9df1ae9a 100644 --- a/opsqueue/src/consumer/dispatcher/mod.rs +++ b/opsqueue/src/consumer/dispatcher/mod.rs @@ -6,7 +6,7 @@ use crate::{ chunk::{Chunk, ChunkId, ChunkIndex}, submission::{Submission, SubmissionId}, }, - db::{Connection, Pool, ReaderPool, magic::Bool}, + db::{magic::Bool, Connection, Pool, ReaderPool}, }; use futures::stream::{StreamExt as _, TryStreamExt as _}; use libsqlite3_sys as ffi; diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index 1b3c5815..738f30e0 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -106,13 +106,13 @@ pub type ChunkStream<'a> = BoxStream<'a, Result>; pub mod test { use itertools::Itertools; use libsqlite3_sys as ffi; - use sqlformat::{FormatOptions, QueryParams, format}; + use sqlformat::{format, FormatOptions, QueryParams}; use sqlx::Row; use sqlx::{QueryBuilder, Sqlite, SqliteConnection}; use super::*; - use crate::common::StrategicMetadataMap; use crate::common::chunk::ChunkSize; + use crate::common::StrategicMetadataMap; unsafe extern "C" fn sqlite_reserved_chunk_lookup_noop( context: *mut ffi::sqlite3_context,