From 0a140edd98fc2cdb73e0f81fc877b72326d13bb3 Mon Sep 17 00:00:00 2001 From: Brandur Date: Wed, 7 Oct 2026 00:02:25 -0500 Subject: [PATCH] Rename `WorkerRegistry` -> `Worker` Renames: * `WorkerRegistry` -> `Worker` * `WorkerRegistry#register` -> `Worker#add` * `WorkerRegistry#register_fn` -> `Worker#add_fn` Another one to try and bring Rust more in line with Go naming conventions (which have also been copied into other languages). I believe one of our earlier versions of Go used `WorkerRegistry` and an explicit decision was made to rename to `Workers` instead. --- rust/CHANGELOG.md | 6 +- rust/riverqueue-cli/src/bench.rs | 6 +- rust/riverqueue/README.md | 12 ++-- rust/riverqueue/examples/basic_worker.rs | 8 +-- rust/riverqueue/examples/cancellation.rs | 8 +-- rust/riverqueue/examples/events.rs | 6 +- rust/riverqueue/examples/graceful_shutdown.rs | 7 +- rust/riverqueue/examples/mixed_go_rust.rs | 8 +-- rust/riverqueue/examples/periodic.rs | 8 +-- rust/riverqueue/examples/sqlite.rs | 6 +- rust/riverqueue/examples/transactions.rs | 6 +- rust/riverqueue/src/client/builder.rs | 8 +-- rust/riverqueue/src/client/mod.rs | 6 +- rust/riverqueue/src/client/tests.rs | 8 +-- rust/riverqueue/src/lib.rs | 2 +- rust/riverqueue/src/maintenance/tests.rs | 12 ++-- rust/riverqueue/src/worker.rs | 70 ++++++++++--------- rust/riverqueue/tests/client_handles.rs | 20 +++--- rust/riverqueue/tests/extension_services.rs | 6 +- .../tests/fetch_only_known_kinds.rs | 6 +- rust/riverqueue/tests/insert_notifications.rs | 6 +- .../tests/leader_election_disabled.rs | 8 +-- rust/riverqueue/tests/lifecycle.rs | 8 +-- rust/riverqueue/tests/metadata_exact.rs | 6 +- rust/riverqueue/tests/peer_attempts.rs | 10 +-- .../tests/poll_only_cancellation.rs | 6 +- rust/riverqueue/tests/postgres.rs | 42 ++++++----- rust/riverqueue/tests/postgres_parity.rs | 8 +-- rust/riverqueue/tests/producer_lifetime.rs | 8 +-- rust/riverqueue/tests/producer_sessions.rs | 16 ++--- rust/riverqueue/tests/resilience_postgres.rs | 24 +++---- rust/riverqueue/tests/resilience_sqlite.rs | 14 ++-- rust/riverqueue/tests/runtime_config.rs | 30 ++++---- rust/riverqueue/tests/sqlite_runtime.rs | 40 +++++------ rust/riverqueue/tests/stuck_jobs.rs | 7 +- rust/riverqueue/tests/work_extensions.rs | 8 +-- rust/riverqueue/tests/yugabyte.rs | 6 +- 37 files changed, 236 insertions(+), 230 deletions(-) diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 81e43aac8..b760cda0d 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -10,11 +10,15 @@ Changes to River for Go are recorded in the [repository changelog](../CHANGELOG. ## [Unreleased] +### Changed + +- **Breaking:** Renamed `WorkerRegistry` to `Workers`, `register` to `add`, and `register_fn` to `add_fn` to align worker registration with Go's naming. Both methods retain their `Result` return type and registration behavior. [PR #1469](https://github.com/riverqueue/river/pull/1469). + ## [0.2.0] - 2026-10-06 ### Changed -- Renamed `Client::start_with_graceful_shutdown` to `Client::start_with_graceful_stop` to match River's standard start/stop terminology. This is a rare breaking name change as the new Rust API stabilizes. [PR #1467](https://github.com/riverqueue/river/pull/1467). +- **Breaking:** Renamed `Client::start_with_graceful_shutdown` to `Client::start_with_graceful_stop` to match River's standard start/stop terminology. This is a rare breaking name change as the new Rust API stabilizes. [PR #1467](https://github.com/riverqueue/river/pull/1467). ## [0.1.0] - 2026-10-06 diff --git a/rust/riverqueue-cli/src/bench.rs b/rust/riverqueue-cli/src/bench.rs index 8d68be49c..ed5c40970 100644 --- a/rust/riverqueue-cli/src/bench.rs +++ b/rust/riverqueue-cli/src/bench.rs @@ -13,7 +13,7 @@ use std::{ use riverqueue::{ Client, EventKind, EventReceiver, EventRecvError, InsertOpts, Job, JobArgs, QueueConfig, - SubscribeConfig, WorkContext, WorkOutcome, Worker, WorkerRegistry, database::SchemaName, + SubscribeConfig, WorkContext, WorkOutcome, Worker, Workers, database::SchemaName, }; use serde::{Deserialize, Serialize}; use sqlx::{ @@ -467,8 +467,8 @@ async fn run_benchmark(options: BenchOptions) -> Result<(), Box Result { - let mut workers = WorkerRegistry::new(); - workers.register::(BenchmarkWorker)?; + let mut workers = Workers::new(); + workers.add::(BenchmarkWorker)?; Client::builder(pool) .id("riverqueue-benchmark") .workers(workers) diff --git a/rust/riverqueue/README.md b/rust/riverqueue/README.md index e9fafc6bb..bbe5e305a 100644 --- a/rust/riverqueue/README.md +++ b/rust/riverqueue/README.md @@ -49,7 +49,7 @@ apply River's migrations, and start a client: use riverqueue::migrate::PostgresMigrator; use riverqueue::sqlx::PgPool; use riverqueue::{ - BoxError, Client, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, WorkerRegistry, + BoxError, Client, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; @@ -75,8 +75,8 @@ async fn main() -> Result<(), Box> { // `riverqueue migrate-up` from `riverqueue-cli` at deploy time instead. PostgresMigrator::new(pool.clone()).migrate_up().await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(send_email)?; + let mut workers = Workers::new(); + workers.add_fn(send_email)?; let client = Client::builder(pool) .workers(workers) @@ -229,8 +229,12 @@ completes the job. After the configured stuck threshold, River can abort a Tokio task that yields, which fails its attempt, but it can't stop CPU-bound work or a blocking call already in progress. +Collect workers in [`Workers`], the equivalent of Go's `Workers` bundle. +Use [`Workers::add`] to add a [`Worker`] implementation, like Go's `AddWorker`. Implement [`Worker`] when a kind needs a custom timeout or next-retry decision. -Use `WorkerRegistry::register_fn` for an async function or capturing closure. +Use [`Workers::add_fn`] for an async function or capturing closure; it combines +Go's `WorkFunc` adapter and registration in one call. Both methods return +`Result` so registration errors can be propagated with `?`. ## Events diff --git a/rust/riverqueue/examples/basic_worker.rs b/rust/riverqueue/examples/basic_worker.rs index ba15bbb6d..8f26e3bb2 100644 --- a/rust/riverqueue/examples/basic_worker.rs +++ b/rust/riverqueue/examples/basic_worker.rs @@ -8,8 +8,8 @@ use std::error::Error; use riverqueue::sqlx::PgPool; use riverqueue::{ - BoxError, Client, EventKind, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, - WorkerRegistry, migrate::PostgresMigrator, + BoxError, Client, EventKind, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Workers, + migrate::PostgresMigrator, }; use serde::{Deserialize, Serialize}; @@ -30,8 +30,8 @@ async fn main() -> Result<(), Box> { let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?; // Apply River's schema before starting a client. PostgresMigrator::new(pool.clone()).migrate_up().await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(send_email)?; + let mut workers = Workers::new(); + workers.add_fn(send_email)?; let client = Client::builder(pool) .workers(workers) .queue("default", QueueConfig::new(10)) diff --git a/rust/riverqueue/examples/cancellation.rs b/rust/riverqueue/examples/cancellation.rs index 5f41656ff..f7bc915e9 100644 --- a/rust/riverqueue/examples/cancellation.rs +++ b/rust/riverqueue/examples/cancellation.rs @@ -11,8 +11,8 @@ use std::{error::Error, time::Duration}; use riverqueue::sqlx::PgPool; use riverqueue::{ - Client, Job, JobArgs, QueueConfig, WorkCancelled, WorkContext, WorkOutcome, Worker, - WorkerRegistry, migrate::PostgresMigrator, + Client, Job, JobArgs, QueueConfig, WorkCancelled, WorkContext, WorkOutcome, Worker, Workers, + migrate::PostgresMigrator, }; use serde::{Deserialize, Serialize}; @@ -49,8 +49,8 @@ impl Worker for CancellableReportWorker { async fn main() -> Result<(), Box> { let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?; PostgresMigrator::new(pool.clone()).migrate_up().await?; - let mut workers = WorkerRegistry::new(); - workers.register(CancellableReportWorker)?; + let mut workers = Workers::new(); + workers.add(CancellableReportWorker)?; let client = Client::builder(pool) .workers(workers) .queue("default", QueueConfig::new(1)) diff --git a/rust/riverqueue/examples/events.rs b/rust/riverqueue/examples/events.rs index ff9d7b603..f3a690f2c 100644 --- a/rust/riverqueue/examples/events.rs +++ b/rust/riverqueue/examples/events.rs @@ -12,7 +12,7 @@ use std::error::Error; use riverqueue::sqlx::PgPool; use riverqueue::{ Client, Event, EventKind, InsertOpts, Job, JobArgs, JobEventKind, QueueConfig, WorkContext, - WorkOutcome, WorkerRegistry, migrate::PostgresMigrator, + WorkOutcome, Workers, migrate::PostgresMigrator, }; use serde::{Deserialize, Serialize}; @@ -38,8 +38,8 @@ async fn main() -> Result<(), Box> { let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?; PostgresMigrator::new(pool.clone()).migrate_up().await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(charge_card)?; + let mut workers = Workers::new(); + workers.add_fn(charge_card)?; let client = Client::builder(pool) .workers(workers) .queue("default", QueueConfig::new(4)) diff --git a/rust/riverqueue/examples/graceful_shutdown.rs b/rust/riverqueue/examples/graceful_shutdown.rs index 645dddb22..4ae268ff5 100644 --- a/rust/riverqueue/examples/graceful_shutdown.rs +++ b/rust/riverqueue/examples/graceful_shutdown.rs @@ -5,8 +5,7 @@ use std::{error::Error, time::Duration}; use riverqueue::sqlx::PgPool; use riverqueue::{ - BoxError, Client, Job, JobArgs, QueueConfig, WorkCancelled, WorkContext, WorkOutcome, - WorkerRegistry, + BoxError, Client, Job, JobArgs, QueueConfig, WorkCancelled, WorkContext, WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; @@ -35,8 +34,8 @@ async fn generate_report( #[tokio::main] async fn main() -> Result<(), Box> { let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(generate_report)?; + let mut workers = Workers::new(); + workers.add_fn(generate_report)?; let client = Client::builder(pool) .workers(workers) .queue("default", QueueConfig::new(10)) diff --git a/rust/riverqueue/examples/mixed_go_rust.rs b/rust/riverqueue/examples/mixed_go_rust.rs index f7f12f68b..1e622de1c 100644 --- a/rust/riverqueue/examples/mixed_go_rust.rs +++ b/rust/riverqueue/examples/mixed_go_rust.rs @@ -45,8 +45,8 @@ use std::error::Error; use riverqueue::sqlx::PgPool; use riverqueue::{ - Client, EventKind, InsertOpts, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, - WorkerRegistry, migrate::PostgresMigrator, + Client, EventKind, InsertOpts, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Workers, + migrate::PostgresMigrator, }; use serde::{Deserialize, Serialize}; @@ -86,8 +86,8 @@ async fn main() -> Result<(), Box> { let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?; PostgresMigrator::new(pool.clone()).migrate_up().await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(resize_image)?; + let mut workers = Workers::new(); + workers.add_fn(resize_image)?; let client = Client::builder(pool) .workers(workers) // Only Rust's queue: Rust never fetches the Go service's jobs. diff --git a/rust/riverqueue/examples/periodic.rs b/rust/riverqueue/examples/periodic.rs index 50476e81d..e42d600a9 100644 --- a/rust/riverqueue/examples/periodic.rs +++ b/rust/riverqueue/examples/periodic.rs @@ -12,7 +12,7 @@ use std::{error::Error, time::Duration}; use riverqueue::sqlx::PgPool; use riverqueue::{ Client, CronSchedule, EventKind, IntervalSchedule, Job, JobArgs, PeriodicJob, PeriodicJobOpts, - QueueConfig, WorkContext, WorkOutcome, WorkerRegistry, migrate::PostgresMigrator, + QueueConfig, WorkContext, WorkOutcome, Workers, migrate::PostgresMigrator, }; use serde::{Deserialize, Serialize}; @@ -45,9 +45,9 @@ async fn main() -> Result<(), Box> { let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?; PostgresMigrator::new(pool.clone()).migrate_up().await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(refresh_cache)?; - workers.register_fn(nightly_report)?; + let mut workers = Workers::new(); + workers.add_fn(refresh_cache)?; + workers.add_fn(nightly_report)?; let client = Client::builder(pool) .workers(workers) .queue("default", QueueConfig::new(4)) diff --git a/rust/riverqueue/examples/sqlite.rs b/rust/riverqueue/examples/sqlite.rs index 9716db67a..ec02fe56d 100644 --- a/rust/riverqueue/examples/sqlite.rs +++ b/rust/riverqueue/examples/sqlite.rs @@ -11,7 +11,7 @@ use std::{error::Error, str::FromStr, time::Duration}; use riverqueue::sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions}; use riverqueue::{ - Client, EventKind, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, WorkerRegistry, + Client, EventKind, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Workers, migrate::SqliteMigrator, }; use serde::{Deserialize, Serialize}; @@ -49,8 +49,8 @@ async fn main() -> Result<(), Box> { // Apply River's schema before starting a client. SqliteMigrator::new(pool.clone()).migrate_up().await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(resize_image)?; + let mut workers = Workers::new(); + workers.add_fn(resize_image)?; let client = Client::builder(pool.clone()) .workers(workers) .queue("default", QueueConfig::new(4)) diff --git a/rust/riverqueue/examples/transactions.rs b/rust/riverqueue/examples/transactions.rs index 31ccd4a44..50e0dd383 100644 --- a/rust/riverqueue/examples/transactions.rs +++ b/rust/riverqueue/examples/transactions.rs @@ -13,7 +13,7 @@ use std::error::Error; use riverqueue::sqlx::{self, PgPool}; use riverqueue::{ - Client, EventKind, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, WorkerRegistry, + Client, EventKind, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Workers, migrate::PostgresMigrator, }; use serde::{Deserialize, Serialize}; @@ -63,8 +63,8 @@ async fn main() -> Result<(), Box> { .execute(&pool) .await?; - let mut workers = WorkerRegistry::new(); - workers.register_fn(confirm_order)?; + let mut workers = Workers::new(); + workers.add_fn(confirm_order)?; let client = Client::builder(pool.clone()) .workers(workers) .queue("default", QueueConfig::new(4)) diff --git a/rust/riverqueue/src/client/builder.rs b/rust/riverqueue/src/client/builder.rs index a53c20565..17aae2e73 100644 --- a/rust/riverqueue/src/client/builder.rs +++ b/rust/riverqueue/src/client/builder.rs @@ -18,7 +18,7 @@ use crate::database::Database; use crate::periodic::{PeriodicJob, PeriodicJobs}; use crate::{ Client, Error, ErrorHandler, FETCH_COOLDOWN_MIN, FETCH_POLL_INTERVAL_DEFAULT, Hook, - InsertMiddleware, Plugin, QUEUE_NUM_WORKERS_MAX, RetryPolicy, WorkMiddleware, WorkerRegistry, + InsertMiddleware, Plugin, QUEUE_NUM_WORKERS_MAX, RetryPolicy, WorkMiddleware, Workers, }; /// Default age at which running jobs are rescued (Go @@ -362,7 +362,7 @@ pub struct ClientBuilder { pub(super) retry_policy: Arc, pub(super) soft_stop_timeout: Option, pub(super) work_middleware: Vec>, - pub(crate) workers: WorkerRegistry, + pub(crate) workers: Workers, } impl std::fmt::Debug for ClientBuilder { @@ -642,9 +642,9 @@ impl ClientBuilder { self } - /// Installs a typed worker registry. + /// Sets the client's workers. #[must_use] - pub fn workers(mut self, workers: WorkerRegistry) -> Self { + pub fn workers(mut self, workers: Workers) -> Self { self.workers = workers; self } diff --git a/rust/riverqueue/src/client/mod.rs b/rust/riverqueue/src/client/mod.rs index eebe80877..70d321102 100644 --- a/rust/riverqueue/src/client/mod.rs +++ b/rust/riverqueue/src/client/mod.rs @@ -72,7 +72,7 @@ use crate::maintenance::LeadershipWakeup; use crate::{ DefaultRetryPolicy, Error, Event, EventKind, EventReceiver, FETCH_COOLDOWN_DEFAULT, JOB_STUCK_THRESHOLD_DEFAULT, JOB_TIMEOUT_DEFAULT, MAX_ATTEMPTS_DEFAULT, RetryPolicy, - SubscribeConfig, WorkerRegistry, + SubscribeConfig, Workers, database::{ClientDatabase, Database, DatabasePool, DatabaseTransactionExecutor, IntoDatabase}, periodic::PeriodicJobs, }; @@ -154,7 +154,7 @@ pub(crate) struct ClientInner { soft_stop_timeout: Option, started: AtomicBool, work_middleware: Vec>, - pub(crate) workers: WorkerRegistry, + pub(crate) workers: Workers, } #[cfg(feature = "sqlite")] @@ -322,7 +322,7 @@ impl Client { retry_policy: Arc::new(DefaultRetryPolicy::default()), soft_stop_timeout: None, work_middleware: Vec::new(), - workers: WorkerRegistry::new(), + workers: Workers::new(), } } diff --git a/rust/riverqueue/src/client/tests.rs b/rust/riverqueue/src/client/tests.rs index 35ed8bbae..bf3dd6eef 100644 --- a/rust/riverqueue/src/client/tests.rs +++ b/rust/riverqueue/src/client/tests.rs @@ -479,9 +479,9 @@ async fn subscription_forwarder_stops_when_the_receiver_drops() { #[river(kind = "subscription_forwarder_test")] struct ForwarderArgs {} - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, std::convert::Infallible>(WorkOutcome::Complete) }) .unwrap(); @@ -674,9 +674,9 @@ async fn readiness_survives_a_notification_listener_panic() { .migrate_up() .await .unwrap(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, std::convert::Infallible>(WorkOutcome::Complete) }) .unwrap(); diff --git a/rust/riverqueue/src/lib.rs b/rust/riverqueue/src/lib.rs index 24455ee7e..ede0cab2d 100644 --- a/rust/riverqueue/src/lib.rs +++ b/rust/riverqueue/src/lib.rs @@ -104,7 +104,7 @@ pub use sqlx; /// The `tokio-util` version of [`WorkContext::cancellation_token`]'s /// [`CancellationToken`](tokio_util::sync::CancellationToken). pub use tokio_util; -pub use worker::{WorkContext, WorkOutcome, Worker, WorkerRegistry, WorkerTimeout}; +pub use worker::{WorkContext, WorkOutcome, Worker, WorkerTimeout, Workers}; /// Default maximum number of attempts for a job. pub const MAX_ATTEMPTS_DEFAULT: i16 = 25; diff --git a/rust/riverqueue/src/maintenance/tests.rs b/rust/riverqueue/src/maintenance/tests.rs index 847b380c5..d390001c5 100644 --- a/rust/riverqueue/src/maintenance/tests.rs +++ b/rust/riverqueue/src/maintenance/tests.rs @@ -32,7 +32,7 @@ use super::{ }; use crate::{ Client, Job, JobArgs, JobState, MaintenanceConfig, QueueConfig, SchemaName, UniqueOpts, - WorkContext, WorkOutcome, Worker, WorkerRegistry, WorkerTimeout, + WorkContext, WorkOutcome, Worker, WorkerTimeout, Workers, database::{PostgresDatabase, PostgresReindexConfig}, }; @@ -80,13 +80,11 @@ impl Worker for ShortTimeoutWorker { } } -fn workers() -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn workers() -> Workers { + let mut workers = Workers::new(); + workers.add::(NoTimeoutWorker).unwrap(); workers - .register::(NoTimeoutWorker) - .unwrap(); - workers - .register::(ShortTimeoutWorker) + .add::(ShortTimeoutWorker) .unwrap(); workers } diff --git a/rust/riverqueue/src/worker.rs b/rust/riverqueue/src/worker.rs index 234db9906..1b0c2b486 100644 --- a/rust/riverqueue/src/worker.rs +++ b/rust/riverqueue/src/worker.rs @@ -820,23 +820,27 @@ where } } -/// Type-erased collection of workers keyed by job kind. +/// Collection of available job workers, keyed by job kind. +/// +/// Add a [`Worker`] with [`Self::add`], or an async function or closure with +/// [`Self::add_fn`], then pass the collection to +/// [`ClientBuilder::workers`](crate::ClientBuilder::workers). #[derive(Clone, Default)] -pub struct WorkerRegistry { +pub struct Workers { workers: HashMap<&'static str, Arc>, } -impl std::fmt::Debug for WorkerRegistry { +impl std::fmt::Debug for Workers { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter - .debug_struct("WorkerRegistry") + .debug_struct("Workers") .field("kinds", &self.kinds()) .finish_non_exhaustive() } } -impl WorkerRegistry { - /// Creates an empty worker registry. +impl Workers { + /// Creates an empty collection of workers. #[must_use] pub fn new() -> Self { Self::default() @@ -872,14 +876,14 @@ impl WorkerRegistry { self.worker_for(row)?.timeout(row) } - /// Registers one worker, rejecting duplicate kinds. + /// Adds one worker, rejecting duplicate kinds. /// /// # Errors /// /// Returns [`Error::InvalidJob`] when the kind or one of its aliases is /// empty or 128 bytes or longer, or when a worker is already registered /// for it. - pub fn register(&mut self, worker: W) -> Result<&mut Self, Error> + pub fn add(&mut self, worker: W) -> Result<&mut Self, Error> where A: JobArgs, W: Worker, @@ -925,7 +929,7 @@ impl WorkerRegistry { Ok(self) } - /// Registers an asynchronous function or closure as a worker. + /// Adds an asynchronous function or closure as a worker. /// /// The function may return any error convertible into [`BoxError`], such /// as `anyhow::Result`; see [`Worker::Error`]. Use [`Worker`] @@ -936,14 +940,14 @@ impl WorkerRegistry { /// /// Returns an error when the job kind or one of its aliases is invalid or /// already registered. - pub fn register_fn(&mut self, function: F) -> Result<&mut Self, Error> + pub fn add_fn(&mut self, function: F) -> Result<&mut Self, Error> where A: JobArgs, E: Into, F: Fn(WorkContext, Job) -> Fut + Send + Sync + 'static, Fut: Future> + Send, { - self.register::(FunctionWorker { function }) + self.add::(FunctionWorker { function }) } /// Returns an error for a row whose kind has no registered worker, which @@ -1212,12 +1216,12 @@ mod tests { } #[tokio::test] - async fn register_fn_accepts_capturing_closure() { + async fn add_fn_accepts_capturing_closure() { let calls = Arc::new(AtomicUsize::new(0)); let calls_for_worker = Arc::clone(&calls); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let calls = Arc::clone(&calls_for_worker); async move { calls.fetch_add(1, Ordering::Relaxed); @@ -1242,9 +1246,9 @@ mod tests { } #[tokio::test] - async fn register_fn_handles_aliases_and_typed_errors() { - let mut workers = WorkerRegistry::new(); - workers.register_fn(function_worker).unwrap(); + async fn add_fn_handles_aliases_and_typed_errors() { + let mut workers = Workers::new(); + workers.add_fn(function_worker).unwrap(); assert_eq!( workers.kinds(), @@ -1310,8 +1314,8 @@ mod tests { #[tokio::test] async fn work_decodes_args_once_and_reports_timeout_first() { - let mut workers = WorkerRegistry::new(); - workers.register(CountedWorker).unwrap(); + let mut workers = Workers::new(); + workers.add(CountedWorker).unwrap(); let mut row = job_row(CountedArgs::KIND, false); row.encoded_args = serde_json::value::to_raw_value(&json!({"timeout_ms": 1234})).unwrap(); @@ -1336,8 +1340,8 @@ mod tests { #[tokio::test] async fn work_with_undecodable_args_fails_without_reporting_timeout() { - let mut workers = WorkerRegistry::new(); - workers.register_fn(function_worker).unwrap(); + let mut workers = Workers::new(); + workers.add_fn(function_worker).unwrap(); let mut row = job_row(FunctionJobArgs::KIND, false); row.encoded_args = serde_json::value::to_raw_value(&json!({"fail": "no"})).unwrap(); @@ -1355,7 +1359,7 @@ mod tests { assert!(timeout_receiver.await.is_err()); } - async fn run_once(workers: &WorkerRegistry, row: &JobRow) -> Result { + async fn run_once(workers: &Workers, row: &JobRow) -> Result { let (timeout_sender, _timeout_receiver) = oneshot::channel(); workers .work( @@ -1380,9 +1384,9 @@ mod tests { } #[tokio::test] - async fn register_fn_accepts_anyhow_results() { - let mut workers = WorkerRegistry::new(); - workers.register_fn(anyhow_function_worker).unwrap(); + async fn add_fn_accepts_anyhow_results() { + let mut workers = Workers::new(); + workers.add_fn(anyhow_function_worker).unwrap(); assert_eq!( run_once(&workers, &job_row(FunctionJobArgs::KIND, false)) @@ -1420,8 +1424,8 @@ mod tests { #[tokio::test] async fn boxed_worker_errors_remain_downcastable() { - let mut workers = WorkerRegistry::new(); - workers.register(BoxedErrorWorker).unwrap(); + let mut workers = Workers::new(); + workers.add(BoxedErrorWorker).unwrap(); let error = run_once(&workers, &job_row(FunctionJobArgs::KIND, false)) .await @@ -1492,11 +1496,11 @@ mod tests { } #[test] - fn register_fn_rejects_duplicate_kinds() { - let mut workers = WorkerRegistry::new(); - workers.register_fn(function_worker).unwrap(); + fn add_fn_rejects_duplicate_kinds() { + let mut workers = Workers::new(); + workers.add_fn(function_worker).unwrap(); - let Err(error) = workers.register_fn(function_worker) else { + let Err(error) = workers.add_fn(function_worker) else { panic!("duplicate registration should fail"); }; @@ -1505,9 +1509,9 @@ mod tests { #[test] fn registry_debug_lists_kinds_without_worker_internals() { - let mut registry = WorkerRegistry::new(); + let mut registry = Workers::new(); registry - .register_fn( + .add_fn( |_context: WorkContext, _job: Job| async move { Ok::<_, std::io::Error>(WorkOutcome::Complete) }, diff --git a/rust/riverqueue/tests/client_handles.rs b/rust/riverqueue/tests/client_handles.rs index cba5ffc13..d01902cba 100644 --- a/rust/riverqueue/tests/client_handles.rs +++ b/rust/riverqueue/tests/client_handles.rs @@ -14,7 +14,7 @@ use riverqueue::{ Client, Error, EventKind, InsertBatch, InsertContext, InsertMiddleware, InsertNext, InsertOpts, InsertedJob, Job, JobArgs, JobDeleteManyParams, JobListParams, JobState, JobUpdateParams, QueueConfig, QueueListParams, QueueSelector, QueueUpdateParams, UniqueOpts, WorkContext, - WorkOutcome, WorkerRegistry, + WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; @@ -108,9 +108,9 @@ macro_rules! scenarios { async fn cancelled_job_that_succeeds_is_completed() { let fixture = Fixture::new().await; let (started_sender, mut started) = tokio::sync::mpsc::unbounded_channel(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let started_sender = started_sender.clone(); async move { let _ = started_sender.send(job.id()); @@ -389,10 +389,10 @@ macro_rules! scenarios { ids.push(id); } let worked = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); let recorder = std::sync::Arc::clone(&worked); workers - .register_fn(move |_context: WorkContext, job: Job| { + .add_fn(move |_context: WorkContext, job: Job| { let recorder = std::sync::Arc::clone(&recorder); async move { recorder.lock().unwrap().push(job.id()); @@ -423,9 +423,9 @@ macro_rules! scenarios { async fn cancel_reaches_a_running_job_on_a_poll_only_client() { let fixture = Fixture::new().await; let (started_sender, mut started) = tokio::sync::mpsc::unbounded_channel(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let started_sender = started_sender.clone(); async move { let _ = started_sender.send(job.id()); @@ -867,10 +867,10 @@ macro_rules! scenarios { }; } -fn workers() -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn workers() -> Workers { + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); diff --git a/rust/riverqueue/tests/extension_services.rs b/rust/riverqueue/tests/extension_services.rs index 190dcaced..5b6604c6b 100644 --- a/rust/riverqueue/tests/extension_services.rs +++ b/rust/riverqueue/tests/extension_services.rs @@ -31,7 +31,7 @@ use riverqueue::__private::{ }; use riverqueue::{ Client, Job, JobArgs, JobRow, JobState, MaintenanceConfig, QueueConfig, WorkContext, - WorkOutcome, WorkerRegistry, + WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; use tokio::sync::{Notify, Semaphore}; @@ -205,10 +205,10 @@ fn client( pilot: &ServicePilot, release: &Arc, ) -> Client { - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); let release = Arc::clone(release); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let release = Arc::clone(&release); async move { release.acquire().await.unwrap().forget(); diff --git a/rust/riverqueue/tests/fetch_only_known_kinds.rs b/rust/riverqueue/tests/fetch_only_known_kinds.rs index d16970908..4ff54524f 100644 --- a/rust/riverqueue/tests/fetch_only_known_kinds.rs +++ b/rust/riverqueue/tests/fetch_only_known_kinds.rs @@ -11,7 +11,7 @@ use std::{collections::HashSet, convert::Infallible, time::Duration}; use riverqueue::{ __private::Database, Client, EventKind, Job, JobArgs, JobState, QueueConfig, WorkContext, - WorkOutcome, WorkerRegistry, + WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; @@ -71,9 +71,9 @@ async fn claims_only_registered_kinds(backend: Backend) { let known = inserter.insert(KnownArgs {}).await.unwrap().job.row.id; let alias = inserter.insert(AliasArgs {}).await.unwrap().job.row.id; - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); diff --git a/rust/riverqueue/tests/insert_notifications.rs b/rust/riverqueue/tests/insert_notifications.rs index cc75475ce..189c14d94 100644 --- a/rust/riverqueue/tests/insert_notifications.rs +++ b/rust/riverqueue/tests/insert_notifications.rs @@ -47,13 +47,13 @@ fn many(queues: &[&str]) -> Vec> { #[cfg(feature = "sqlite")] #[tokio::test] async fn fetch_cooldown_validates_against_queue_poll_intervals() { - use riverqueue::{Error, QueueConfig, WorkContext, WorkOutcome, WorkerRegistry}; + use riverqueue::{Error, QueueConfig, WorkContext, WorkOutcome, Workers}; let (pool, path) = support::sqlite_file_pool(1).await; let workers = || { - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn( + .add_fn( |_context: WorkContext, _job: riverqueue::Job| async { Ok::<_, std::convert::Infallible>(WorkOutcome::Complete) }, diff --git a/rust/riverqueue/tests/leader_election_disabled.rs b/rust/riverqueue/tests/leader_election_disabled.rs index 07d68c87d..35f28467d 100644 --- a/rust/riverqueue/tests/leader_election_disabled.rs +++ b/rust/riverqueue/tests/leader_election_disabled.rs @@ -24,7 +24,7 @@ use riverqueue::{ }, Client, ClientBuilder, EventKind, EventReceiver, Job, JobArgs, JobRow, JobState, MaintenanceConfig, NeverSchedule, PeriodicJob, PeriodicJobOpts, QueueConfig, WorkContext, - WorkOutcome, WorkerRegistry, + WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; use tokio_util::sync::CancellationToken; @@ -40,10 +40,10 @@ const ELECT_INTERVAL: Duration = Duration::from_millis(50); #[river(kind = "rust_leader_election_disabled_noop")] struct NoopArgs {} -fn noop_workers() -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn noop_workers() -> Workers { + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); diff --git a/rust/riverqueue/tests/lifecycle.rs b/rust/riverqueue/tests/lifecycle.rs index 842409182..4e5a74bf2 100644 --- a/rust/riverqueue/tests/lifecycle.rs +++ b/rust/riverqueue/tests/lifecycle.rs @@ -15,7 +15,7 @@ use std::{ use riverqueue::{ Client, Job, JobArgs, JobState, QueueConfig, RunHandle, WorkCancelled, WorkContext, - WorkOutcome, WorkerRegistry, + WorkOutcome, Workers, }; use riverqueue_migrate::SqliteMigrator; use serde::{Deserialize, Serialize}; @@ -75,11 +75,11 @@ impl Gate { .forget(); } - fn workers(&self) -> WorkerRegistry { + fn workers(&self) -> Workers { let gate = self.clone(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let gate = gate.clone(); async move { gate.started.add_permits(1); diff --git a/rust/riverqueue/tests/metadata_exact.rs b/rust/riverqueue/tests/metadata_exact.rs index ab7076713..accb6868e 100644 --- a/rust/riverqueue/tests/metadata_exact.rs +++ b/rust/riverqueue/tests/metadata_exact.rs @@ -6,7 +6,7 @@ mod support; use riverqueue::{Client, InsertOpts, JobArgs, JobMetadata, JobRow, JobUpdateParams}; #[cfg(feature = "sqlite")] -use riverqueue::{Job, JobState, QueueConfig, WorkContext, WorkOutcome, WorkerRegistry}; +use riverqueue::{Job, JobState, QueueConfig, WorkContext, WorkOutcome, Workers}; use serde::{Deserialize, Serialize}; #[cfg(feature = "sqlite")] use std::{convert::Infallible, time::Duration}; @@ -250,9 +250,9 @@ async fn sqlite_reads_metadata_with_large_numbers() { #[tokio::test] async fn sqlite_snooze_preserves_large_metadata_numbers() { let (pool, path) = support::sqlite_file_pool(4).await; - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Snooze(Duration::from_hours(1))) }) .unwrap(); diff --git a/rust/riverqueue/tests/peer_attempts.rs b/rust/riverqueue/tests/peer_attempts.rs index e905440fd..d5e3c66c1 100644 --- a/rust/riverqueue/tests/peer_attempts.rs +++ b/rust/riverqueue/tests/peer_attempts.rs @@ -23,7 +23,7 @@ use riverqueue::__private::{ }; use riverqueue::{ BoxError, Client, ErrorHandler, ErrorHandlerDecision, Event, EventKind, InsertOpts, Job, - JobArgs, JobRow, JobState, QueueConfig, WorkContext, WorkOutcome, WorkResult, WorkerRegistry, + JobArgs, JobRow, JobState, QueueConfig, WorkContext, WorkOutcome, WorkResult, Workers, }; use serde::{Deserialize, Serialize}; use tokio::sync::{Notify, mpsc}; @@ -478,10 +478,10 @@ impl Run { signals: Signals::default(), }; let env_slot = Arc::new(Mutex::new(None::)); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); let worker_env = Arc::clone(&env_slot); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let env = worker_env .lock() .unwrap() @@ -491,13 +491,13 @@ impl Run { async move { Ok::<_, BoxError>(run.await) } }) .unwrap() - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, BoxError>(WorkOutcome::Complete) }) .unwrap(); let busy_env = Arc::clone(&env_slot); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let env = busy_env.lock().unwrap().clone().expect("scenario started"); async move { let released = env.busy.notified(); diff --git a/rust/riverqueue/tests/poll_only_cancellation.rs b/rust/riverqueue/tests/poll_only_cancellation.rs index c1d256200..87af53492 100644 --- a/rust/riverqueue/tests/poll_only_cancellation.rs +++ b/rust/riverqueue/tests/poll_only_cancellation.rs @@ -12,7 +12,7 @@ use std::{convert::Infallible, sync::Arc, time::Duration}; use riverqueue::{ __private::Database, Client, EventKind, Job, JobArgs, JobState, QueueConfig, WorkContext, - WorkOutcome, WorkerRegistry, + WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; use tokio::sync::Semaphore; @@ -62,10 +62,10 @@ impl Backend { async fn polls_for_remote_cancellation(backend: Backend) { for while_stopping in [false, true] { let started = Arc::new(Semaphore::new(0)); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); let worker_started = Arc::clone(&started); workers - .register_fn(move |context: WorkContext, _job: Job| { + .add_fn(move |context: WorkContext, _job: Job| { let started = Arc::clone(&worker_started); async move { started.add_permits(1); diff --git a/rust/riverqueue/tests/postgres.rs b/rust/riverqueue/tests/postgres.rs index df1abdb6f..235f07d90 100644 --- a/rust/riverqueue/tests/postgres.rs +++ b/rust/riverqueue/tests/postgres.rs @@ -22,7 +22,7 @@ use riverqueue::{ Client, EventKind, InsertBatch, InsertOpts, IntervalSchedule, Job, JobArgs, JobListOrderBy, JobListParams, JobRow, JobState, JobUpdateParams, MaintenanceConfig, PeriodicJob, PeriodicJobOpts, QueueConfig, QueueListParams, UniqueOpts, WorkContext, WorkError, WorkOutcome, - Worker, WorkerRegistry, WorkerTimeout, + Worker, WorkerTimeout, Workers, database::{PostgresDatabase, PostgresReindexConfig, PostgresReindexSchedule}, }; use riverqueue_migrate::{Direction, MigrateOpts}; @@ -1262,8 +1262,8 @@ async fn maintenance_cleans_old_jobs_and_queues_and_reindexes() { .await .unwrap(); - let mut cleanup_workers = WorkerRegistry::new(); - cleanup_workers.register::(EchoWorker).unwrap(); + let mut cleanup_workers = Workers::new(); + cleanup_workers.add::(EchoWorker).unwrap(); let cleanup_client = Client::builder( PostgresDatabase::new(pool.clone()).with_reindex( PostgresReindexConfig::default() @@ -1615,18 +1615,18 @@ async fn rescuer_honors_worker_timeout_and_retry_overrides() { .await .unwrap(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register::(RescueDefaultTimeoutWorker) + .add::(RescueDefaultTimeoutWorker) .unwrap(); workers - .register::(RescueDisabledTimeoutWorker) + .add::(RescueDisabledTimeoutWorker) .unwrap(); workers - .register::(RescueLongTimeoutWorker) + .add::(RescueLongTimeoutWorker) .unwrap(); workers - .register::(RescueRetryOverrideWorker) + .add::(RescueRetryOverrideWorker) .unwrap(); let client = Client::builder( PostgresDatabase::new(pool.clone()) @@ -1704,9 +1704,9 @@ async fn resumable_cursor_and_transactional_checkpoints() { let cursor_values = Arc::new(Mutex::new(Vec::new())); let validate_runs = Arc::new(AtomicUsize::new(0)); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register::(ResumableCheckpointWorker { + .add::(ResumableCheckpointWorker { cursor_values: Arc::clone(&cursor_values), pool: pool.clone(), validate_runs: Arc::clone(&validate_runs), @@ -1830,9 +1830,9 @@ async fn resumable_cursor_and_transactional_checkpoints() { async fn stop_and_cancel_fails_a_job_ignoring_cancellation() { let database = support::PostgresSchema::current("rs_interrupt").await; - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register::(IgnoresCancelWorker) + .add::(IgnoresCancelWorker) .unwrap(); let client = Client::builder(database.pool.clone()) .id("rust-interrupt-client") @@ -2010,12 +2010,10 @@ async fn transactional_inserts_become_visible_on_commit() { /// Builds the maintenance client shared by the pilot and rescuer tests. fn maintenance_client(pool: &PgPool, pilot: TestPilot) -> Client { - let mut maintenance_workers = WorkerRegistry::new(); + let mut maintenance_workers = Workers::new(); + maintenance_workers.add::(EchoWorker).unwrap(); maintenance_workers - .register::(EchoWorker) - .unwrap(); - maintenance_workers - .register::(TransactionalWorker { pool: pool.clone() }) + .add::(TransactionalWorker { pool: pool.clone() }) .unwrap(); Client::builder(pool.clone()) .id("rust-maintenance-client") @@ -2088,11 +2086,11 @@ async fn wait_for_state(client: &Client, id: i64, expected: JobState) -> JobRow /// Builds a client on the pool's current schema that works the basic /// conformance job kinds. fn worker_client(pool: &PgPool, resumable: ResumableWorker) -> Client { - let mut workers = WorkerRegistry::new(); - workers.register::(CancelWorker).unwrap(); - workers.register::(EchoWorker).unwrap(); - workers.register::(FailWorker).unwrap(); - workers.register::(resumable).unwrap(); + let mut workers = Workers::new(); + workers.add::(CancelWorker).unwrap(); + workers.add::(EchoWorker).unwrap(); + workers.add::(FailWorker).unwrap(); + workers.add::(resumable).unwrap(); Client::builder(pool.clone()) .id("rust-conformance-client") .workers(workers) diff --git a/rust/riverqueue/tests/postgres_parity.rs b/rust/riverqueue/tests/postgres_parity.rs index 6833344df..8a91190a2 100644 --- a/rust/riverqueue/tests/postgres_parity.rs +++ b/rust/riverqueue/tests/postgres_parity.rs @@ -8,7 +8,7 @@ mod support; use std::{convert::Infallible, time::Duration}; use riverqueue::{ - Client, Error, Job, JobArgs, JobState, QueueConfig, WorkContext, WorkOutcome, WorkerRegistry, + Client, Error, Job, JobArgs, JobState, QueueConfig, WorkContext, WorkOutcome, Workers, database::PostgresDatabase, }; use serde::{Deserialize, Serialize}; @@ -302,10 +302,10 @@ async fn job_list_single_finalized_state_by_time() { #[river(kind = "parity_noop")] struct NoopArgs {} -fn noop_workers() -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn noop_workers() -> Workers { + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); diff --git a/rust/riverqueue/tests/producer_lifetime.rs b/rust/riverqueue/tests/producer_lifetime.rs index 6429c092b..67f7c8b1f 100644 --- a/rust/riverqueue/tests/producer_lifetime.rs +++ b/rust/riverqueue/tests/producer_lifetime.rs @@ -26,7 +26,7 @@ use riverqueue::__private::{ }; use riverqueue::{ Client, Error, ExtensionPhase, InsertOpts, Job, JobArgs, JobRow, JobState, QueueConfig, - WorkContext, WorkOutcome, WorkerRegistry, + WorkContext, WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value, json}; @@ -58,11 +58,11 @@ impl Gate { self.release.add_permits(jobs); } - fn workers(&self) -> WorkerRegistry { + fn workers(&self) -> Workers { let gate = self.clone(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |_context: WorkContext, job: Job| { + .add_fn(move |_context: WorkContext, job: Job| { let gate = gate.clone(); async move { gate.started.push(job.id()); diff --git a/rust/riverqueue/tests/producer_sessions.rs b/rust/riverqueue/tests/producer_sessions.rs index 51a249a10..e9db7df8f 100644 --- a/rust/riverqueue/tests/producer_sessions.rs +++ b/rust/riverqueue/tests/producer_sessions.rs @@ -23,7 +23,7 @@ use riverqueue::__private::{ }; use riverqueue::{ Client, Error, ExtensionPhase, InsertOpts, Job, JobArgs, JobRow, JobState, QueueConfig, - QueueUpdateParams, WorkContext, WorkOutcome, WorkerRegistry, + QueueUpdateParams, WorkContext, WorkOutcome, Workers, }; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -49,10 +49,10 @@ fn fast_queue(max_workers: usize) -> QueueConfig { .with_fetch_poll_interval(Duration::from_millis(10)) } -fn workers() -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn workers() -> Workers { + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, job: Job| async move { + .add_fn(|_context: WorkContext, job: Job| async move { if job.args.fail { return Err(std::io::Error::other("failed on purpose")); } @@ -321,10 +321,10 @@ async fn assert_abandoned_attempts_finish(builder: riverqueue::ClientBuilder) { let (release, blocked) = std::sync::mpsc::channel::<()>(); let blocked = Arc::new(Mutex::new(blocked)); let started = Arc::new(Notify::new()); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); let worker_started = Arc::clone(&started); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let blocked = Arc::clone(&blocked); let started = Arc::clone(&worker_started); async move { @@ -589,9 +589,9 @@ impl PilotProducer for GatedPilot { /// is handled before the claim is released. async fn assert_cancellation_during_claim_reaches_the_attempt(builder: riverqueue::ClientBuilder) { let pilot = GatedPilot::default(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn( + .add_fn( |context: WorkContext, _job: Job| async move { if context.cancellation_token().is_cancelled() { return Err(std::io::Error::other("started cancelled")); diff --git a/rust/riverqueue/tests/resilience_postgres.rs b/rust/riverqueue/tests/resilience_postgres.rs index 9c59c5184..64530a11d 100644 --- a/rust/riverqueue/tests/resilience_postgres.rs +++ b/rust/riverqueue/tests/resilience_postgres.rs @@ -25,7 +25,7 @@ use riverqueue::{ __private::{DatabaseConnection, JobSetStateParams, Pilot, PilotError}, AttemptError, BoxError, Client, ErrorHandler, ErrorHandlerDecision, EventKind, InsertOpts, Job, JobArgs, JobEventKind, JobRow, JobState, QueueConfig, RetryPolicy, WorkCancelled, WorkContext, - WorkOutcome, WorkResult, WorkerRegistry, + WorkOutcome, WorkResult, Workers, database::{PostgresDatabase, SchemaName}, }; use riverqueue_migrate::PostgresMigrator; @@ -89,12 +89,12 @@ impl ErrorHandler for StuckSignal { } } -fn blocking_workers(gate: &Gate, timeline: &Arc) -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn blocking_workers(gate: &Gate, timeline: &Arc) -> Workers { + let mut workers = Workers::new(); let blocking_gate = gate.clone(); let blocking_timeline = Arc::clone(timeline); workers - .register_fn(move |_context: WorkContext, job: Job| { + .add_fn(move |_context: WorkContext, job: Job| { let gate = blocking_gate.clone(); let timeline = Arc::clone(&blocking_timeline); async move { @@ -109,7 +109,7 @@ fn blocking_workers(gate: &Gate, timeline: &Arc) -> WorkerRegi .unwrap(); let later_timeline = Arc::clone(timeline); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let timeline = Arc::clone(&later_timeline); async move { timeline @@ -349,19 +349,19 @@ impl Drop for FaultProxy { } } -fn completing_workers() -> WorkerRegistry { +fn completing_workers() -> Workers { gated_workers(&Gate::default()) } -fn gated_workers(gate: &Gate) -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn gated_workers(gate: &Gate) -> Workers { + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); workers - .register_fn( + .add_fn( |_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Snooze(Duration::MAX)) }, @@ -369,7 +369,7 @@ fn gated_workers(gate: &Gate) -> WorkerRegistry { .unwrap(); let shutdown_gate = gate.clone(); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let gate = shutdown_gate.clone(); async move { gate.started.add_permits(1); @@ -385,7 +385,7 @@ fn gated_workers(gate: &Gate) -> WorkerRegistry { .unwrap(); let gate = gate.clone(); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let gate = gate.clone(); async move { let active = gate.active.fetch_add(1, Ordering::SeqCst) + 1; diff --git a/rust/riverqueue/tests/resilience_sqlite.rs b/rust/riverqueue/tests/resilience_sqlite.rs index 734cb03cd..3c72146e6 100644 --- a/rust/riverqueue/tests/resilience_sqlite.rs +++ b/rust/riverqueue/tests/resilience_sqlite.rs @@ -17,7 +17,7 @@ use chrono::{DateTime, Utc}; use riverqueue::{ AttemptError, BoxError, Client, ErrorHandler, ErrorHandlerDecision, EventKind, InsertOpts, Job, JobArgs, JobEventKind, JobRow, JobState, MaintenanceConfig, QueueConfig, RetryPolicy, - UniqueOpts, WorkCancelled, WorkContext, WorkOutcome, WorkResult, WorkerRegistry, + UniqueOpts, WorkCancelled, WorkContext, WorkOutcome, WorkResult, Workers, }; use riverqueue_migrate::SqliteMigrator; use serde::{Deserialize, Serialize}; @@ -134,20 +134,20 @@ impl Drop for TestDatabase { } } -fn completing_workers() -> WorkerRegistry { +fn completing_workers() -> Workers { gated_workers(&Gate::default()) } -fn gated_workers(gate: &Gate) -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn gated_workers(gate: &Gate) -> Workers { + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async { + .add_fn(|_context: WorkContext, _job: Job| async { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); let shutdown_gate = gate.clone(); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let gate = shutdown_gate.clone(); async move { gate.started.add_permits(1); @@ -162,7 +162,7 @@ fn gated_workers(gate: &Gate) -> WorkerRegistry { .unwrap(); let gate = gate.clone(); workers - .register_fn(move |_context: WorkContext, job: Job| { + .add_fn(move |_context: WorkContext, job: Job| { let gate = gate.clone(); async move { gate.started.add_permits(1); diff --git a/rust/riverqueue/tests/runtime_config.rs b/rust/riverqueue/tests/runtime_config.rs index 01b3fd1d6..c9b6285bb 100644 --- a/rust/riverqueue/tests/runtime_config.rs +++ b/rust/riverqueue/tests/runtime_config.rs @@ -16,7 +16,7 @@ use riverqueue::{ BoxError, Client, EventKind, EventReceiver, EventRecvError, Extensions, Hook, InsertContext, InsertMiddleware, InsertNext, InsertedJob, Job, JobArgs, JobRow, JobState, Metric, PeriodicJobs, Plugin, QueueConfig, SubscribeConfig, WorkContext, WorkError, WorkMiddleware, - WorkNext, WorkOutcome, Worker, WorkerRegistry, database::PostgresDatabase, + WorkNext, WorkOutcome, Worker, Workers, database::PostgresDatabase, }; use serde::{Deserialize, Serialize}; use sqlx::AssertSqlSafe; @@ -258,8 +258,8 @@ async fn setup_runtime() -> (Client, Arc, support::PostgresSchema let pool = database.pool.clone(); let schema = database.schema.clone(); - let mut workers = WorkerRegistry::new(); - workers.register::(RuntimeWorker).unwrap(); + let mut workers = Workers::new(); + workers.add::(RuntimeWorker).unwrap(); let counts = Arc::new(RuntimeCounts::default()); let client = Client::builder(PostgresDatabase::new(pool).with_schema(schema)) .default_max_attempts(7) @@ -288,8 +288,8 @@ async fn completion_burst_does_not_lag_large_subscription() { let pool = database.pool.clone(); let schema = database.schema.clone(); - let mut workers = WorkerRegistry::new(); - workers.register::(BurstWorker).unwrap(); + let mut workers = Workers::new(); + workers.add::(BurstWorker).unwrap(); let client = Client::builder(PostgresDatabase::new(pool.clone()).with_schema(schema.clone())) .id("rust-runtime-burst-test") .without_notifications() @@ -367,9 +367,9 @@ async fn external_terminal_state_wins_worker_completion_race() { let finish = Arc::new(Semaphore::new(0)); let started = Arc::new(Semaphore::new(0)); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register::(TerminalRaceWorker { + .add::(TerminalRaceWorker { finish: Arc::clone(&finish), started: Arc::clone(&started), }) @@ -458,9 +458,9 @@ async fn remote_cancellation_overrides_worker_snooze() { let schema = database.schema.clone(); let started = Arc::new(Semaphore::new(0)); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register::(CancelSnoozeWorker { + .add::(CancelSnoozeWorker { started: Arc::clone(&started), }) .unwrap(); @@ -508,9 +508,9 @@ async fn shutdown_waits_for_active_work_and_soft_stop_escalates() { let graceful_finish = Arc::new(Semaphore::new(0)); let graceful_started = Arc::new(Semaphore::new(0)); - let mut graceful_workers = WorkerRegistry::new(); + let mut graceful_workers = Workers::new(); graceful_workers - .register::(ShutdownWorker { + .add::(ShutdownWorker { finish: Arc::clone(&graceful_finish), started: Arc::clone(&graceful_started), }) @@ -583,9 +583,9 @@ async fn shutdown_waits_for_active_work_and_soft_stop_escalates() { ); let escalation_started = Arc::new(Semaphore::new(0)); - let mut escalation_workers = WorkerRegistry::new(); + let mut escalation_workers = Workers::new(); escalation_workers - .register::(ShutdownWorker { + .add::(ShutdownWorker { finish: Arc::new(Semaphore::new(0)), started: Arc::clone(&escalation_started), }) @@ -758,8 +758,8 @@ async fn poll_only_and_subscription_configuration() { fn start_without_runtime_returns_error_and_is_restartable() { let runtime = tokio::runtime::Runtime::new().unwrap(); let database = runtime.block_on(support::PostgresSchema::new("rt_missing")); - let mut workers = WorkerRegistry::new(); - workers.register::(BurstWorker).unwrap(); + let mut workers = Workers::new(); + workers.add::(BurstWorker).unwrap(); let client = Client::builder( PostgresDatabase::new(database.pool.clone()).with_schema(database.schema.clone()), ) diff --git a/rust/riverqueue/tests/sqlite_runtime.rs b/rust/riverqueue/tests/sqlite_runtime.rs index ca1f0a9d5..d88fb9ea0 100644 --- a/rust/riverqueue/tests/sqlite_runtime.rs +++ b/rust/riverqueue/tests/sqlite_runtime.rs @@ -16,7 +16,7 @@ use riverqueue::__private::{ClientBuilderExt, ExtensionClient, PreparedInsertPar use riverqueue::__private::{MaintenanceService, MaintenanceServiceContext}; use riverqueue::{ BoxError, Client, EventKind, Hook, InsertBatch, InsertOpts, Job, JobArgs, JobRow, JobState, - MaintenanceConfig, QueueConfig, UniqueOpts, WorkContext, WorkOutcome, WorkerRegistry, + MaintenanceConfig, QueueConfig, UniqueOpts, WorkContext, WorkOutcome, Workers, database::DatabaseKind, }; use riverqueue_migrate::SqliteMigrator; @@ -399,10 +399,10 @@ fn remove_sqlite_files(database_path: &std::path::Path) { } } -fn runtime_workers(worked: Arc) -> WorkerRegistry { - let mut workers = WorkerRegistry::new(); +fn runtime_workers(worked: Arc) -> Workers { + let mut workers = Workers::new(); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let worked = Arc::clone(&worked); async move { worked.add_permits(1); @@ -823,9 +823,9 @@ async fn sqlite_reinsert_preserves_wire_fields_and_runs_the_canonical_pipeline() .unwrap(); let mut pilot = SqlitePilot::new(); pilot.insert = Some(SelectionBehavior::Success); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async move { + .add_fn(|_context: WorkContext, _job: Job| async move { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); @@ -1163,9 +1163,9 @@ async fn sqlite_runs_jobs_and_persists_output() { let pool = setup().await; let worked = Arc::new(Semaphore::new(0)); let worked_for_worker = Arc::clone(&worked); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let worked = Arc::clone(&worked_for_worker); async move { context @@ -1223,9 +1223,9 @@ async fn sqlite_worker_cancelling_its_own_token_fails_the_attempt_normally() { struct GaveUp; let pool = setup().await; - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(|context: WorkContext, _job: Job| async move { + .add_fn(|context: WorkContext, _job: Job| async move { // Like a worker that cancels its subtasks through a drop guard // on its own token before returning. drop(context.cancellation_token().clone().drop_guard()); @@ -1280,9 +1280,9 @@ async fn sqlite_attempt_errors_record_when_the_attempt_started() { let pool = setup().await; let worker_started = Arc::new(Mutex::new(None)); let worker_started_for_worker = Arc::clone(&worker_started); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |_context: WorkContext, _job: Job| { + .add_fn(move |_context: WorkContext, _job: Job| { let worker_started = Arc::clone(&worker_started_for_worker); async move { *worker_started.lock().unwrap() = Some(chrono::Utc::now()); @@ -1475,9 +1475,9 @@ async fn sqlite_outbox_cancels_work_from_another_client() { let pool = setup().await; let started = Arc::new(Semaphore::new(0)); let started_for_worker = Arc::clone(&started); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |context: WorkContext, _job: Job| { + .add_fn(move |context: WorkContext, _job: Job| { let started = Arc::clone(&started_for_worker); async move { started.add_permits(1); @@ -1535,9 +1535,9 @@ async fn sqlite_completion_event_follows_external_cancelled_state() { let pool = setup().await; let finish = Arc::new(Semaphore::new(0)); let started = Arc::new(Semaphore::new(0)); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn({ + .add_fn({ let finish = Arc::clone(&finish); let started = Arc::clone(&started); move |_context: WorkContext, _job: Job| { @@ -1601,9 +1601,9 @@ async fn sqlite_fetches_and_discards_unregistered_kinds() { .opts(InsertOpts::default().with_max_attempts(1)) .await .unwrap(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(|_context: WorkContext, _job: Job| async move { + .add_fn(|_context: WorkContext, _job: Job| async move { Ok::<_, Infallible>(WorkOutcome::Complete) }) .unwrap(); @@ -1811,10 +1811,10 @@ async fn extension_claimed_rows_fail_undecodable_attempts_like_river_claims() { async fn set_state_extension_sees_jobs_deleted_while_worked() { let (pool, database_path) = setup_file_pool(Duration::from_secs(5)).await; let pilot = ClaimingPilot::default(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); let delete_pool = pool.clone(); workers - .register_fn(move |_context: WorkContext, job: Job| { + .add_fn(move |_context: WorkContext, job: Job| { let pool = delete_pool.clone(); async move { sqlx::query("DELETE FROM river_job WHERE id = ?") diff --git a/rust/riverqueue/tests/stuck_jobs.rs b/rust/riverqueue/tests/stuck_jobs.rs index 1277e8180..82fb98145 100644 --- a/rust/riverqueue/tests/stuck_jobs.rs +++ b/rust/riverqueue/tests/stuck_jobs.rs @@ -12,8 +12,7 @@ use std::{ }; use riverqueue::{ - Client, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Worker, WorkerRegistry, - WorkerTimeout, + Client, Job, JobArgs, QueueConfig, WorkContext, WorkOutcome, Worker, WorkerTimeout, Workers, }; use serde::{Deserialize, Serialize}; use tokio::sync::Notify; @@ -97,9 +96,9 @@ async fn stuck_log_line_reports_the_worker_timeout() { tracing::subscriber::set_default(tracing_subscriber::registry().with(lines.clone())); let (pool, path) = sqlite_file_pool(4).await; let started = Arc::new(Notify::new()); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register(StuckWorker { + .add(StuckWorker { started: Arc::clone(&started), }) .unwrap(); diff --git a/rust/riverqueue/tests/work_extensions.rs b/rust/riverqueue/tests/work_extensions.rs index 2d2ea0010..86e311161 100644 --- a/rust/riverqueue/tests/work_extensions.rs +++ b/rust/riverqueue/tests/work_extensions.rs @@ -17,7 +17,7 @@ use std::{ use riverqueue::{ Client, ErrorHandler, ErrorHandlerDecision, EventKind, Hook, InsertOpts, Job, JobArgs, JobRow, JobState, Metric, QueueConfig, WorkContext, WorkError, WorkMiddleware, WorkNext, WorkOutcome, - WorkerRegistry, + Workers, }; use riverqueue_migrate::SqliteMigrator; use serde::{Deserialize, Serialize}; @@ -178,11 +178,11 @@ impl ErrorHandler for PanickingErrorHandler { } } -fn workers(trace: &Trace) -> WorkerRegistry { +fn workers(trace: &Trace) -> Workers { let trace = trace.clone(); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |_context: WorkContext, job: Job| { + .add_fn(move |_context: WorkContext, job: Job| { let trace = trace.clone(); async move { trace.push("worker"); diff --git a/rust/riverqueue/tests/yugabyte.rs b/rust/riverqueue/tests/yugabyte.rs index efd3fbe94..9a758a746 100644 --- a/rust/riverqueue/tests/yugabyte.rs +++ b/rust/riverqueue/tests/yugabyte.rs @@ -18,7 +18,7 @@ use std::{convert::Infallible, sync::Arc, time::Duration}; use riverqueue::{ Client, EventKind, InsertOpts, Job, JobArgs, JobState, QueueConfig, QueueSelector, UniqueOpts, - WorkContext, WorkOutcome, WorkerRegistry, database::PostgresDatabase, + WorkContext, WorkOutcome, Workers, database::PostgresDatabase, }; use serde::{Deserialize, Serialize}; use sqlx::{ @@ -193,9 +193,9 @@ async fn polls_without_listen_notify() { let simulated = Simulated::new(server).await; let started = Arc::new(Semaphore::new(0)); let worker_started = Arc::clone(&started); - let mut workers = WorkerRegistry::new(); + let mut workers = Workers::new(); workers - .register_fn(move |context: WorkContext, job: Job| { + .add_fn(move |context: WorkContext, job: Job| { let started = Arc::clone(&worker_started); async move { if job.args.value == 0 {