diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 354e224d..55f052ce 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,7 +22,7 @@ jobs: steps: - uses: actions/checkout@0c366fd6a839edf440554fa01a7085ccba70ac98 - uses: ./.github/actions/setup-nix-direnv-rust - - run: just test-integration + - run: OPSQUEUE_PYTEST_TIMEOUT=600 just test-integration lint: name: Lint & Style Check runs-on: ubuntu-latest diff --git a/.semaphore/semaphore.yml b/.semaphore/semaphore.yml index 38915015..084c700c 100644 --- a/.semaphore/semaphore.yml +++ b/.semaphore/semaphore.yml @@ -80,7 +80,7 @@ blocks: jobs: - name: "Run Integration tests (via Nix)" commands: - - "just nix-test-integration" + - "OPSQUEUE_PYTEST_TIMEOUT=600 just nix-test-integration" - name: "Style" dependencies: [] skip: diff --git a/justfile b/justfile index ee1baa47..5a5c542c 100644 --- a/justfile +++ b/justfile @@ -2,6 +2,10 @@ _default: just --list --unsorted +# Timeout (seconds) for Python integration test recipes. +# Override with OPSQUEUE_PYTEST_TIMEOUT, e.g. OPSQUEUE_PYTEST_TIMEOUT=600 just nix-test-integration +pytest_timeout_seconds := env_var_or_default("OPSQUEUE_PYTEST_TIMEOUT", "60") + # Build-and-run the opsqueue binary (development profile) [group('run')] run *OPSQUEUE_ARGS: @@ -44,10 +48,12 @@ test-unit *TEST_ARGS: test-integration *TEST_ARGS: build-bin build-python #!/usr/bin/env bash set -euo pipefail + export OPSQUEUE_BIN="$PWD/target/debug/opsqueue" + cd libs/opsqueue_python source "./.setup_local_venv.sh" - timeout 600 pytest --color=yes {{TEST_ARGS}} + timeout {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} # Python integration test suite, using artefacts built through Nix. Args are forwarded to pytest [group('nix')] @@ -55,13 +61,14 @@ nix-test-integration *TEST_ARGS: nix-build-bin #!/usr/bin/env bash set -euo pipefail nix_build_python_library_dir=$(just nix-build-python) + nix_build_bin_dir=$(just nix-build-bin) cd libs/opsqueue_python/tests export PYTHONPATH="${nix_build_python_library_dir}/lib/python3.13/site-packages" - export OPSQUEUE_VIA_NIX=true + export OPSQUEUE_BIN="${nix_build_bin_dir}/bin/opsqueue" export RUST_LOG="opsqueue=debug" - timeout 600 pytest --color=yes {{TEST_ARGS}} + timeout {{pytest_timeout_seconds}} pytest --color=yes {{TEST_ARGS}} # Run all linters, fast and slow [group('lint')] @@ -80,6 +87,11 @@ lint-heavy: clippy mypy [group('lint')] clippy: cargo clippy --no-deps --fix --allow-dirty --allow-staged -- -Dwarnings + cargo clippy --no-deps -- -Dwarnings + cargo clippy --no-deps --fix --allow-dirty --allow-staged --no-default-features -- -Dwarnings + cargo clippy --no-deps --no-default-features -- -Dwarnings + cargo clippy --no-deps --fix --allow-dirty --allow-staged --all-features -- -Dwarnings + cargo clippy --no-deps --all-features -- -Dwarnings # Python static analysis type-checker [group('lint')] diff --git a/libs/opsqueue_python/pyproject.toml b/libs/opsqueue_python/pyproject.toml index 35732e91..88016e62 100644 --- a/libs/opsqueue_python/pyproject.toml +++ b/libs/opsqueue_python/pyproject.toml @@ -54,4 +54,4 @@ addopts = "-n 4 --dist=worksteal" # Individual tests should be very fast. They should never take multiple seconds # If after 120sec (accomodating for a toaster-like PC, or an overloaded Semaphore runner) # there is no progress, assume a deadlock -# timeout=120 +timeout=20 diff --git a/libs/opsqueue_python/src/common.rs b/libs/opsqueue_python/src/common.rs index e5d2d92f..32f666c7 100644 --- a/libs/opsqueue_python/src/common.rs +++ b/libs/opsqueue_python/src/common.rs @@ -230,8 +230,7 @@ impl Chunk { tracing::debug!("Fetching chunk content from object store: submission_id={}, prefix={}, chunk_index={}", c.submission_id, prefix, c.chunk_index); let res = object_store_client .retrieve_chunk(&prefix, c.chunk_index, ChunkType::Input) - .await? - .to_vec(); + .await?; tracing::debug!("Fetched chunk content: {res:?}"); (res, Some(prefix)) } diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs index 7d4266c6..41f9399d 100644 --- a/libs/opsqueue_python/src/producer.rs +++ b/libs/opsqueue_python/src/producer.rs @@ -123,6 +123,8 @@ impl ProducerClient { /// /// Will return an error if the submission is already complete, failed, or /// cancelled, or if the submission could not be found. + #[allow(clippy::type_complexity)] + #[allow(clippy::result_large_err)] pub fn cancel_submission( &self, py: Python<'_>, @@ -285,6 +287,7 @@ impl ProducerClient { } #[allow(clippy::type_complexity)] + #[allow(clippy::result_large_err)] pub fn try_stream_completed_submission_chunks( &self, py: Python<'_>, @@ -314,6 +317,7 @@ impl ProducerClient { #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] #[allow(clippy::type_complexity)] + #[allow(clippy::result_large_err)] pub fn run_submission_chunks( &self, py: Python<'_>, @@ -365,6 +369,7 @@ impl ProducerClient { /// This interval is then doubled for each subsequent poll, /// until we check every few seconds. #[allow(clippy::type_complexity)] + #[allow(clippy::result_large_err)] pub fn blocking_stream_completed_submission_chunks( &self, py: Python<'_>, diff --git a/libs/opsqueue_python/tests/conftest.py b/libs/opsqueue_python/tests/conftest.py index 4b0ebb41..ea832fe6 100644 --- a/libs/opsqueue_python/tests/conftest.py +++ b/libs/opsqueue_python/tests/conftest.py @@ -31,11 +31,8 @@ class OpsqueueProcess: @functools.cache def opsqueue_bin_location() -> Path: - if os.environ.get("OPSQUEUE_VIA_NIX"): - deriv_path = ( - subprocess.check_output(["just", "nix-build-bin"]).decode("utf-8").strip() - ) - return Path(deriv_path) / "bin" / "opsqueue" + if explicit_bin := os.environ.get("OPSQUEUE_BIN", "").strip(): + return Path(explicit_bin) else: subprocess.run( ["cargo", "build", "--quiet", "--bin", "opsqueue"], @@ -70,6 +67,8 @@ def opsqueue_service( # temp_dbname = f"/tmp/opsqueue_tests-{uuid.uuid4()}.db" command = [ + "setpriv", + "--pdeathsig=SIGKILL", str(opsqueue_bin_location()), "--port", str(port), @@ -81,9 +80,11 @@ def opsqueue_service( env["RUST_LOG"] = "off" with subprocess.Popen(command, cwd=PROJECT_ROOT, env=env) as process: + assert process.poll() is None, "Opsqueue process failed to start" try: wrapper = OpsqueueProcess(port=port, process=process) yield wrapper + assert process.poll() is None, "Opsqueue process failed during run" finally: process.terminate() @@ -121,11 +122,12 @@ def background_process( @contextmanager def multiple_background_processes( function: Callable[[int], None], count: int -) -> Generator[None, None, None]: +) -> Generator[list[multiprocess.Process], None, None]: with ExitStack() as stack: - for p in range(count): + yield [ stack.enter_context(background_process(function, args=(p,))) - yield + for p in range(count) + ] type StrategyDescription = str | tuple[str, str, StrategyDescription] @@ -133,7 +135,13 @@ def multiple_background_processes( basic_strategies: Iterable[StrategyDescription] = ("Random", "Newest", "Oldest") any_strategies: Iterable[StrategyDescription] = ( *(basic_strategies), - *(("PreferDistinct", "id", s) for s in basic_strategies), + *( + ("PreferDistinct", "id", s) + for s in ( + *basic_strategies, + *(("PreferDistinct", "second_id", s) for s in basic_strategies), + ) + ), ) diff --git a/libs/opsqueue_python/tests/test_roundtrip.py b/libs/opsqueue_python/tests/test_roundtrip.py index ad105414..39505cfe 100644 --- a/libs/opsqueue_python/tests/test_roundtrip.py +++ b/libs/opsqueue_python/tests/test_roundtrip.py @@ -62,9 +62,78 @@ def run_consumer() -> None: assert res == sum(range(1, 101)) -def test_empty_submission( +def test_complete_then_fail_chunks( opsqueue: OpsqueueProcess, any_consumer_strategy: StrategyDescription ) -> None: + """ + A most test that round-trips all chunks as completed and failed. + This simulates the consumer completing chunks while the server released the reservation. + + Each consumer completes 1/n_consumers chunks before the reservation expired + Each consumer receives a reservation expiration for each chunk + Each consumer completes 1/n_consumers chunks after the reservation expired + + Registration expirations are looped via the client to allow in-flight completions to still be registered + """ + producer_client = ProducerClient( + f"localhost:{opsqueue.port}", + "file:///tmp/opsqueue/test_complete_then_fail_chunks", + ) + n_ops = 100 + chunk_size = 10 + n_chunks = n_ops // chunk_size + n_consumers = 3 + assert n_chunks // n_consumers > 1, ( + "Need more than one chunk per consumer for this test" + ) + + def run_consumer(_consumer_id: int) -> None: + consumer_client = ConsumerClient( + f"localhost:{opsqueue.port}", + "file:///tmp/opsqueue/test_complete_then_fail_chunks", + ) + strategy = strategy_from_description(any_consumer_strategy) + + for i in range(n_chunks): + [chunk] = consumer_client.reserve_chunks(strategy=strategy) + if i % n_consumers == 0: + # Each consumers completes 1/n_consumers chunks before the reservation expired. + consumer_client.complete_chunk( + chunk.submission_id, + chunk.submission_prefix, + chunk.chunk_index, + chunk.input_content, + ) + # Simulate the reservation expiring while the consumer's completion wasn't registered. + consumer_client.fail_chunk( + chunk.submission_id, + chunk.submission_prefix, + chunk.chunk_index, + "Simulated failure", + ) + if i % n_consumers == 1: + # Each consumers completes 1/n_consumers chunks after its reservation expired. + consumer_client.complete_chunk( + chunk.submission_id, + chunk.submission_prefix, + chunk.chunk_index, + chunk.input_content, + ) + + with multiple_background_processes(run_consumer, n_consumers) as _consumers: + input_iter = range(0, n_ops) + + output_iter: Iterator[int] = producer_client.run_submission( + input_iter, + chunk_size=chunk_size, + strategic_metadata={"id": 42, "second_id": 69}, + ) + res = sum(output_iter) + + assert res == sum(range(0, n_ops)) + + +def test_empty_submission(opsqueue: OpsqueueProcess) -> None: """ Empty submissions ought to be supported without problems. Opsqueue should immediately consider these 'completed' @@ -208,8 +277,8 @@ def test_many_consumers( opsqueue: OpsqueueProcess, any_consumer_strategy: StrategyDescription ) -> None: """ - Ensure the system still works if we have many consumers concurrently - working on thes same submission + Ensure the system still works if we have many consumers concurrently working + on the same submission """ producer_client = ProducerClient( f"localhost:{opsqueue.port}", "file:///tmp/opsqueue/test_many_consumers" @@ -362,7 +431,7 @@ def assert_submission_failed_has_metadata(x: SubmissionFailed) -> None: def test_cancel_submission_not_found( - opsqueue: OpsqueueProcess, any_consumer_strategy: StrategyDescription + opsqueue: OpsqueueProcess, ) -> None: """Attempting to cancel a submission that doesn't exist raises a SubmissionNotFoundError. @@ -377,7 +446,7 @@ def test_cancel_submission_not_found( def test_cancel_in_progress( - opsqueue: OpsqueueProcess, any_consumer_strategy: StrategyDescription + opsqueue: OpsqueueProcess, ) -> None: """Cancelling a submission that is in progress succeeds and returns none. Attempting to cancel a submission that is already cancelled should raise a @@ -400,7 +469,7 @@ def test_cancel_in_progress( def test_cancel_already_cancelled( - opsqueue: OpsqueueProcess, any_consumer_strategy: StrategyDescription + opsqueue: OpsqueueProcess, ) -> None: """Attempting to cancel a submission that is already cancelled should raise a SubmissionNotCancellableError. diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index e5998f64..1a27aa1c 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -3,6 +3,7 @@ use std::fmt::Display; use std::time::Duration; use crate::common::StrategicMetadataMap; +#[cfg(feature = "server-logic")] use crate::E; use chrono::{DateTime, Utc}; use ux::u63; diff --git a/opsqueue/src/consumer/client.rs b/opsqueue/src/consumer/client.rs index 1105cb02..8941365f 100644 --- a/opsqueue/src/consumer/client.rs +++ b/opsqueue/src/consumer/client.rs @@ -1,7 +1,11 @@ use std::{ collections::HashMap, + error::Error, str::FromStr, - sync::{atomic::AtomicBool, Arc}, + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Arc, + }, time::Duration, }; @@ -39,12 +43,6 @@ use super::{ use backon::{BackoffBuilder, FibonacciBuilder, Retryable}; -type InFlightRequests = Arc< - Mutex<( - usize, - HashMap>, - )>, ->; type WebsocketTcpStream = WebSocketStream>; /// A wrapper around the actual client, @@ -158,6 +156,50 @@ fn retry_policy() -> impl BackoffBuilder { .without_max_times() } +#[allow(clippy::type_complexity)] +#[derive(Debug, Clone)] +struct InFlightRequests( + Arc<( + AtomicUsize, + Mutex>>, + )>, +); + +impl InFlightRequests { + fn next_nonce(&self) -> usize { + self.0 .0.fetch_add(1, Ordering::SeqCst) + } + + async fn next_nonce_with_oneshot( + &self, + ) -> (usize, oneshot::Receiver) { + let (oneshot_sender, oneshot_receiver) = oneshot::channel(); + let mut guard = self.0 .1.lock().await; + // This is called within the lock, so we know the nonce and the oneshot_sender are inserted atomically. + let nonce = self.next_nonce(); + guard.insert(nonce, oneshot_sender); + (nonce, oneshot_receiver) + } + + async fn send( + &self, + envelop: Envelope, + ) -> Result<(), SyncServerToClientResponse> { + let mut in_flight_requests = self.0 .1.lock().await; + let oneshot_sender = in_flight_requests + .remove(&envelop.nonce) + .expect("Received response with nonce that matches none of the open requests"); + oneshot_sender.send(envelop.contents) + } + + async fn clear(&self) { + let mut in_flight_requests = self.0 .1.lock().await; + // This is called within the lock, so we know the nonce and the `onceshot_sender`s are cleared atomically. + self.0 .0.store(0, Ordering::SeqCst); + in_flight_requests.clear(); + } +} + #[derive(Debug)] pub struct Client { in_flight_requests: InFlightRequests, @@ -177,7 +219,8 @@ impl Client { let endpoint_uri = Uri::from_str(&endpoint_url)?; tracing::debug!("Connecting to: {}", endpoint_uri); - let in_flight_requests: InFlightRequests = Arc::new(Mutex::new((0, HashMap::new()))); + let in_flight_requests = + InFlightRequests(Arc::new((AtomicUsize::new(0), Mutex::new(HashMap::new())))); let (websocket_conn, _resp) = tokio_tungstenite::connect_async(endpoint_uri).await?; let (ws_sink, mut ws_stream) = websocket_conn.split(); @@ -282,16 +325,42 @@ impl Client { let msg: ServerToClientMessage = msg.try_into().expect("Unparsable ServerToClientMessage"); match msg { ServerToClientMessage::Sync(envelope) => { - let mut in_flight_requests = in_flight_requests.lock().await; - // Handle the response to some earlier request - let oneshot_receiver = in_flight_requests.1.remove(&envelope.nonce).expect("Received response with nonce that matches none of the open requests"); - let _ = oneshot_receiver.send(envelope.contents); - + let Err(contents) = in_flight_requests.send(envelope).await else { continue; }; + tracing::warn!("Failed to send envelope to in-flight requests"); + match contents { + SyncServerToClientResponse::ChunksReserved(reservation) => { + let Ok(chunks) = reservation else { continue; }; + for chunk in chunks.iter() { + let chunk_id = ChunkId::from((chunk.0.submission_id, chunk.0.chunk_index)); + // Send message to the server to indicate that we are no longer reserving this chunk, so that it can be re-reserved by other consumers. + let Err(internal_error): Result<_, InternalConsumerClientError> = Self::async_request( + &in_flight_requests, + &ws_sink, + ClientToServerMessage::FailChunk { + id: chunk_id, + failure: "Requester disappeared".into(), + }, + ).await else { continue; }; + tracing::error!(error = &internal_error as &dyn Error, chunk_id = ?chunk_id, "Failed to return unhandled reservation to server"); + } + }, + } }, ServerToClientMessage::Async(msg) => { match msg { AsyncServerToClientMessage::ChunkReservationExpired(chunk_id) => { tracing::info!("Server indicated that we took too long with {chunk_id:?}, and now our reservation has expired."); + // Send message to the server to indicate that we are no longer reserving this chunk, so that it can be re-reserved by other consumers. + // This round-trip is necessary to avoid a race condition where we sent the success or failure of the chunk to the server, while we lost the reservation window. + let Err(internal_error): Result<_, InternalConsumerClientError> = Self::async_request( + &in_flight_requests, + &ws_sink, + ClientToServerMessage::FailChunk { + id: chunk_id, + failure: "Reservation expired".into(), + }, + ).await else { continue; }; + tracing::error!(error = &internal_error as &dyn Error, chunk_id = ?chunk_id, "Failed to return expired reservation to server"); }, } } @@ -305,43 +374,34 @@ impl Client { } // Clear any and all in-flight requests on exit of the background task. // This ensures that any waiting requests immediately return with an error as well. - let mut in_flight_requests = in_flight_requests.lock().await; - in_flight_requests.1.clear(); - in_flight_requests.0 = 0; + in_flight_requests.clear().await; } async fn sync_request( &self, request: ClientToServerMessage, ) -> Result { - let (oneshot_sender, oneshot_receiver) = oneshot::channel(); - { - let mut in_flight_requests = self.in_flight_requests.lock().await; - let nonce = in_flight_requests.0; - in_flight_requests.0 = in_flight_requests.0.wrapping_add(1); - let envelope = Envelope { - nonce, - contents: request, - }; - in_flight_requests.1.insert(nonce, oneshot_sender); - let () = self.ws_sink.lock().await.send(envelope.into()).await?; - } + let (nonce, oneshot_receiver) = self.in_flight_requests.next_nonce_with_oneshot().await; + let envelope = Envelope { + nonce, + contents: request, + }; + let () = self.ws_sink.lock().await.send(envelope.into()).await?; let resp = oneshot_receiver.await?; Ok(resp) } async fn async_request( - &self, + in_flight_requests: &InFlightRequests, + ws_sink: &Mutex>, request: ClientToServerMessage, ) -> Result<(), InternalConsumerClientError> { - let mut in_flight_requests = self.in_flight_requests.lock().await; - let nonce = in_flight_requests.0; - in_flight_requests.0 = in_flight_requests.0.wrapping_add(1); + let nonce = in_flight_requests.next_nonce(); let envelope = Envelope { nonce, contents: request, }; - let () = self.ws_sink.lock().await.send(envelope.into()).await?; + let () = ws_sink.lock().await.send(envelope.into()).await?; Ok(()) } @@ -363,8 +423,12 @@ impl Client { id: ChunkId, output_content: chunk::Content, ) -> Result<(), InternalConsumerClientError> { - self.async_request(ClientToServerMessage::CompleteChunk { id, output_content }) - .await + Self::async_request( + &self.in_flight_requests, + &self.ws_sink, + ClientToServerMessage::CompleteChunk { id, output_content }, + ) + .await } pub async fn fail_chunk( @@ -372,8 +436,12 @@ impl Client { id: ChunkId, failure: String, ) -> Result<(), InternalConsumerClientError> { - self.async_request(ClientToServerMessage::FailChunk { id, failure }) - .await + Self::async_request( + &self.in_flight_requests, + &self.ws_sink, + ClientToServerMessage::FailChunk { id, failure }, + ) + .await } } diff --git a/opsqueue/src/consumer/dispatcher/metastate.rs b/opsqueue/src/consumer/dispatcher/metastate.rs index 8e743a3f..5267bd01 100644 --- a/opsqueue/src/consumer/dispatcher/metastate.rs +++ b/opsqueue/src/consumer/dispatcher/metastate.rs @@ -20,12 +20,17 @@ impl MetaState { pub fn decrement(&self, key: &str, val: &MetaStateVal) { let ripe_for_removal = { - let meta_state_field = self - .0 - .get(key) - .expect("decrements should be paired with increments."); - meta_state_field.decrement(val); - meta_state_field.is_empty() + if let Some(meta_state_field) = self.0.get(key) { + meta_state_field.decrement(val); + meta_state_field.is_empty() + } else { + tracing::error!( + key = key, + val = val, + "decrements should be paired with increments." + ); + false + } }; if ripe_for_removal { // The actual removal happens after the main code