Increment retries on tried Chunks

.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

.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:

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,24 +48,27 @@ 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')]
 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')]

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

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))
             }

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<'_>,

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,19 +122,26 @@ 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]
 
 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),
+        )
+    ),
 )
 
 

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.

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;