Skip to content

Commit 8d8e607

Browse files
ReinierMaasReinierMaas
committed
Rust: edition 2024 with resolver 3
1 parent 8d03017 commit 8d8e607

26 files changed

Lines changed: 182 additions & 105 deletions

File tree

‎Cargo.toml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[workspace]
22

3-
# Resolver 2 is default in the 2021 Edition, but the workspace doesn't know that
4-
resolver = "2"
3+
# Resolver 3 is default in the 2024 Edition, but the workspace doesn't know that
4+
resolver = "3"
55

66
members = [
77
"opsqueue/",

‎libs/opsqueue_python/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[package]
22
name = "opsqueue_python"
33
version.workspace = true
4-
edition = "2021"
4+
edition = "2024"
55

66
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
77
[lib]

‎libs/opsqueue_python/src/async_util.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ use pyo3::{Bound, IntoPyObject, PyAny, PyResult, Python};
44
use pyo3_async_runtimes::TaskLocals;
55
use std::{
66
future::Future,
7-
pin::{pin, Pin},
7+
pin::{Pin, pin},
88
task::{Context, Poll},
99
};
1010

‎libs/opsqueue_python/src/common.rs‎

Lines changed: 25 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,9 @@ use std::sync::Arc;
44
use std::time::Duration;
55

66
use chrono::{DateTime, Utc};
7+
use opsqueue::common::StrategicMetadataMap;
78
use opsqueue::common::errors::TryFromIntError;
89
use opsqueue::common::submission::Metadata;
9-
use opsqueue::common::StrategicMetadataMap;
1010
use opsqueue::object_store::{ChunkRetrievalError, ChunkType, ObjectStoreClient};
1111
use opsqueue::tracing::CarrierMap;
1212
use pyo3::prelude::*;
@@ -227,7 +227,12 @@ impl Chunk {
227227
Some(bytes) => (bytes, None),
228228
None => {
229229
let prefix = s.prefix.unwrap();
230-
tracing::debug!("Fetching chunk content from object store: submission_id={}, prefix={}, chunk_index={}", c.submission_id, prefix, c.chunk_index);
230+
tracing::debug!(
231+
"Fetching chunk content from object store: submission_id={}, prefix={}, chunk_index={}",
232+
c.submission_id,
233+
prefix,
234+
c.chunk_index
235+
);
231236
let res = object_store_client
232237
.retrieve_chunk(&prefix, c.chunk_index, ChunkType::Input)
233238
.await?;
@@ -431,16 +436,30 @@ impl SubmissionCompleted {
431436
#[pymethods]
432437
impl SubmissionFailed {
433438
fn __repr__(&self) -> String {
434-
format!("SubmissionFailed(id={0}, chunks_total={1}, failed_at={2}, failed_chunk_id={3}, metadata={4:?}, strategic_metadata={5:?})",
435-
self.id.__repr__(), self.chunks_total, self.failed_at, self.failed_chunk_id, self.metadata, self.strategic_metadata)
439+
format!(
440+
"SubmissionFailed(id={0}, chunks_total={1}, failed_at={2}, failed_chunk_id={3}, metadata={4:?}, strategic_metadata={5:?})",
441+
self.id.__repr__(),
442+
self.chunks_total,
443+
self.failed_at,
444+
self.failed_chunk_id,
445+
self.metadata,
446+
self.strategic_metadata
447+
)
436448
}
437449
}
438450

439451
#[pymethods]
440452
impl SubmissionCancelled {
441453
fn __repr__(&self) -> String {
442-
format!("SubmissionCancelled(id={0}, chunks_total={1}, chunks_done={2}, metadata={3:?}, strategic_metadata={4:?}, cancelled_at={5})",
443-
self.id.__repr__(), self.chunks_total, self.chunks_done, self.metadata, self.strategic_metadata, self.cancelled_at)
454+
format!(
455+
"SubmissionCancelled(id={0}, chunks_total={1}, chunks_done={2}, metadata={3:?}, strategic_metadata={4:?}, cancelled_at={5})",
456+
self.id.__repr__(),
457+
self.chunks_total,
458+
self.chunks_done,
459+
self.metadata,
460+
self.strategic_metadata,
461+
self.cancelled_at
462+
)
444463
}
445464
}
446465

‎libs/opsqueue_python/src/consumer.rs‎

Lines changed: 31 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -2,18 +2,18 @@ use std::future::IntoFuture;
22
use std::sync::Arc;
33
use std::time::Duration;
44

5-
use futures::{stream, StreamExt, TryStreamExt};
5+
use futures::{StreamExt, TryStreamExt, stream};
66
use opsqueue::{
7+
E,
78
common::errors::{
8-
IncorrectUsage, LimitIsZero,
99
E::{self, L, R},
10+
IncorrectUsage, LimitIsZero,
1011
},
1112
consumer::client::InternalConsumerClientError,
1213
object_store::{
1314
ChunkRetrievalError, ChunkStorageError, ChunkType, NewObjectStoreClientError,
1415
ObjectStoreClient,
1516
},
16-
E,
1717
};
1818
use pyo3::{
1919
create_exception,
@@ -350,18 +350,23 @@ impl ConsumerClient {
350350
let submission_prefix = chunk.submission_prefix.clone();
351351
let chunk_index = chunk.chunk_index;
352352
tracing::debug!(
353-
"Running fun for chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
354-
submission_id,
355-
chunk_index,
356-
&submission_prefix
357-
);
358-
let res = Python::attach(|py| {
353+
"Running fun for chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
354+
submission_id,
355+
chunk_index,
356+
&submission_prefix
357+
);
358+
let res = Python::with_gil(|py| {
359359
let res = unbound_fun.bind(py).call1((chunk,))?;
360360
res.extract()
361361
});
362362
match res {
363363
Ok(res) => {
364-
tracing::debug!("Completing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}", submission_id, chunk_index, &submission_prefix);
364+
tracing::debug!(
365+
"Completing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
366+
submission_id,
367+
chunk_index,
368+
&submission_prefix
369+
);
365370
self.complete_chunk_gilless(
366371
submission_id,
367372
submission_prefix.clone(),
@@ -374,15 +379,20 @@ impl ConsumerClient {
374379
CError(R(R(e))) => CError(R(R(R(L(e))))),
375380
})?;
376381
tracing::debug!(
377-
"Completed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
378-
submission_id,
379-
chunk_index,
380-
&submission_prefix
381-
);
382+
"Completed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
383+
submission_id,
384+
chunk_index,
385+
&submission_prefix
386+
);
382387
}
383388
Err(failure) => {
384389
let failure_str = crate::common::format_pyerr(&failure);
385-
tracing::warn!("Failing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}, reason: {failure_str}", submission_id, chunk_index, &submission_prefix);
390+
tracing::warn!(
391+
"Failing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}, reason: {failure_str}",
392+
submission_id,
393+
chunk_index,
394+
&submission_prefix
395+
);
386396
self.fail_chunk_gilless(
387397
submission_id,
388398
submission_prefix.clone(),
@@ -394,11 +404,11 @@ impl ConsumerClient {
394404
CError(R(e)) => CError(R(R(R(L(e))))),
395405
})?;
396406
tracing::warn!(
397-
"Failed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
398-
submission_id,
399-
chunk_index,
400-
&submission_prefix
401-
);
407+
"Failed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
408+
submission_id,
409+
chunk_index,
410+
&submission_prefix
411+
);
402412

403413
// On exceptions that are not PyExceptions (but PyBaseExceptions), like KeyboardInterrupt etc, return.
404414
if !Python::attach(|py| failure.is_instance_of::<PyException>(py)) {

‎libs/opsqueue_python/src/errors.rs‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,11 +4,11 @@ use std::error::Error;
44

55
use opsqueue::common::chunk::ChunkId;
66
use opsqueue::common::errors::{
7-
ChunkNotFound, IncorrectUsage, SubmissionNotCancellable, SubmissionNotFound,
8-
UnexpectedOpsqueueConsumerServerResponse, E,
7+
ChunkNotFound, E, IncorrectUsage, SubmissionNotCancellable, SubmissionNotFound,
8+
UnexpectedOpsqueueConsumerServerResponse,
99
};
1010
use pyo3::exceptions::PyBaseException;
11-
use pyo3::{import_exception, Bound, PyErr, Python};
11+
use pyo3::{Bound, PyErr, Python, import_exception};
1212

1313
use crate::common;
1414
use crate::common::{ChunkIndex, SubmissionId};
@@ -74,7 +74,7 @@ pub struct FatalPythonException(#[from] pub PyErr);
7474

7575
impl From<CError<FatalPythonException>> for PyErr {
7676
fn from(value: CError<FatalPythonException>) -> Self {
77-
value.0 .0
77+
value.0.0
7878
}
7979
}
8080

@@ -141,7 +141,7 @@ impl From<CError<SubmissionNotCancellable>> for PyErr {
141141

142142
impl From<CError<SubmissionNotFound>> for PyErr {
143143
fn from(value: CError<SubmissionNotFound>) -> Self {
144-
let submission_id = value.0 .0;
144+
let submission_id = value.0.0;
145145
SubmissionNotFoundError::new_err(u64::from(submission_id))
146146
}
147147
}
@@ -153,15 +153,15 @@ pub struct SubmissionFailed(
153153

154154
impl From<CError<SubmissionFailed>> for PyErr {
155155
fn from(value: CError<SubmissionFailed>) -> Self {
156-
let submission: crate::common::SubmissionFailed = value.0 .0;
157-
let chunk: crate::common::ChunkFailed = value.0 .1;
156+
let submission: crate::common::SubmissionFailed = value.0.0;
157+
let chunk: crate::common::ChunkFailed = value.0.1;
158158
SubmissionFailedError::new_err((submission, chunk))
159159
}
160160
}
161161

162162
impl From<CError<crate::producer::SubmissionNotCompletedYetError>> for PyErr {
163163
fn from(value: CError<crate::producer::SubmissionNotCompletedYetError>) -> Self {
164-
let submission_id = value.0 .0;
164+
let submission_id = value.0.0;
165165
SubmissionNotCompletedYetError::new_err((value.0.to_string(), submission_id))
166166
}
167167
}
@@ -171,7 +171,7 @@ impl From<CError<ChunkNotFound>> for PyErr {
171171
let ChunkId {
172172
submission_id,
173173
chunk_index,
174-
} = value.0 .0;
174+
} = value.0.0;
175175
ChunkNotFoundError::new_err((
176176
value.0.to_string(),
177177
(

‎libs/opsqueue_python/src/producer.rs‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -7,25 +7,23 @@ use pyo3::{
77
types::PyIterator,
88
};
99

10-
use futures::{stream::BoxStream, StreamExt, TryStreamExt};
10+
use futures::{StreamExt, TryStreamExt, stream::BoxStream};
1111
use opsqueue::{
12+
E,
1213
common::errors::E::{self, L, R},
1314
common::errors::{SubmissionNotCancellable, SubmissionNotFound},
14-
object_store::{ChunksStorageError, NewObjectStoreClientError},
15-
producer::client::{Client as ActualClient, InternalProducerClientError},
16-
};
17-
use opsqueue::{
18-
common::{chunk, submission, StrategicMetadataMap},
15+
common::{StrategicMetadataMap, chunk, submission},
1916
object_store::{ChunkRetrievalError, ChunkType},
17+
object_store::{ChunksStorageError, NewObjectStoreClientError},
2018
producer::ChunkContents,
19+
producer::client::{Client as ActualClient, InternalProducerClientError},
2120
tracing::CarrierMap,
22-
E,
2321
};
2422
use ux::u63;
2523

2624
use crate::{
2725
async_util,
28-
common::{run_unless_interrupted, start_runtime, SubmissionId, SubmissionStatus},
26+
common::{SubmissionId, SubmissionStatus, run_unless_interrupted, start_runtime},
2927
errors::{self, CError, CPyResult, FatalPythonException},
3028
};
3129

‎opsqueue/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[package]
22
name = "opsqueue"
33
version.workspace = true
4-
edition = "2021"
4+
edition = "2024"
55
description = "lightweight batch processing queue for heavy loads"
66
repository = "https://github.com/channable/opsqueue"
77
license = "MIT"

‎opsqueue/app/main.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler, SdkTracerProvider};
77
use opsqueue::{common::submission::db::periodically_cleanup_old, config::Config, prometheus};
88
use std::{
99
error::Error,
10-
sync::{atomic::AtomicBool, Arc},
10+
sync::{Arc, atomic::AtomicBool},
1111
time::Duration,
1212
};
1313
use tokio_util::sync::CancellationToken;

‎opsqueue/src/common/chunk.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -218,11 +218,11 @@ impl Chunk {
218218
#[cfg(feature = "server-logic")]
219219
pub mod db {
220220
use super::*;
221-
use crate::common::errors::{ChunkNotFound, DatabaseError, SubmissionNotFound, E};
221+
use crate::common::errors::{ChunkNotFound, DatabaseError, E, SubmissionNotFound};
222222
use crate::db::{Connection, True, WriterConnection};
223223
use axum_prometheus::metrics::{counter, gauge};
224-
use sqlx::{query, query_as};
225224
use sqlx::{QueryBuilder, Sqlite};
225+
use sqlx::{query, query_as};
226226

227227
impl<'q> sqlx::Encode<'q, Sqlite> for super::ChunkIndex {
228228
fn encode_by_ref(
@@ -597,9 +597,9 @@ pub mod db {
597597
#[cfg(test)]
598598
#[cfg(feature = "server-logic")]
599599
pub mod test {
600+
use crate::common::StrategicMetadataMap;
600601
use crate::common::submission::db::insert_submission_raw;
601602
use crate::common::submission::{Submission, SubmissionStatus};
602-
use crate::common::StrategicMetadataMap;
603603
use crate::db::{Connection as _, WriterPool};
604604

605605
use super::db::*;

0 commit comments

Comments
 (0)