diff --git a/Makefile b/Makefile index 4a77f3d6b..880c1ef60 100644 --- a/Makefile +++ b/Makefile @@ -206,9 +206,14 @@ check/js/package: build/js check/rust/dependencies: ## Audit Rust advisories, licenses, bans, and sources cd rust && cargo deny check +# Cargo caches temporary registry dependencies by path and version. A fresh +# build directory prevents stale sources and binaries after same-version edits. +# Verified archives still go to the normal target/package directory. .PHONY: check/rust/package check/rust/package: ## Build and verify publishable crate archives without publishing - cd rust && cargo package --workspace --allow-dirty --locked + cd rust && package_build_dir=$$(mktemp -d) && \ + trap 'rm -rf "$$package_build_dir"' EXIT && \ + CARGO_BUILD_BUILD_DIR="$$package_build_dir" cargo package --workspace --allow-dirty --locked # The baseline is the latest published riverqueue-v* tag, and # cargo-semver-checks infers the allowed change from the version bump. It diff --git a/rust/riverqueue-cli/src/bench.rs b/rust/riverqueue-cli/src/bench.rs index 410d04e8e..8d68be49c 100644 --- a/rust/riverqueue-cli/src/bench.rs +++ b/rust/riverqueue-cli/src/bench.rs @@ -440,7 +440,7 @@ async fn run_benchmark(options: BenchOptions) -> Result<(), Box Result<(), Box> { ``` Apply migrations before any client starts, and start clients inside a Tokio -runtime. `Client::start` returns a `RunHandle`: await `wait`, `shutdown` (a -soft stop that lets running jobs finish), or `shutdown_now` (which cancels +runtime. `Client::start` returns a `RunHandle`: await `wait`, `stop` (a +soft stop that lets running jobs finish), or `stop_and_cancel` (which cancels them). `RunHandle::stopper` returns a cloneable `Stopper` for stopping the client from another task, such as a signal handler. The handle controls the running client: dropping every `Client` clone doesn't stop it, dropping the @@ -224,8 +224,8 @@ async fn build_report( } ``` -During a client's hard stop (`RunHandle::shutdown_now` or -`Stopper::stop_now`), a job whose worker returns `WorkCancelled`, anywhere in +During a client's hard stop (`RunHandle::stop_and_cancel` or +`Stopper::stop_and_cancel`), a job whose worker returns `WorkCancelled`, anywhere in its error's source chain, becomes available again without using up its attempt. Any other error is recorded and consumes the attempt, and `Ok` completes the job. After the configured stuck threshold, River can abort a diff --git a/rust/riverqueue/docs/mixed-deployments.md b/rust/riverqueue/docs/mixed-deployments.md index ace84b343..21ed2fc08 100644 --- a/rust/riverqueue/docs/mixed-deployments.md +++ b/rust/riverqueue/docs/mixed-deployments.md @@ -116,7 +116,7 @@ with `SQLITE_BUSY` when a Rust process commits in between. ## Rolling back Rolling back doesn't touch the schema. Stop Rust clients gracefully with -`RunHandle::shutdown` and let the Go clients continue. Jobs that Rust inserted +`RunHandle::stop` and let the Go clients continue. Jobs that Rust inserted are ordinary River rows that Go workers can run, and any job a stopped Rust client left running is recovered by the rescuer. Only migrate down as a separately planned operation once no deployed client needs the newer schema. diff --git a/rust/riverqueue/examples/basic_worker.rs b/rust/riverqueue/examples/basic_worker.rs index ecbed9e4e..ba15bbb6d 100644 --- a/rust/riverqueue/examples/basic_worker.rs +++ b/rust/riverqueue/examples/basic_worker.rs @@ -46,6 +46,6 @@ async fn main() -> Result<(), Box> { .await?; while completed.recv().await?.as_job().map(|event| event.job.id) != Some(inserted.id()) {} - run.shutdown().await?; + run.stop().await?; Ok(()) } diff --git a/rust/riverqueue/examples/cancellation.rs b/rust/riverqueue/examples/cancellation.rs index ecd6470f0..5f41656ff 100644 --- a/rust/riverqueue/examples/cancellation.rs +++ b/rust/riverqueue/examples/cancellation.rs @@ -59,6 +59,6 @@ async fn main() -> Result<(), Box> { let job = client.insert(CancellableReport { report_id: 42 }).await?; client.jobs().cancel(job.id()).await?; - run.shutdown().await?; + run.stop().await?; Ok(()) } diff --git a/rust/riverqueue/examples/events.rs b/rust/riverqueue/examples/events.rs index 21c50c280..ff9d7b603 100644 --- a/rust/riverqueue/examples/events.rs +++ b/rust/riverqueue/examples/events.rs @@ -89,6 +89,6 @@ async fn main() -> Result<(), Box> { seen += 1; } - run.shutdown().await?; + run.stop().await?; Ok(()) } diff --git a/rust/riverqueue/examples/graceful_shutdown.rs b/rust/riverqueue/examples/graceful_shutdown.rs index 87cff07ad..645dddb22 100644 --- a/rust/riverqueue/examples/graceful_shutdown.rs +++ b/rust/riverqueue/examples/graceful_shutdown.rs @@ -53,7 +53,7 @@ async fn main() -> Result<(), Box> { stopper.stop(); } if tokio::signal::ctrl_c().await.is_ok() { - stopper.stop_now(); + stopper.stop_and_cancel(); } }); diff --git a/rust/riverqueue/examples/mixed_go_rust.rs b/rust/riverqueue/examples/mixed_go_rust.rs index 9ad10ff70..f7f12f68b 100644 --- a/rust/riverqueue/examples/mixed_go_rust.rs +++ b/rust/riverqueue/examples/mixed_go_rust.rs @@ -109,6 +109,6 @@ async fn main() -> Result<(), Box> { while completed.recv().await?.as_job().map(|event| event.job.id) != Some(inserted.id()) {} println!("send_receipt is waiting in the default queue for the Go service"); - run.shutdown().await?; + run.stop().await?; Ok(()) } diff --git a/rust/riverqueue/examples/periodic.rs b/rust/riverqueue/examples/periodic.rs index 34a0baa7e..50476e81d 100644 --- a/rust/riverqueue/examples/periodic.rs +++ b/rust/riverqueue/examples/periodic.rs @@ -79,6 +79,6 @@ async fn main() -> Result<(), Box> { break; } } - run.shutdown().await?; + run.stop().await?; Ok(()) } diff --git a/rust/riverqueue/examples/sqlite.rs b/rust/riverqueue/examples/sqlite.rs index 5d5970bc6..9716db67a 100644 --- a/rust/riverqueue/examples/sqlite.rs +++ b/rust/riverqueue/examples/sqlite.rs @@ -67,7 +67,7 @@ async fn main() -> Result<(), Box> { while completed.recv().await?.as_job().map(|event| event.job.id) != Some(inserted.id()) {} println!("job {} completed", inserted.id()); - run.shutdown().await?; + run.stop().await?; pool.close().await; std::fs::remove_dir_all(&directory).ok(); Ok(()) diff --git a/rust/riverqueue/examples/transactions.rs b/rust/riverqueue/examples/transactions.rs index f873fe62f..31ccd4a44 100644 --- a/rust/riverqueue/examples/transactions.rs +++ b/rust/riverqueue/examples/transactions.rs @@ -91,6 +91,6 @@ async fn main() -> Result<(), Box> { .await?; assert!(confirmed); - run.shutdown().await?; + run.stop().await?; Ok(()) } diff --git a/rust/riverqueue/src/client/builder.rs b/rust/riverqueue/src/client/builder.rs index 09d1f2a2a..4b62f1a80 100644 --- a/rust/riverqueue/src/client/builder.rs +++ b/rust/riverqueue/src/client/builder.rs @@ -568,11 +568,11 @@ impl ClientBuilder { /// The timeout must be positive. /// /// The client starts this timer when fetching stops, however the stop was - /// requested: [`RunHandle::shutdown`](crate::RunHandle::shutdown), + /// requested: [`RunHandle::stop`](crate::RunHandle::stop), /// [`Stopper::stop`](crate::Stopper::stop), or the signal passed to /// [`Client::start_with_graceful_shutdown`]. Jobs still running when it /// expires are cancelled as if by - /// [`Stopper::stop_now`](crate::Stopper::stop_now). + /// [`Stopper::stop_and_cancel`](crate::Stopper::stop_and_cancel). #[must_use] pub fn soft_stop_timeout(mut self, timeout: Duration) -> Self { self.soft_stop_timeout = Some(timeout); diff --git a/rust/riverqueue/src/client/run.rs b/rust/riverqueue/src/client/run.rs index eb26532d8..be8cef96a 100644 --- a/rust/riverqueue/src/client/run.rs +++ b/rust/riverqueue/src/client/run.rs @@ -60,7 +60,7 @@ impl Client { /// client stops fetching jobs and lets running jobs finish, and the /// builder's `soft_stop_timeout` escalates to cancelling them when set. /// Hard stops remain available through [`RunHandle::stopper`] and - /// [`RunHandle::shutdown_now`]. The client drops `signal` without + /// [`RunHandle::stop_and_cancel`]. The client drops `signal` without /// awaiting it further once it stops for any other reason. /// /// This mirrors the graceful shutdown hooks of Tokio servers such as @@ -452,8 +452,8 @@ impl Supervisor { /// soft stop to a hard stop after `soft_stop_timeout`. /// /// The escalation belongs to the client rather than to a caller awaiting -/// [`RunHandle::shutdown`], so it applies however the stop was requested and -/// dropping a shutdown future never changes it. This matches Go's client, +/// [`RunHandle::stop`], so it applies however the stop was requested and +/// dropping a stop future never changes it. This matches Go's client, /// which starts its soft stop timer when fetching stops. async fn watch_stop( fetch_cancel: CancellationToken, @@ -495,7 +495,7 @@ async fn watch_stop( /// immediately; observe completion through the handle. /// /// Requests are idempotent and ordered by severity: calling [`Stopper::stop`] -/// after [`Stopper::stop_now`] does not undo the hard stop, and requests made +/// after [`Stopper::stop_and_cancel`] does not undo the hard stop, and requests made /// after the client stopped do nothing. A stopper only affects the run it came /// from, not a later restart of the same [`Client`]. /// @@ -511,7 +511,7 @@ async fn watch_stop( /// stopper.stop(); /// let _ = tokio::signal::ctrl_c().await; /// // A second Ctrl-C cancels jobs that are still running. -/// stopper.stop_now(); +/// stopper.stop_and_cancel(); /// }); /// run.wait().await /// # } @@ -530,7 +530,7 @@ impl Stopper { /// stop at once, while each queue's producer keeps reporting its /// running jobs until they finish. When the builder's `soft_stop_timeout` /// is set, jobs still running after that timeout are cancelled as if by - /// [`Stopper::stop_now`]. + /// [`Stopper::stop_and_cancel`]. pub fn stop(&self) { self.fetch_cancel.cancel(); } @@ -550,7 +550,7 @@ impl Stopper { /// attempt. A job whose cancellation was requested with /// [`Jobs::cancel`](crate::Jobs::cancel) is cancelled rather than made /// available. - pub fn stop_now(&self) { + pub fn stop_and_cancel(&self) { self.fetch_cancel.cancel(); self.work_cancel.cancel(); } @@ -558,8 +558,8 @@ impl Stopper { /// Controls one running client instance. /// -/// [`RunHandle::wait`], [`RunHandle::shutdown`], and -/// [`RunHandle::shutdown_now`] take `&mut self`, can be called repeatedly, and +/// [`RunHandle::wait`], [`RunHandle::stop`], and +/// [`RunHandle::stop_and_cancel`] take `&mut self`, can be called repeatedly, and /// are cancel safe: dropping one of their futures, for example from /// `tokio::time::timeout` or `tokio::select!`, leaves the client and the /// handle as they were, apart from any stop the method already requested. To @@ -570,9 +570,9 @@ impl Stopper { /// The client's result is reported to the first call that observes it /// stopping; later calls return `Ok(())`. /// -/// Dropping the handle requests a hard stop, like [`Stopper::stop_now`], but -/// cannot wait for in-flight work to be recorded. Use [`RunHandle::shutdown`] -/// or [`RunHandle::shutdown_now`] when shutdown must finish before returning, +/// Dropping the handle requests a hard stop, like [`Stopper::stop_and_cancel`], but +/// cannot wait for in-flight work to be recorded. Use [`RunHandle::stop`] +/// or [`RunHandle::stop_and_cancel`] when shutdown must finish before returning, /// or [`RunHandle::detach`] to deliberately leave the client running. /// /// # Examples @@ -585,11 +585,11 @@ impl Stopper { /// /// let mut run = client.start()?; /// // ... serve until the application stops ... -/// if tokio::time::timeout(Duration::from_secs(30), run.shutdown()) +/// if tokio::time::timeout(Duration::from_secs(30), run.stop()) /// .await /// .is_err() /// { -/// run.shutdown_now().await?; +/// run.stop_and_cancel().await?; /// } /// # Ok(()) /// # } @@ -631,7 +631,7 @@ impl RunHandle { /// fails, or the process exits. Nothing observes its result, and jobs /// running when the process exits are left `running` for the rescuer. /// Most applications should keep the handle and await - /// [`RunHandle::shutdown`] instead. + /// [`RunHandle::stop`] instead. pub fn detach(mut self) { // Without a join handle, dropping the handle requests no stop. self.join.take(); @@ -647,20 +647,20 @@ impl RunHandle { /// This method is cancel safe. Dropping the future after its first poll /// leaves the soft stop in progress, including any `soft_stop_timeout` /// escalation, and never escalates to a hard stop by itself. The handle - /// remains usable: call [`RunHandle::shutdown_now`] to cancel running jobs + /// remains usable: call [`RunHandle::stop_and_cancel`] to cancel running jobs /// or [`RunHandle::wait`] to keep waiting. /// /// # Errors /// /// Returns the error that stopped the client, as [`RunHandle::wait`] does. - pub async fn shutdown(&mut self) -> Result<(), Error> { + pub async fn stop(&mut self) -> Result<(), Error> { self.stopper.stop(); self.wait().await } /// Requests a hard stop and waits for the client to stop. /// - /// This is [`Stopper::stop_now`] followed by [`RunHandle::wait`]. The stop + /// This is [`Stopper::stop_and_cancel`] followed by [`RunHandle::wait`]. The stop /// is requested when the future is first polled. /// /// # Cancel safety @@ -671,8 +671,8 @@ impl RunHandle { /// # Errors /// /// Returns the error that stopped the client, as [`RunHandle::wait`] does. - pub async fn shutdown_now(&mut self) -> Result<(), Error> { - self.stopper.stop_now(); + pub async fn stop_and_cancel(&mut self) -> Result<(), Error> { + self.stopper.stop_and_cancel(); self.wait().await } @@ -742,7 +742,7 @@ impl RunHandle { impl Drop for RunHandle { fn drop(&mut self) { if self.join.is_some() { - self.stopper.stop_now(); + self.stopper.stop_and_cancel(); } } } diff --git a/rust/riverqueue/src/client/tests.rs b/rust/riverqueue/src/client/tests.rs index 929143872..b34aaad03 100644 --- a/rust/riverqueue/src/client/tests.rs +++ b/rust/riverqueue/src/client/tests.rs @@ -589,7 +589,7 @@ async fn readiness_survives_a_notification_listener_panic() { client.inner.notifier_start_panics.load(Ordering::Acquire), 0 ); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); pool.close().await; for suffix in ["", "-shm", "-wal"] { let mut file = path.as_os_str().to_owned(); diff --git a/rust/riverqueue/src/maintenance/tests.rs b/rust/riverqueue/src/maintenance/tests.rs index 9f7993307..847b380c5 100644 --- a/rust/riverqueue/src/maintenance/tests.rs +++ b/rust/riverqueue/src/maintenance/tests.rs @@ -1008,7 +1008,7 @@ async fn periodic_start_hooks_and_run_on_start_follow_each_leadership_gain() { wait_for(1).await; client.request_resign().await.unwrap(); wait_for(2).await; - handle.shutdown().await.unwrap(); + handle.stop().await.unwrap(); assert_eq!(starts.load(Ordering::SeqCst), 2); assert_eq!(periodic_count().await, 2); database.cleanup().await; @@ -1089,7 +1089,7 @@ async fn maintenance_start_retries_then_resigns(poll_only: bool) { .expect("maintenance start should be retried in a new term"); let second_term = elected_at().await.unwrap(); assert_ne!(second_term, first_term); - handle.shutdown().await.unwrap(); + handle.stop().await.unwrap(); assert_eq!(attempts.load(Ordering::SeqCst), 4); database.cleanup().await; } diff --git a/rust/riverqueue/tests/client_handles.rs b/rust/riverqueue/tests/client_handles.rs index 9e0623632..cba5ffc13 100644 --- a/rust/riverqueue/tests/client_handles.rs +++ b/rust/riverqueue/tests/client_handles.rs @@ -93,7 +93,7 @@ macro_rules! scenarios { .unwrap(); let mut run = client.start().unwrap(); wait_for_completion(&client, id).await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let attempted_by = client.jobs().get(id).await.unwrap().attempted_by; let mut expected = previous[1..].to_vec(); @@ -134,7 +134,7 @@ macro_rules! scenarios { .unwrap(); client.jobs().cancel(id).await.unwrap(); wait_for_completion(&client, id).await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); fixture.cleanup().await; } @@ -408,7 +408,7 @@ macro_rules! scenarios { .unwrap(); let mut run = client.start().unwrap(); wait_for_completion(&client, ids[4]).await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!( *worked.lock().unwrap(), @@ -465,7 +465,7 @@ macro_rules! scenarios { let finalized_at = job.finalized_at.unwrap(); assert!((chrono::Utc::now() - finalized_at).num_seconds().abs() < 2); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); fixture.cleanup().await; } @@ -861,7 +861,7 @@ macro_rules! scenarios { ); assert!(!client.local_queues().configs().contains_key("dynamic")); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); fixture.cleanup().await; } }; diff --git a/rust/riverqueue/tests/extension_services.rs b/rust/riverqueue/tests/extension_services.rs index d3d0b670f..190dcaced 100644 --- a/rust/riverqueue/tests/extension_services.rs +++ b/rust/riverqueue/tests/extension_services.rs @@ -312,7 +312,7 @@ async fn assert_maintenance_services_restart_within_their_term(builder: riverque >= 3 }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let terms = pilot .calls diff --git a/rust/riverqueue/tests/fetch_only_known_kinds.rs b/rust/riverqueue/tests/fetch_only_known_kinds.rs index 4fd737d8c..d16970908 100644 --- a/rust/riverqueue/tests/fetch_only_known_kinds.rs +++ b/rust/riverqueue/tests/fetch_only_known_kinds.rs @@ -102,7 +102,7 @@ async fn claims_only_registered_kinds(backend: Backend) { } assert_eq!(completed, HashSet::from([known, alias])); - tokio::time::timeout(TIMEOUT, run.shutdown()) + tokio::time::timeout(TIMEOUT, run.stop()) .await .expect("the client should stop") .unwrap(); diff --git a/rust/riverqueue/tests/leader_election_disabled.rs b/rust/riverqueue/tests/leader_election_disabled.rs index 16523ff3a..07d68c87d 100644 --- a/rust/riverqueue/tests/leader_election_disabled.rs +++ b/rust/riverqueue/tests/leader_election_disabled.rs @@ -217,7 +217,7 @@ async fn works_jobs_without_electing(backend: Backend, poll_only: bool) { assert_eq!(next_completed(&mut events).await.id, inserted.job.row.id); assert_eq!(backend.leader_id().await, None); - tokio::time::timeout(Duration::from_secs(5), run.shutdown()) + tokio::time::timeout(Duration::from_secs(5), run.stop()) .await .expect("the client should stop") .unwrap(); @@ -258,7 +258,7 @@ async fn stays_ineligible_after_leader_stops(backend: Backend) { Some("eligible_leader") ); - tokio::time::timeout(Duration::from_secs(5), leader_run.shutdown()) + tokio::time::timeout(Duration::from_secs(5), leader_run.stop()) .await .expect("the leader should stop") .unwrap(); @@ -273,7 +273,7 @@ async fn stays_ineligible_after_leader_stops(backend: Backend) { assert_eq!(pilot.maintenance_starts.load(Ordering::SeqCst), 0); assert_eq!(pilot.runtime_starts.load(Ordering::SeqCst), 1); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); backend.cleanup().await; } diff --git a/rust/riverqueue/tests/lifecycle.rs b/rust/riverqueue/tests/lifecycle.rs index b5fd3427d..cc4264f33 100644 --- a/rust/riverqueue/tests/lifecycle.rs +++ b/rust/riverqueue/tests/lifecycle.rs @@ -203,7 +203,7 @@ async fn dropping_a_shutdown_future_keeps_the_soft_stop() { // the gated job keeps it pending, as `tokio::time::timeout` would. tokio::select! { biased; - result = run.shutdown() => panic!("shutdown finished while a job was held: {result:?}"), + result = run.stop() => panic!("shutdown finished while a job was held: {result:?}"), () = std::future::ready(()) => {} } gate.release(); @@ -220,7 +220,7 @@ async fn dropping_a_shutdown_future_keeps_the_soft_stop() { JobState::Available ); // The handle is still usable after the dropped future. - run.shutdown_now().await.unwrap(); + run.stop_and_cancel().await.unwrap(); } #[tokio::test(flavor = "multi_thread")] @@ -297,19 +297,19 @@ async fn lifecycle_methods_are_idempotent() { let stopper = run.stopper(); stopper.stop(); stopper.stop(); - run.shutdown().await.unwrap(); - run.shutdown().await.unwrap(); - run.shutdown_now().await.unwrap(); + run.stop().await.unwrap(); + run.stop().await.unwrap(); + run.stop_and_cancel().await.unwrap(); run.wait().await.unwrap(); run.wait_ready().await.unwrap(); - stopper.stop_now(); + stopper.stop_and_cancel(); stopper.stop(); // A stopper affects only its own run, not a restart of the client. let job = insert_gated(&client).await; let mut restarted = client.start().unwrap(); gate.wait_started().await; - stopper.stop_now(); + stopper.stop_and_cancel(); gate.release(); let completed = async { while client.jobs().get(job).await.unwrap().state != JobState::Completed { @@ -320,7 +320,7 @@ async fn lifecycle_methods_are_idempotent() { .await .expect("the restarted client did not complete its job"); assert_eq!(gate.ending(job), Some(Ending::Released)); - restarted.shutdown().await.unwrap(); + restarted.stop().await.unwrap(); } #[tokio::test(flavor = "multi_thread")] @@ -373,7 +373,7 @@ async fn stop_from_another_task_while_waiting() { } #[tokio::test(flavor = "multi_thread")] -async fn stop_now_from_another_task_interrupts_running_jobs() { +async fn stop_and_cancel_from_another_task_interrupts_running_jobs() { let database = TestDatabase::new().await; let gate = Gate::new(); let client = database.client(&gate, 2); @@ -397,7 +397,7 @@ async fn stop_now_from_another_task_interrupts_running_jobs() { .execute(&pool) .await .unwrap(); - stopper.stop_now(); + stopper.stop_and_cancel(); }); wait_stopped(&mut run).await; stop_task.await.unwrap(); @@ -451,7 +451,7 @@ async fn stop_during_an_outage_does_not_wait_for_a_connection() { // Give polls and fetches time to start waiting; a stop must end them // wherever they are. tokio::time::sleep(Duration::from_millis(300)).await; - tokio::time::timeout(Duration::from_secs(5), run.shutdown()) + tokio::time::timeout(Duration::from_secs(5), run.stop()) .await .expect("the stop waited for a pool connection") .unwrap(); @@ -471,5 +471,5 @@ async fn wait_ready_waits_for_queue_registration() { .await .unwrap(); assert_eq!(queues, 1); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } diff --git a/rust/riverqueue/tests/metadata_exact.rs b/rust/riverqueue/tests/metadata_exact.rs index 72121d22c..ab7076713 100644 --- a/rust/riverqueue/tests/metadata_exact.rs +++ b/rust/riverqueue/tests/metadata_exact.rs @@ -288,7 +288,7 @@ async fn sqlite_snooze_preserves_large_metadata_numbers() { }) .await .unwrap(); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!(row.metadata.get_raw("beyond_float").unwrap().get(), "1e400"); assert_eq!( row.metadata.get_raw("long_decimal").unwrap().get(), diff --git a/rust/riverqueue/tests/peer_attempts.rs b/rust/riverqueue/tests/peer_attempts.rs index 25f12ac67..e905440fd 100644 --- a/rust/riverqueue/tests/peer_attempts.rs +++ b/rust/riverqueue/tests/peer_attempts.rs @@ -600,7 +600,7 @@ impl Run { } async fn stop(mut self) { - tokio::time::timeout(WAIT, self.handle.shutdown()) + tokio::time::timeout(WAIT, self.handle.stop()) .await .expect("client stops") .unwrap(); @@ -1006,7 +1006,7 @@ async fn assert_stops_interrupt_and_cancellations_fail( assert!(row.errors[0].error.contains("ended without an outcome")); run.stop().await; } else { - tokio::time::timeout(WAIT, run.handle.shutdown_now()) + tokio::time::timeout(WAIT, run.handle.stop_and_cancel()) .await .expect("client stops") .unwrap(); @@ -1098,7 +1098,7 @@ async fn assert_soft_stops_keep_claims_open( run.assert_checks().await; run.settled(run.coordinator).await; } else { - run.handle.stopper().stop_now(); + run.handle.stopper().stop_and_cancel(); run.assert_checks().await; } let peer = run.client.jobs().get(run.env.peers[0]).await.unwrap(); @@ -1175,7 +1175,7 @@ async fn assert_coordinator_lifetime_bounds_operations( run.env.signals.wait("claiming").await; // A hard stop cancels the coordinator without a write, which SQLite // couldn't take while the claim holds its write lock. - run.handle.stopper().stop_now(); + run.handle.stopper().stop_and_cancel(); run.assert_checks().await; let peer = run.client.jobs().get(run.env.peers[0]).await.unwrap(); assert_eq!(peer.state, JobState::Available); @@ -1343,7 +1343,7 @@ async fn assert_abandoned_attempts_release_peers( ) .await; run.env.signals.wait("coordinator blocks").await; - tokio::time::timeout(WAIT, run.handle.shutdown_now()) + tokio::time::timeout(WAIT, run.handle.stop_and_cancel()) .await .expect("client stops") .unwrap(); diff --git a/rust/riverqueue/tests/poll_only_cancellation.rs b/rust/riverqueue/tests/poll_only_cancellation.rs index bb6724fc0..c1d256200 100644 --- a/rust/riverqueue/tests/poll_only_cancellation.rs +++ b/rust/riverqueue/tests/poll_only_cancellation.rs @@ -105,7 +105,7 @@ async fn polls_for_remote_cancellation(backend: Backend) { let job = &event.as_job().expect("a job event").job; assert_eq!(job.id, id); assert_eq!(job.state, JobState::Cancelled, "{job:?}"); - tokio::time::timeout(TIMEOUT, run.shutdown()) + tokio::time::timeout(TIMEOUT, run.stop()) .await .expect("the client should stop") .unwrap(); diff --git a/rust/riverqueue/tests/postgres.rs b/rust/riverqueue/tests/postgres.rs index 81e5204b5..df1abdb6f 100644 --- a/rust/riverqueue/tests/postgres.rs +++ b/rust/riverqueue/tests/postgres.rs @@ -523,7 +523,7 @@ async fn client_cancels_a_running_job() { assert!(cancelled.finalized_at.is_some()); assert!(cancelled.metadata.contains_key("cancel_attempted_at")); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); database.cleanup().await; } @@ -559,7 +559,7 @@ async fn client_completes_a_job_with_output_and_event() { } } - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); database.cleanup().await; } @@ -578,7 +578,7 @@ async fn client_discards_a_failing_job_after_max_attempts() { assert_eq!(failed.errors.len(), 1); assert_eq!(failed.errors[0].error, "intentional failure"); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); database.cleanup().await; } @@ -603,7 +603,7 @@ async fn client_discards_a_job_of_an_unregistered_kind() { "job kind is not registered in the client's Workers bundle: rust_unregistered_kind" ); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); database.cleanup().await; } @@ -614,10 +614,10 @@ async fn client_restarts_after_shutdown() { let mut run_handle = client.start().unwrap(); run_handle.wait_ready().await.unwrap(); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); let mut restarted_handle = client.start().unwrap(); restarted_handle.wait_ready().await.unwrap(); - restarted_handle.shutdown().await.unwrap(); + restarted_handle.stop().await.unwrap(); database.cleanup().await; } @@ -648,7 +648,7 @@ async fn client_resumes_resumable_steps_on_retry() { assert_eq!(resumable_first_runs.load(Ordering::SeqCst), 1); assert_eq!(resumable_second_runs.load(Ordering::SeqCst), 2); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); database.cleanup().await; } @@ -1155,7 +1155,7 @@ async fn job_admin_lists_updates_retries_and_deletes() { .unwrap(); wait_for_state(&client, inserted.job.row.id, JobState::Completed).await; let failed = wait_for_state(&client, failed.job.row.id, JobState::Discarded).await; - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); let listed = client .jobs() @@ -1215,7 +1215,7 @@ async fn local_queue_added_at_runtime_works_jobs() { .unwrap(); wait_for_state(&client, dynamic.job.row.id, JobState::Completed).await; client.local_queues().remove("dynamic").await.unwrap(); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); // The removed queue's row stays behind alongside the configured one. let queues = client @@ -1328,7 +1328,7 @@ async fn maintenance_cleans_old_jobs_and_queues_and_reindexes() { ); tokio::time::sleep(Duration::from_millis(20)).await; } - cleanup_handle.shutdown().await.unwrap(); + cleanup_handle.stop().await.unwrap(); database.cleanup().await; } @@ -1408,7 +1408,7 @@ async fn maintenance_client_runs_pilot_periodic_scheduled_and_transactional_jobs ); assert_eq!(pilot.maintenance_starts.load(Ordering::SeqCst), 1); assert_eq!(pilot.runtime_starts.load(Ordering::SeqCst), 1); - maintenance_handle.shutdown().await.unwrap(); + maintenance_handle.stop().await.unwrap(); assert_eq!(pilot.maintenance_stops.load(Ordering::SeqCst), 1); assert_eq!(pilot.runtime_stops.load(Ordering::SeqCst), 1); @@ -1444,7 +1444,7 @@ async fn maintenance_leader_rescues_stuck_jobs_and_resigns_on_shutdown() { .await .unwrap(); assert_eq!(leader_id, "rust-maintenance-client"); - maintenance_handle.shutdown().await.unwrap(); + maintenance_handle.stop().await.unwrap(); let leader_count: i64 = sqlx::query_scalar("SELECT count(*) FROM river_leader") .fetch_one(&pool) .await @@ -1532,7 +1532,7 @@ async fn queue_admin_gets_pauses_resumes_and_updates() { // Starting the client records its configured queue. let mut run_handle = client.start().unwrap(); run_handle.wait_ready().await.unwrap(); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); let queue = client.queues().get("default").await.unwrap(); assert_eq!(queue.name, "default"); @@ -1681,7 +1681,7 @@ async fn rescuer_honors_worker_timeout_and_retry_overrides() { assert!(!row.metadata.contains_key("river:rescue_count")); } - handle.shutdown().await.unwrap(); + handle.stop().await.unwrap(); database.cleanup().await; } @@ -1818,7 +1818,7 @@ async fn resumable_cursor_and_transactional_checkpoints() { assert!(!rolled_back.metadata.contains_key("river:resumable_cursor")); } - handle.shutdown().await.unwrap(); + handle.stop().await.unwrap(); database.cleanup().await; } @@ -1827,7 +1827,7 @@ async fn resumable_cursor_and_transactional_checkpoints() { /// error is recorded, and the job is retried or, at its maximum attempts, /// discarded. #[tokio::test] -async fn shutdown_now_fails_a_job_ignoring_cancellation() { +async fn stop_and_cancel_fails_a_job_ignoring_cancellation() { let database = support::PostgresSchema::current("rs_interrupt").await; let mut workers = WorkerRegistry::new(); @@ -1867,7 +1867,7 @@ async fn shutdown_now_fails_a_job_ignoring_cancellation() { let mut handle = client.start().unwrap(); wait_for_state(&client, retried, JobState::Running).await; wait_for_state(&client, discarded, JobState::Running).await; - handle.shutdown_now().await.unwrap(); + handle.stop_and_cancel().await.unwrap(); for (id, states) in [ (retried, &[JobState::Available, JobState::Retryable][..]), diff --git a/rust/riverqueue/tests/postgres_parity.rs b/rust/riverqueue/tests/postgres_parity.rs index 9c2b40d60..6833344df 100644 --- a/rust/riverqueue/tests/postgres_parity.rs +++ b/rust/riverqueue/tests/postgres_parity.rs @@ -348,7 +348,7 @@ async fn schema_names_are_quoted_like_go() { let inserted = client.insert(NoopArgs {}).await.unwrap(); wait_for_job_state(&client, inserted.job.row.id, JobState::Completed).await; - handle.shutdown().await.unwrap(); + handle.stop().await.unwrap(); database.cleanup().await; } @@ -521,7 +521,7 @@ async fn leader_renews_while_maintenance_is_blocked() { assert!(blocked_deletes().await > 0); // Shutdown cancels the blocked statement server-side instead of waiting. - tokio::time::timeout(Duration::from_secs(10), handle.shutdown()) + tokio::time::timeout(Duration::from_secs(10), handle.stop()) .await .expect("shutdown should cancel blocked maintenance") .unwrap(); diff --git a/rust/riverqueue/tests/producer_lifetime.rs b/rust/riverqueue/tests/producer_lifetime.rs index 4e117f1bc..6429c092b 100644 --- a/rust/riverqueue/tests/producer_lifetime.rs +++ b/rust/riverqueue/tests/producer_lifetime.rs @@ -329,7 +329,7 @@ async fn assert_shutdown_retries_with_growing_deadlines(builder: riverqueue::Cli .unwrap(); let mut run = client.start().unwrap(); run.wait_ready().await.unwrap(); - tokio::time::timeout(WAIT, run.shutdown()) + tokio::time::timeout(WAIT, run.stop()) .await .expect("client stops") .unwrap(); @@ -374,7 +374,7 @@ async fn assert_stuck_keep_alives_time_out(builder: riverqueue::ClientBuilder) { (Duration::from_secs(9)..Duration::from_secs(15)).contains(&waited), "{waited:?}" ); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } /// A panic in `job_finished` or `configuration_changed` stops the client @@ -502,7 +502,7 @@ async fn assert_removal_waits_for_the_drain(builder: riverqueue::ClientBuilder) pilot.starts.load(Ordering::SeqCst) > starts }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } /// A running producer applies an updated configuration without restarting: @@ -556,7 +556,7 @@ async fn assert_updates_apply_while_running(builder: riverqueue::ClientBuilder) .await .expect("jobs complete"); } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!(pilot.starts.load(Ordering::SeqCst), 1, "no restart"); } @@ -635,7 +635,7 @@ async fn assert_queue_settings_reach_the_session(builder: impl Fn() -> riverqueu }) }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let Some(Call::Started(started)) = pilot.calls.snapshot().first().cloned() else { panic!("session never started"); }; diff --git a/rust/riverqueue/tests/producer_sessions.rs b/rust/riverqueue/tests/producer_sessions.rs index 695f1eb46..51a249a10 100644 --- a/rust/riverqueue/tests/producer_sessions.rs +++ b/rust/riverqueue/tests/producer_sessions.rs @@ -298,7 +298,7 @@ async fn assert_every_attempt_finishes_once(builder: riverqueue::ClientBuilder) ids.iter().all(|id| finished.contains(id)) }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let finished = counts(&pilot.finished.snapshot()); assert_eq!(finished.len(), ids.len(), "{finished:?}"); @@ -349,7 +349,7 @@ async fn assert_abandoned_attempts_finish(builder: riverqueue::ClientBuilder) { .await .expect("worker starts"); - tokio::time::timeout(WAIT, run.shutdown_now()) + tokio::time::timeout(WAIT, run.stop_and_cancel()) .await .expect("client stops") .unwrap(); @@ -399,7 +399,7 @@ async fn assert_panicked_attempts_finish(builder: riverqueue::ClientBuilder) { finished.contains(&id) }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!(pilot.finished.snapshot(), [id]); } @@ -535,7 +535,7 @@ async fn assert_queue_changes_reach_the_session( .await .unwrap_or_else(|_| panic!("metadata {metadata} not reported in time")); } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } /// A job whose worker reports whether its attempt started cancelled. @@ -623,7 +623,7 @@ async fn assert_cancellation_during_claim_reaches_the_attempt(builder: riverqueu }) .await .expect("job finishes"); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!( client.jobs().get(id).await.unwrap().state, @@ -677,7 +677,7 @@ async fn assert_sessions_see_metadata_text( configurations.last().unwrap().queue.metadata, "only the text changed" ); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[cfg(all(feature = "postgres", river_postgres_tests))] diff --git a/rust/riverqueue/tests/resilience_postgres.rs b/rust/riverqueue/tests/resilience_postgres.rs index 21128e1b5..9c59c5184 100644 --- a/rust/riverqueue/tests/resilience_postgres.rs +++ b/rust/riverqueue/tests/resilience_postgres.rs @@ -538,7 +538,7 @@ async fn claimed_rows_decode_individually_and_leniently() { let event = event.as_job().unwrap().clone(); events_by_id.insert(event.job.id, event); } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); for id in decodable_ids { assert_eq!(events_by_id[&id].kind, JobEventKind::Completed); @@ -662,7 +662,7 @@ async fn completion_retries_a_transient_database_error() { || async { schema.job_state(job.job.row.id).await == "completed" }, ) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let injected: i64 = sqlx::query_scalar(AssertSqlSafe(format!( "SELECT last_value FROM {}.completion_fault", @@ -719,7 +719,7 @@ async fn completion_waits_for_a_row_lock() { || async { schema.job_state(job.job.row.id).await == "completed" }, ) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); schema.drop().await; } @@ -749,7 +749,7 @@ async fn completion_leaves_rows_moved_out_of_running_and_keeps_working() { || async { schema.job_state(later.job.row.id).await == "completed" }, ) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!(schema.job_state(pending.job.row.id).await, "pending"); @@ -778,7 +778,7 @@ async fn completion_does_not_rewrite_a_newer_attempt_number() { schema.job_state(job.job.row.id).await == "completed" }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!(schema.job_attempt(job.job.row.id).await, 5); @@ -820,7 +820,7 @@ async fn hard_shutdown_interrupts_only_cooperative_cancellations() { cancel_attempted.job.row.id )) .await; - tokio::time::timeout(Duration::from_secs(10), run.shutdown_now()) + tokio::time::timeout(Duration::from_secs(10), run.stop_and_cancel()) .await .unwrap() .unwrap(); @@ -908,7 +908,7 @@ async fn stuck_job_keeps_its_worker_slot_until_it_ends() { schema.job_state(later.job.row.id).await == "completed" }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let blocking_finished = timeline.blocking_finished.lock().unwrap().unwrap(); let later_started = timeline.later_started.lock().unwrap().unwrap(); @@ -948,7 +948,7 @@ async fn shutdown_leaves_a_job_still_stuck_after_abort_running() { let mut run = client.start().unwrap(); gate.wait_started().await; let started = std::time::Instant::now(); - tokio::time::timeout(Duration::from_secs(1), run.shutdown_now()) + tokio::time::timeout(Duration::from_secs(1), run.stop_and_cancel()) .await .expect("shutdown waited for a task that cannot be aborted") .unwrap(); @@ -970,7 +970,7 @@ async fn out_of_range_snooze_is_clamped_and_cancel_time_matches_go() { schema.job_state(snoozed.job.row.id).await == "scheduled" }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let snoozed = client.jobs().get(snoozed.job.row.id).await.unwrap(); assert_eq!(snoozed.attempt, 0); assert!(snoozed.scheduled_at > chrono::Utc::now() + chrono::Duration::days(365 * 200)); @@ -1014,7 +1014,7 @@ async fn snooze_preserves_metadata_numbers_beyond_float_range() { schema.job_state(snoozed.job.row.id).await == "scheduled" }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let snoozed = client.jobs().get(snoozed.job.row.id).await.unwrap(); // PostgreSQL expands `1e400` in jsonb; the snooze must keep it intact. assert!(snoozed.metadata.get_raw("unrelated").unwrap().get().len() > 400); @@ -1077,7 +1077,7 @@ async fn client_survives_database_outage_and_catches_up() { }) .await; } - tokio::time::timeout(Duration::from_secs(10), run.shutdown()) + tokio::time::timeout(Duration::from_secs(10), run.stop()) .await .unwrap() .unwrap(); @@ -1122,7 +1122,7 @@ async fn client_started_during_an_outage_becomes_ready_after_recovery() { schema.job_state(job.job.row.id).await == "completed" }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); drop(proxy); schema.drop().await; @@ -1153,7 +1153,7 @@ async fn listener_does_not_occupy_a_pool_connection() { || async { schema.job_state(job.job.row.id).await == "completed" }, ) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); pool.close().await; schema.drop().await; @@ -1199,7 +1199,7 @@ async fn queue_update_keeps_the_producer_within_max_workers() { }) .await; } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!( gate.max_active.load(Ordering::SeqCst), 1, @@ -1310,7 +1310,7 @@ async fn extension_set_state_hook_runs_in_the_completion_transaction() { .unwrap(); completed.push(event.as_job().unwrap().job.id); } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); completed.sort_unstable(); let mut expected = vec![kept.job.row.id, deleted.job.row.id]; diff --git a/rust/riverqueue/tests/resilience_sqlite.rs b/rust/riverqueue/tests/resilience_sqlite.rs index d9ec10f12..734cb03cd 100644 --- a/rust/riverqueue/tests/resilience_sqlite.rs +++ b/rust/riverqueue/tests/resilience_sqlite.rs @@ -331,7 +331,7 @@ async fn claimed_rows_decode_individually_and_accept_go_integer_ranges() { let event = event.as_job().unwrap().clone(); events_by_id.insert(event.job.id, event); } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); for id in decodable_ids { assert_eq!(events_by_id[&id].kind, JobEventKind::Completed); @@ -473,7 +473,7 @@ async fn rescuer_recovers_undecodable_stuck_jobs() { }) .await; } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let tags: String = sqlx::query_scalar("SELECT json(tags) FROM river_job WHERE id = ?") .bind(stuck[0]) @@ -635,7 +635,7 @@ async fn invalid_json_columns_fail_attempts_without_stalling_the_queue() { }) .await; } - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test(flavor = "multi_thread")] @@ -667,7 +667,7 @@ async fn completion_retries_while_a_foreign_writer_holds_the_lock() { || async { job_state(&database.pool, job.job.row.id).await == "completed" }, ) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); foreign.close().await; } @@ -700,7 +700,7 @@ async fn completion_cancels_on_a_null_cancel_attempted_at_key() { job_state(&database.pool, job.job.row.id).await == "cancelled" }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test(flavor = "multi_thread")] @@ -744,7 +744,7 @@ async fn hard_shutdown_interrupts_only_cooperative_cancellations() { .execute(&database.pool) .await .unwrap(); - tokio::time::timeout(Duration::from_secs(10), run.shutdown_now()) + tokio::time::timeout(Duration::from_secs(10), run.stop_and_cancel()) .await .unwrap() .unwrap(); @@ -854,7 +854,7 @@ async fn notification_poll_failures_do_not_stop_the_client() { job_state(&foreign, job.job.row.id).await == "completed" }) .await; - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); foreign.close().await; } diff --git a/rust/riverqueue/tests/runtime_config.rs b/rust/riverqueue/tests/runtime_config.rs index 365b96f32..01b3fd1d6 100644 --- a/rust/riverqueue/tests/runtime_config.rs +++ b/rust/riverqueue/tests/runtime_config.rs @@ -346,7 +346,7 @@ async fn completion_burst_does_not_lag_large_subscription() { .is_err(), "unexpected extra completion event" ); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); let completed_count: i64 = sqlx::query_scalar(AssertSqlSafe(format!( "SELECT count(*) FROM {} WHERE state = 'completed'", @@ -447,7 +447,7 @@ async fn external_terminal_state_wins_worker_completion_race() { ); } - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); database.cleanup().await; } @@ -495,7 +495,7 @@ async fn remote_cancellation_overrides_worker_snooze() { ); assert_eq!(row.attempt, 1); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); database.cleanup().await; } @@ -551,7 +551,7 @@ async fn shutdown_waits_for_active_work_and_soft_stop_escalates() { graceful_handle.stopper().stop(); let finish = Arc::clone(&graceful_finish); let graceful_shutdown = tokio::spawn(async move { - graceful_handle.shutdown().await.unwrap(); + graceful_handle.stop().await.unwrap(); finish.available_permits() }); graceful_finish.add_permits(1); @@ -619,7 +619,7 @@ async fn shutdown_waits_for_active_work_and_soft_stop_escalates() { escalation_handle.wait_ready().await.unwrap(); escalation_started.acquire().await.unwrap().forget(); let shutdown_started = tokio::time::Instant::now(); - tokio::time::timeout(Duration::from_secs(2), escalation_handle.shutdown()) + tokio::time::timeout(Duration::from_secs(2), escalation_handle.stop()) .await .unwrap() .unwrap(); @@ -746,7 +746,7 @@ async fn poll_only_and_subscription_configuration() { )); assert_eq!(lagged.recv().await.unwrap().kind(), EventKind::QueuePaused); - run_handle.shutdown().await.unwrap(); + run_handle.stop().await.unwrap(); assert_eq!( client.jobs().get(inserted.job.row.id).await.unwrap().state, JobState::Completed @@ -778,7 +778,7 @@ fn start_without_runtime_returns_error_and_is_restartable() { .start() .expect("failed start must not poison the client"); run.wait_ready().await.unwrap(); - run.shutdown_now().await.unwrap(); + run.stop_and_cancel().await.unwrap(); database.cleanup().await; }); drop(client); diff --git a/rust/riverqueue/tests/sqlite_runtime.rs b/rust/riverqueue/tests/sqlite_runtime.rs index e24ccea83..ca1f0a9d5 100644 --- a/rust/riverqueue/tests/sqlite_runtime.rs +++ b/rust/riverqueue/tests/sqlite_runtime.rs @@ -469,7 +469,7 @@ async fn sqlite_pilot_completion_continue_and_mark_are_atomic() { assert!(pilot.fetch_calls.load(Ordering::SeqCst) >= 1); assert_eq!(pilot.completion_calls.load(Ordering::SeqCst), 1); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } } @@ -520,7 +520,7 @@ async fn sqlite_pilot_completion_error_rolls_back_side_effects() { JobState::Running ); - run.shutdown_now().await.unwrap(); + run.stop_and_cancel().await.unwrap(); } #[tokio::test] @@ -562,7 +562,7 @@ async fn sqlite_pilot_fetch_error_rolls_back_selection_side_effects() { JobState::Available ); - let _ = run.shutdown_now().await; + let _ = run.stop_and_cancel().await; } #[tokio::test] @@ -599,7 +599,7 @@ async fn sqlite_pilot_fetch_transient_error_retries_without_stopping_the_queue() }) .await .unwrap(); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); let fetch_calls = pilot.fetch_calls.load(Ordering::SeqCst); let effects: i64 = @@ -649,7 +649,7 @@ async fn sqlite_queue_start_retries_transient_write_contention() { .await .unwrap(); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); pool.close().await; remove_sqlite_files(&database_path); } @@ -727,7 +727,7 @@ async fn sqlite_transient_renewal_contention_preserves_leadership_services() { .await .expect("leader resigns on request"); - run.shutdown_now().await.unwrap(); + run.stop_and_cancel().await.unwrap(); assert!(service.stops.load(Ordering::SeqCst) >= 1); pool.close().await; remove_sqlite_files(&database_path); @@ -1037,7 +1037,7 @@ async fn sqlite_pilot_rescue_selection_and_update_share_a_transaction() { assert_eq!(effects, 1); assert!(pilot.rescue_calls.load(Ordering::SeqCst) >= 1); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1093,7 +1093,7 @@ async fn sqlite_pilot_rescue_error_rolls_back_selection_side_effects() { JobState::Running ); - run.shutdown_now().await.unwrap(); + run.stop_and_cancel().await.unwrap(); } #[tokio::test] @@ -1155,7 +1155,7 @@ async fn sqlite_queue_events_are_emitted_once_per_transition() { .is_err() ); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1213,7 +1213,7 @@ async fn sqlite_runs_jobs_and_persists_output() { Some(r#"{"doubled":42}"#) ); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1268,7 +1268,7 @@ async fn sqlite_worker_cancelling_its_own_token_fails_the_attempt_normally() { assert_eq!(row.errors.len(), 1); assert_eq!(row.errors[0].error, "worker gave up"); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1320,7 +1320,7 @@ async fn sqlite_attempt_errors_record_when_the_attempt_started() { "error at {at}, worker started at {worker_started}" ); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1467,7 +1467,7 @@ async fn sqlite_long_fetch_cooldown_still_fetches_first() { .expect("the first fetch doesn't wait for a cooldown") .unwrap() .forget(); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1527,7 +1527,7 @@ async fn sqlite_outbox_cancels_work_from_another_client() { JobState::Cancelled ); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1589,7 +1589,7 @@ async fn sqlite_completion_event_follows_external_cancelled_state() { assert_eq!(event.job.id, inserted.job.row.id); assert_eq!(event.job.state, JobState::Cancelled); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } #[tokio::test] @@ -1630,7 +1630,7 @@ async fn sqlite_fetches_and_discards_unregistered_kinds() { .await .unwrap(); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); } /// An extension that claims every available job itself and records the jobs @@ -1796,7 +1796,7 @@ async fn extension_claimed_rows_fail_undecodable_attempts_like_river_claims() { }) .await .expect("decodable claimed job did not complete"); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); // Both attempts, including the undecodable row's, finished once. let mut finished = pilot.finished.lock().unwrap().clone(); finished.sort_unstable(); @@ -1847,7 +1847,7 @@ async fn set_state_extension_sees_jobs_deleted_while_worked() { }) .await .expect("set-state extension never saw the deleted job"); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); // The job's row is gone, so only its ID reaches the extension. assert!(!pilot.set_state_rows.lock().unwrap().contains(&deleted.id())); diff --git a/rust/riverqueue/tests/stuck_jobs.rs b/rust/riverqueue/tests/stuck_jobs.rs index 906efb88a..1277e8180 100644 --- a/rust/riverqueue/tests/stuck_jobs.rs +++ b/rust/riverqueue/tests/stuck_jobs.rs @@ -134,7 +134,7 @@ async fn stuck_log_line_reports_the_worker_timeout() { }) .await .expect("stuck job logged"); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); assert_eq!(*lines.timeouts.lock().unwrap(), ["Some(5ms)"]); sqlite_cleanup(pool, path).await; diff --git a/rust/riverqueue/tests/work_extensions.rs b/rust/riverqueue/tests/work_extensions.rs index 1b9af082e..2d2ea0010 100644 --- a/rust/riverqueue/tests/work_extensions.rs +++ b/rust/riverqueue/tests/work_extensions.rs @@ -249,7 +249,7 @@ async fn work_until(client: &Client, kind: EventKind, job_id: i64) -> JobRow { }) .await .expect("job event"); - run.shutdown().await.unwrap(); + run.stop().await.unwrap(); row } diff --git a/rust/riverqueue/tests/yugabyte.rs b/rust/riverqueue/tests/yugabyte.rs index ad68f9418..efd3fbe94 100644 --- a/rust/riverqueue/tests/yugabyte.rs +++ b/rust/riverqueue/tests/yugabyte.rs @@ -247,7 +247,7 @@ async fn polls_without_listen_notify() { assert_eq!(job.id, cancellable.job.row.id, "{server:?}"); assert_eq!(job.state, JobState::Cancelled, "{server:?}"); - tokio::time::timeout(TIMEOUT, run.shutdown()) + tokio::time::timeout(TIMEOUT, run.stop()) .await .expect("the client should stop") .unwrap();