Implement pausing for submissions by SemMulder · Pull Request #132 · channable/opsqueue

SemMulder · 2026-07-15T15:20:15Z

Introduce submissions_paused and chunks_paused tables
(alongside the existing submissions_{completed,failed,cancelled} and
chunks_{completed,failed} tables).

A submission can now be created in a Paused state. It's then stored in
submissions_paused and its chunks are stored in chunks_paused.
Because paused chunks are not in the chunks table, the consumer
dispatcher naturally skips them without any changes to the dispatch
query.

Unpausing moves the submission and the chunks to submissions and
chunks and notifies waiting consumers.

Paused submissions are cancellable; cancel_submission now handles
the case where the submission is found in submissions_paused.

We don't allow pausing submissions after creation. That proved to
have too many edge cases we would need to resolve.

Also cleans up some code encountered along the way.

SemMulder · 2026-07-16T08:07:42Z

TODO: How to deal with reserved chunks when pausing?

SemMulder · 2026-07-17T15:13:55Z

TODO: How to deal with reserved chunks when pausing?

Discussed this with @ReinierMaas yesterday, we tried to make it work s.t. chunk completions for cancelled and paused submissions would still be processed. However, implementing that proved to be difficult:

What if the completed chunk was the last one? Would the submission status then progress from cancelled/paused to completed?
There were a lot of subtleties that needed to be thought through around multiple consumers that could be working on the same chunk.

After struggling with it for a while, I thought it would be easier to invoke the idempotency assumption we have for how consumers process chunks, and just ignore the completion in case the submission is cancelled or paused. This results in less edge-cases and simpler code in the end, at the cost of slightly higher compute cost. But I think it's worth the trade-off: there are enough edge-cases in the codebase as it is :).

@ReinierMaas, let me know if you disagree ;).

SemMulder · 2026-07-17T15:28:28Z

    pass


-class ChunkNotFoundError(IncorrectUsageError):


Note that ChunkNotFoundError can be dropped since it is (and was) dead code.

SemMulder · 2026-07-17T15:30:57Z

+                )
+                .await
            })
-            .await;


Note that we were silently dropping errors here before, which went unnoticed because of the underscore prefix of _chunk_size.

SemMulder · 2026-07-22T15:36:48Z

After struggling with it for a while, I thought it would be easier to invoke the idempotency assumption we have for how consumers process chunks, and just ignore the completion in case the submission is cancelled or paused.

Thought some more: if the assumption is that a paused job should not have any work being done, this will violate that. Maybe easier to just only allow pausing of not yet started submissions for now?

Other option is to make pausing async, and only transition from InProgress to Paused after the pending chunks have been completed/failed. But that will likely blow up this PR. TBD.

SemMulder · 2026-07-23T08:30:11Z

After struggling with it for a while, I thought it would be easier to invoke the idempotency assumption we have for how consumers process chunks, and just ignore the completion in case the submission is cancelled or paused.

Thought some more: if the assumption is that a paused job should not have any work being done, this will violate that. Maybe easier to just only allow pausing of not yet started submissions for now?

Other option is to make pausing async, and only transition from InProgress to Paused after the pending chunks have been completed/failed. But that will likely blow up this PR. TBD.

After sleeping on this for a night, I'm leaning towards ignoring the problem, getting this PR merged, and opening a new ticket to resolve this in the future. Otherwise this PR will become too big IMO (and it is already quite hefty).

Will discuss with @ReinierMaas.

ReinierMaas · 2026-07-23T10:10:18Z

I thought a bit on it. I think the easiest way is to only allow a DAG:

active (paused -> running) -> completed (cancelled | failed | success)

We can insert a submissions in either paused or running and then only move forward from there to keep it as simple as possible.

From paused we can move to any of the completed states, say we don't have any chunks that is an automatic completed successfully. I am not sure how we can move from paused to failed, maybe you have an example for that? We can move from paused to cancelled by cancelling the submission instead of unpausing. If a paused submissions is unpaused it moves to running.

From running we can move to to any of the completed states:

no chunks or all chunks succeeded -> success
cancellation of the submission -> cancelled
a chunk exceeds its retries -> failed

If we need more complicated logic we can pick that up as a separate issue.

To be abundantly clear how the DAG state transitions would look like:

graph LR
    %% Active States
    P((Paused))
    R((Running))

    %% Completed States
    S([Success])
    F([Failed])
    C([Cancelled])

    %% Internal Active Transitions
    P -- unpause --> R

    %% Transitions to Completed States
    P -- no chunks --> S
    P -.-> F
    P -- user cancellation --> C

    R -- no chunks / all succeed --> S
    R -- chunk retries exceeded --> F
    R -- user cancellation --> C

SemMulder · 2026-07-23T10:57:30Z

I am not sure how we can move from paused to failed

I think this is impossible, it would always go via running.

We can insert a submissions in either paused or running and then only move forward from there to keep it as simple as possible.

This would indeed make it simpler. In practice, that would amount to removing the pause endpoint from this PR. For our current use-cases this would be fine. That is, we currently only need the ability to submit paused submissions, we don't need to pause them mid-way.

Note that this does mean that we will have to implement /job/return as a no-op once we start implementing JM Delegation. However, I believe that's fine. From there:

This request is non-binding and can be ignored for tasks that have started running already.

But will double-check that with @radekchannable as well.

SemMulder · 2026-07-23T12:14:21Z

But will double-check that with @radekchannable as well.

Double-checked, we indeed don't have to pause anything when JM asks.

SemMulder · 2026-07-23T12:52:24Z

I am not sure how we can move from paused to failed

I think this is impossible, it would always go via running.

This was wrong 😅, you can cancel a paused submission, then it would end up in failed.

SemMulder · 2026-07-23T14:37:43Z

-    #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))]
-    #[allow(clippy::result_large_err, clippy::type_complexity)]
-    /// Submit chunks and then stream the completed output chunks.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if upload, submission creation, or streaming fails.
-    pub fn run_submission_chunks(
-        &self,
-        py: Python<'_>,
-        chunk_contents: Py<PyIterator>,
-        metadata: Option<submission::Metadata>,
-        strategic_metadata: Option<StrategicMetadataMap>,
-        chunk_size: Option<i64>,
-        otel_trace_carrier: CarrierMap,
-    ) -> CPyResult<
-        PyChunksIter,
-        E![
-            FatalPythonException,
-            errors::SubmissionFailed,
-            ChunksStorageError,
-            InternalProducerClientError,
-        ],
-    > {
-        let submission_id = self
-            .insert_submission_chunks(
-                py,
-                chunk_contents,
-                metadata,
-                strategic_metadata,
-                chunk_size,
-                otel_trace_carrier,
-            )
-            .map_err(|CError(e)| {
-                CError(match e {
-                    L(e) => L(e),
-                    R(e) => R(R(e)),
-                })
-            })?;
-        let res = self
-            .blocking_stream_completed_submission_chunks(py, submission_id)
-            .map_err(|CError(e)| {
-                CError(match e {
-                    L(e) => L(e),
-                    R(L(e)) => R(L(e)),
-                    R(R(e)) => R(R(R(e))),
-                })
-            })?;
-        Ok(res)
-    }
-


This was dead code, because it is re-implemented on the Python side.

Shouldn't we then prefer the Rust side over the Python side as that is exposed to all downstream consumers.

Note that this is in opsqueue_python, so it's not exposed to any downstream consumers ATM :).

Copilot

Pull request overview

Implements first-class “paused submissions” in opsqueue by storing paused submissions/chunks in separate tables, exposing unpause functionality via the producer API, and updating Rust + Python clients and tests to support pause/unpause and related status reporting.

Changes:

Add submissions_paused / chunks_paused tables and extend submission status to include Paused.
Add producer endpoint + clients for unpausing, and adjust producer insertion to optionally start paused.
Update metrics, error types, Python bindings, and roundtrip tests (including configurable timeouts).

Reviewed changes

Copilot reviewed 18 out of 19 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
opsqueue/src/prometheus.rs	Adds paused/unpaused counters and updates chunk backlog count typing.
opsqueue/src/producer/server.rs	Adds `/submissions/unpause/{submission_id}` endpoint and avoids notifying consumers for paused inserts.
opsqueue/src/producer/common.rs	Extends insert request payload with `paused: bool`.
opsqueue/src/producer/client.rs	Adds Rust producer client `unpause_submission` + updates tests for new count types/status.
opsqueue/src/consumer/strategy.rs	Updates tests to pass new `paused` parameter on submission creation.
opsqueue/src/consumer/client.rs	Updates tests to pass new `paused` parameter on submission creation.
opsqueue/src/common/submission.rs	Core implementation: paused submission type/status, insert paused, unpause/cancel paused, updated count APIs.
opsqueue/src/common/errors.rs	Removes `ChunkNotFound` error type.
opsqueue/src/common/chunk.rs	Adds paused chunk storage ops; changes chunk completion/retry logic and count types.
opsqueue/migrations/20260715143000_pausing.up.sql	Introduces `submissions_paused` and `chunks_paused` schema.
opsqueue/migrations/20260715143000_pausing.down.sql	Drops paused tables on rollback.
libs/opsqueue_python/tests/test_roundtrip.py	Adds timeouts to blocking calls + new tests for timeout and pause/unpause behavior.
libs/opsqueue_python/src/producer.rs	Exposes pause/unpause and adds timeout support for blocking streaming methods.
libs/opsqueue_python/src/lib.rs	Exports `SubmissionPaused` to Python module.
libs/opsqueue_python/src/errors.rs	Removes chunk-not-found mapping; maps `tokio::time::Elapsed` to Python `TimeoutError`.
libs/opsqueue_python/src/common.rs	Adds Python-side `SubmissionStatus.Paused` and `SubmissionPaused` model.
libs/opsqueue_python/python/opsqueue/producer.py	Adds `paused` + `timeout` parameters and exposes `unpause_submission`.
libs/opsqueue_python/python/opsqueue/exceptions.py	Removes `ChunkNotFoundError` exception.

Comments suppressed due to low confidence (2)

opsqueue/src/common/submission.rs:559

With the counter increment moved to unpause_submission (after the transaction commits), unpause_submission_raw should not also increment SUBMISSIONS_UNPAUSED_COUNTER, otherwise successful unpauses will be double-counted.

            Err(E::R(SubmissionNotFound(id)))
        } else {
            counter!(crate::prometheus::SUBMISSIONS_UNPAUSED_COUNTER).increment(1);
            Ok(())
        }

opsqueue/src/common/chunk.rs:318

complete_chunk now bubbles up SubmissionNotFound, which is expected when a submission is cancelled while a chunk is reserved (submission is no longer in submissions). This turns an expected race into an error (and ends up logged as an error by the completer). Consider treating SubmissionNotFound as a non-fatal no-op here, while still propagating real database errors.

        .await?;

        counter!(crate::prometheus::CHUNKS_COMPLETED_COUNTER).increment(1);
        Ok(())
    }

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

SemMulder · 2026-08-04T14:28:41Z

+        conn.transaction(move |mut tx| {
+            Box::pin(async move {
+                unpause_submission_raw(id, &mut tx).await?;
+                super::chunk::db::restore_paused_chunks(id, &mut tx).await?;
+                Ok(())
+            })
+        })
+        .await
+    }


This pattern is there for all of the other metrics as well. I would suggest going for consistency in this PR, and opening an issue for solving it separately.

SemMulder · 2026-08-04T14:29:03Z

+        counter!(crate::prometheus::SUBMISSIONS_PAUSED_COUNTER).increment(1);
+        counter!(crate::prometheus::SUBMISSIONS_TOTAL_COUNTER).increment(1);
+        counter!(crate::prometheus::CHUNKS_TOTAL_COUNTER).increment(chunks_total);
+        res


See the comment above.

ReinierMaas

I think we are nearly there. I did find some changes we should still make but they are minor and not architectural.

ReinierMaas · 2026-07-24T11:37:01Z

+        with pytest.raises(TimeoutError):
+            producer_client.run_submission(
+                [1],
+                chunk_size=1,
+                timeout=0.1,
+            )


We always check something on the exception raised by pytest, i.e. from a bit higher in the file:

# We expect the intended attributes to be there: assert isinstance(exc_info.value.failure, str) assert isinstance(exc_info.value.submission, SubmissionFailed) assert isinstance(exc_info.value.chunk, ChunkFailed)

Just checking the instance of exc_info would be enough here, we have experienced pytest catching the wrong exceptions and giving the green checkmark incidentally

Suggested change

with pytest.raises(TimeoutError):

producer_client.run_submission(

[1],

chunk_size=1,

timeout=0.1,

)

with pytest.raises(TimeoutError) as exc_info:

producer_client.run_submission(

[1],

chunk_size=1,

timeout=0.1,

)

assert isinstance(exc_info, TimeoutError)

Hmm, as far as I understand pytest.raises(TimeoutError) already does isinstance(exc_info, TimeoutError)?

As in the following should fail:

with pytest.raised(NotTheRaisedError): raise AnotherError()

It does:

>>> import pytest >>> with pytest.raises(TimeoutError): ... raise Exception("Not timeout error") ... Traceback (most recent call last): File "<stdin>", line 2, in <module> Exception: Not timeout error >>> with pytest.raises(TimeoutError): ... raise TimeoutError() ... >>>

Ah nice, then we no longer need it for the outer layer and only need to use it to validate the data that is included in the exception. Thanks for confirming!

u63 forces us to wrap literals in `u63::new`, and we need to convert to u64 at actual usage sites anyway.

…properly

…ly completed, failed, or cancelled chunks Because of the idempotency assumption for processing chunks, nothing should break if we just ignore the error. Besides, we were already ignoring the error accidentally.

Introduce `submissions_paused` and `chunks_paused` tables (alongside the existing `submissions_{completed,failed,cancelled}` and `chunks_{completed,failed}` tables). A submission can now be created in a Paused state. It's then stored in `submissions_paused` and its chunks are stored in `chunks_paused`. Because paused chunks are not in the `chunks` table, the consumer dispatcher naturally skips them without any changes to the dispatch query. Unpausing moves the submission and the chunks to `submissions` and `chunks` and notifies waiting consumers. Paused submissions are cancellable; `cancel_submission` now handles the case where the submission is found in `submissions_paused`. We don't allow pausing submissions after creation. That proved to have too many edge cases we would need to resolve.

Copilot

Pull request overview

Copilot reviewed 18 out of 19 changed files in this pull request and generated no new comments.

Suppressed comments (3)

opsqueue/src/common/submission.rs:810

submission_status can transiently return Ok(None) during an unpause: if the unpause transaction commits after the submissions lookup (line 809) but before the final submissions_paused lookup, both queries will see “not found” and the function returns None even though the submission exists (now InProgress). To avoid this race, query submissions_paused before submissions (or run the lookups inside a single read transaction / single SQL statement so you get a consistent snapshot).

        // NOTE: The order is important here; a concurrent writer could move a submission
        // from InProgress to Completed/Failed in-between the queries.

        let submission_row = submission_status_in_progress_query(id)
            .fetch_optional(conn.get_inner())

opsqueue/src/common/submission.rs:530

Unpausing a zero-chunk paused submission will move it into submissions with chunks_total = 0, but nothing will ever complete it (no chunks will be processed and maybe_complete_submission is never called on unpause). This leaves the submission stuck InProgress forever. Consider completing the submission during unpause when chunks_done == chunks_total (which covers the zero-chunk case).

            Box::pin(async move {
                unpause_submission_raw(id, &mut tx).await?;
                super::chunk::db::restore_paused_chunks(id, &mut tx).await?;
                Ok(())

opsqueue/src/common/chunk.rs:311

complete_chunk always calls maybe_complete_submission even when complete_chunk_raw didn’t actually move a chunk (e.g. the chunk was already completed/failed/cancelled). In those cases the submission may already be gone from submissions (cancelled submissions are moved out), so maybe_complete_submission can return SubmissionNotFound and turn an “ignored duplicate completion” into a hard error. A robust fix is to have complete_chunk_raw return whether it moved a row and only call maybe_complete_submission when it did.

                crate::common::submission::db::maybe_complete_submission(
                    chunk_id.submission_id,
                    &mut tx,
                )
                .await

Copilot

Pull request overview

Copilot reviewed 18 out of 19 changed files in this pull request and generated no new comments.

Suppressed comments (1)

opsqueue/src/common/chunk.rs:326

complete_chunk now returns Ok(()) even when the chunk row wasn’t moved (e.g., the chunk was already completed/failed/cancelled). In that case, CHUNKS_COMPLETED_COUNTER is still incremented unconditionally, which can overcount completions (notably when the function is called twice for the same chunk). Increment the counter only when the chunk was actually moved.

        .await?;

        counter!(crate::prometheus::CHUNKS_COMPLETED_COUNTER).increment(1);
        Ok(())

Copilot

Pull request overview

Copilot reviewed 18 out of 19 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

libs/opsqueue_python/tests/test_roundtrip.py:756

This call can block indefinitely if the consumer fails to start or the submission never completes. Since this test suite now standardizes on SUBMISSION_COMPLETED_TIMEOUT, pass it here as well to avoid hanging CI on failures.

    with background_process(run_consumer):
        producer_client.blocking_stream_completed_submission(submission_id)
        assert isinstance(

ReinierMaas · 2026-08-13T13:50:23Z

+    /// Unpause a paused submission, making it available to consumers.
+    ///
+    /// Will return an error if the submission is not currently paused.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the submission is not found or if an internal client error occurs.
+    #[allow(clippy::result_large_err, clippy::type_complexity)]
+    pub fn unpause_submission(
+        &self,
+        py: Python<'_>,
+        id: SubmissionId,
+    ) -> CPyResult<


Should we add the pyo3 signature like we do for insert_submission_direct, or is this automatically inferred?

ReinierMaas · 2026-08-13T13:53:44Z

@@ -442,17 +426,39 @@ impl ProducerClient {
        &self,
        py: Python<'_>,
        submission_id: SubmissionId,
+        timeout: Option<f64>,
    ) -> CPyResult<
        PyChunksIter,
        E![
            FatalPythonException,
+            TryFromFloatSecsError,
+            Elapsed,
            errors::SubmissionFailed,
            InternalProducerClientError
        ],
    > {


Do we need to add the pyo3 signature here as well?

Nice addition of the timeout!

ReinierMaas · 2026-08-13T13:57:38Z

+        with pytest.raises(TimeoutError):
+            producer_client.run_submission(
+                [1],
+                chunk_size=1,
+                timeout=0.1,
+            )


Ah nice, then we no longer need it for the outer layer and only need to use it to validate the data that is included in the exception. Thanks for confirming!

ReinierMaas · 2026-08-13T14:04:45Z

    /// Mark a chunk as completed and update related submission state.
    ///
    /// # Errors
    ///
    /// Returns an error if the chunk/submission cannot be updated in the database.
    #[tracing::instrument(skip(conn))]
    pub async fn complete_chunk(
        chunk_id: ChunkId,
        output_content: Option<Vec<u8>>,
        mut conn: impl WriterConnection,
-    ) -> Result<(), E<DatabaseError, E<SubmissionNotFound, ChunkNotFound>>> {
-        let _chunk_size: Result<ChunkSize, E<DatabaseError, E<SubmissionNotFound, ChunkNotFound>>> =
-            conn.transaction(move |mut tx| {
+    ) -> Result<(), E<DatabaseError, SubmissionNotFound>> {
+        let chunks_moved = conn
+            .transaction(move |mut tx| {
                Box::pin(async move {
-                    let completed_work =
+                    let chunks_moved =
                        complete_chunk_raw(chunk_id, output_content, &mut tx).await?;
-                    crate::common::submission::db::maybe_complete_submission(
-                        chunk_id.submission_id,
-                        &mut tx,
-                    )
-                    .await
-                    .map_err(|e| match e {
-                        E::L(e) => E::L(e),
-                        E::R(e) => E::R(E::L(e)),
-                    })?;
-                    Ok(completed_work.unwrap_or_default())
+                    if chunks_moved {
+                        crate::common::submission::db::maybe_complete_submission(
+                            chunk_id.submission_id,
+                            &mut tx,
+                        )
+                        .await?;
+                    } else {
+                        tracing::warn!(
+                            "Could not complete chunk {:?} because it was either: \
+                completed, failed, or cancelled before. Ignoring.",
+                            chunk_id
+                        );
+                    }
+
+                    Result::<bool, E<DatabaseError, SubmissionNotFound>>::Ok(chunks_moved)
                })
            })
-            .await;
+            .await?;

-        counter!(crate::prometheus::CHUNKS_COMPLETED_COUNTER).increment(1);
+        if chunks_moved {
+            counter!(crate::prometheus::CHUNKS_COMPLETED_COUNTER).increment(1);
+        }
        Ok(())
    }


Let's rename chunks_moved to chunk_moved, it became a boolean and we pass a ChunkId down. I was suprised that we update the counter with 1 and not the number of moved chunks to then find out it is a boolean.

ReinierMaas · 2026-08-13T14:41:52Z

+            query_builder.push_values(query_chunks, |mut b, chunk| {
+                b.push_bind(chunk.submission_id)
+                    .push_bind(chunk.chunk_index)
+                    .push_bind(chunk.input_content.clone());


This is cloning the entire input contents, when in database storage is used as opposed to GCS storage backend. The pattern also exists for the other paths (i.e. master), so not something we should change in this branch, but a relative small change can circumvent this cloning:

diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index c27c17e..16662c1 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -558,12 +558,12 @@ pub mod db { /// Returns an error if any batch insert fails. #[tracing::instrument(skip(chunks, conn))] pub async fn insert_many_chunks( - chunks: &[Chunk], + chunks: impl IntoIterator<Item = Chunk>, mut conn: impl WriterConnection, ) -> sqlx::Result<()> { const ROWS_PER_QUERY: usize = 1000; - let mut iter = chunks.iter().peekable(); + let mut iter = chunks.into_iter().peekable(); while iter.peek().is_some() { let query_chunks = iter.by_ref().take(ROWS_PER_QUERY); @@ -573,7 +573,7 @@ pub mod db { query_builder.push_values(query_chunks, |mut b, chunk| { b.push_bind(chunk.submission_id) .push_bind(chunk.chunk_index) - .push_bind(chunk.input_content.clone()); + .push_bind(chunk.input_content); }); let query = query_builder.build(); diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 991744d..7a9fb45 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -418,7 +418,7 @@ pub mod db { &mut tx, ) .await?; - super::chunk::db::insert_many_chunks(&chunks, &mut tx).await?; + super::chunk::db::insert_many_chunks(chunks, &mut tx).await?; Ok(()) } .boxed()

ReinierMaas · 2026-08-13T14:52:38Z

+    /// Unpause a paused submission. Atomically moves it back from `submissions_paused`
+    /// to `submissions` and its chunks from `chunks_paused` to `chunks`.


Maybe we should document that it atomically moves from submissions_paused to submissions_completed if it is a 0-chunk submission that is unpaused.

ReinierMaas · 2026-08-13T15:09:43Z

+                match maybe_complete_submission(submission_id, conn).await {
+                    // Forward our database errors to the caller.
+                    Err(E::L(e)) => return Err(e),
+                    // If the submission ID can't be found, that's too bad, but it's not our problem anymore I guess.
+                    Err(E::R(_)) => {
+                        tracing::warn!(%submission_id, "Presumed zero-length submission not found");
+                    }
+                    // If everything went OK, this *could* still indicate a bug in producer code, so let's just log it.
+                    // Our future selves might thank us.
+                    Ok(true) => {
+                        tracing::debug!(%submission_id, "Zero-length submission marked as completed");
+                    }
+                    // This should never happen. If it does, better log it.
+                    Ok(false) => {
+                        tracing::warn!(%submission_id, "Zero-length submission wasn't zero-length?!");
+                    }


This logging is skipped on the unpausing path. The unpausing path does forward the SubmissionNotFound that is caught here? The unpausing path also does this atomically, i.e. within a transaction.

Also on all the other paths we unconditionally execute the function and here we protect it with the if len == 0 branch... I am not completely sure how to best standardize this.

ReinierMaas · 2026-08-13T15:10:46Z

    ) -> Result<Option<SubmissionStatus>, DatabaseError> {
        // NOTE: The order is important here; a concurrent writer could move a submission
        // from InProgress to Completed/Failed in-between the queries.
+        // TODO: Rewrite the queries here into a single query using `UNION ALL`.


Is this TODO something you want to still resolve? Or are you going to reference an issue TODO(issue_nr): before merging?

ReinierMaas · 2026-08-13T15:13:55Z

        // NOTE: The order is important here; a concurrent writer could move a submission
        // from InProgress to Completed/Failed in-between the queries.


Suggested change

// NOTE: The order is important here; a concurrent writer could move a submission

// from InProgress to Completed/Failed in-between the queries.

// NOTE: The order is important here; a concurrent writer could move a submission

// from Paused to InProgress/Cancelled in-between the queries.

// from InProgress to Completed/Failed/Cancelled in-between the queries.

Also couldn't we prevent this by using an transaction and performing all the queries on the same snapshot? We are using WAL-mode:

WAL mode permits simultaneous readers and writers. It can do this because changes do not overwrite the original database file, but rather go into the separate write-ahead log file. That means that readers can continue to read the old, original, unaltered content from the original database file at the same time that the writer is appending to the write-ahead log. In WAL mode, SQLite exhibits "snapshot isolation". When a read transaction starts, that reader continues to see an unchanging "snapshot" of the database file as it existed at the moment in time when the read transaction started. Any write transactions that commit while the read transaction is active are still invisible to the read transaction, because the reader is seeing a snapshot of database file from a prior moment in time. ~ https://sqlite.org/isolation.html

ReinierMaas · 2026-08-13T15:39:53Z

          ?
        ");
-        let explained = explain_query_plan(&formatted_query, &mut conn).await;
-        assert_non_regressing_query_plan(&formatted_query, &explained);


Why did you remove the assert_non_regressing_query_plan calls? This happened multiple times.

SemMulder force-pushed the sm/allow-pausing-submissions branch from a9f5556 to bcd4161 Compare July 15, 2026 15:32

SemMulder force-pushed the sm/allow-pausing-submissions branch from b09f93b to 7e7052d Compare July 17, 2026 14:24

SemMulder force-pushed the sm/allow-pausing-submissions branch from 990a5d2 to 71ace4e Compare July 17, 2026 15:26

SemMulder commented Jul 17, 2026

View reviewed changes

Comment thread opsqueue/src/common/submission.rs

SemMulder commented Jul 17, 2026

View reviewed changes

Comment thread opsqueue/src/common/submission.rs Outdated

SemMulder force-pushed the sm/allow-pausing-submissions branch 6 times, most recently from e39c8a9 to 82397bd Compare July 22, 2026 15:16

SemMulder force-pushed the sm/allow-pausing-submissions branch 2 times, most recently from 3161c6a to 9fc1444 Compare July 23, 2026 14:36

SemMulder commented Jul 23, 2026

View reviewed changes

SemMulder marked this pull request as ready for review July 23, 2026 15:02

SemMulder requested review from ReinierMaas and Copilot July 23, 2026 15:02

Copilot started reviewing on behalf of SemMulder July 23, 2026 15:03 View session

Copilot AI reviewed Jul 23, 2026

View reviewed changes

ReinierMaas requested changes Jul 24, 2026

View reviewed changes

SemMulder force-pushed the sm/allow-pausing-submissions branch from c9fc39e to 622b0fd Compare August 5, 2026 11:34

SemMulder added 6 commits August 5, 2026 13:41

Use u64 in count_* functions

64f0fe4

u63 forces us to wrap literals in `u63::new`, and we need to convert to u64 at actual usage sites anyway.

Make tests timeout properly, in preparation for showing failure logs …

8780d96

…properly

Make complete_chunk and fail_chunk not error when processing previous…

77c8b11

…ly completed, failed, or cancelled chunks Because of the idempotency assumption for processing chunks, nothing should break if we just ignore the error. Besides, we were already ignoring the error accidentally.

Address feedback

ad14db9

Pull out queries and EXPLAIN the now shared query

22c5d1e

SemMulder force-pushed the sm/allow-pausing-submissions branch from 622b0fd to 22c5d1e Compare August 5, 2026 12:01

SemMulder requested a review from Copilot August 5, 2026 12:08

Copilot started reviewing on behalf of SemMulder August 5, 2026 12:09 View session

Copilot AI reviewed Aug 5, 2026

View reviewed changes

SemMulder force-pushed the sm/allow-pausing-submissions branch 2 times, most recently from 417143a to 9c58c87 Compare August 5, 2026 13:58

SemMulder requested a review from Copilot August 5, 2026 13:59

Copilot started reviewing on behalf of SemMulder August 5, 2026 13:59 View session

Copilot AI reviewed Aug 5, 2026

View reviewed changes

SemMulder force-pushed the sm/allow-pausing-submissions branch from 9c58c87 to 0abcc15 Compare August 5, 2026 14:11

SemMulder requested a review from Copilot August 5, 2026 14:18

Copilot started reviewing on behalf of SemMulder August 5, 2026 14:19 View session

Copilot AI reviewed Aug 5, 2026

View reviewed changes

Comment thread libs/opsqueue_python/src/producer.rs

Address Copilot's suppressed comments

3ee2454

SemMulder force-pushed the sm/allow-pausing-submissions branch from 0abcc15 to 3ee2454 Compare August 5, 2026 14:44

SemMulder requested a review from ReinierMaas August 5, 2026 14:52

ReinierMaas reviewed Aug 13, 2026

View reviewed changes

This was referenced Aug 13, 2026

Remove unnecessary input_content.clone() call #166

Open

Pyo3: Type signatures for all exposed functions #167

Open

		/// Unpause a paused submission. Atomically moves it back from `submissions_paused`
		/// to `submissions` and its chunks from `chunks_paused` to `chunks`.

		// NOTE: The order is important here; a concurrent writer could move a submission
		// from InProgress to Completed/Failed in-between the queries.

Uh oh!

Conversation

SemMulder commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SemMulder commented Jul 16, 2026

Uh oh!

SemMulder commented Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SemMulder Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

SemMulder commented Jul 22, 2026

Uh oh!

SemMulder commented Jul 23, 2026

Uh oh!

ReinierMaas commented Jul 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SemMulder commented Jul 23, 2026

Uh oh!

SemMulder commented Jul 23, 2026

Uh oh!

SemMulder commented Jul 23, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ReinierMaas left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

SemMulder commented Jul 15, 2026 •

edited

Loading

SemMulder commented Jul 17, 2026 •

edited

Loading

SemMulder Jul 17, 2026 •

edited

Loading

ReinierMaas commented Jul 23, 2026 •

edited

Loading

SemMulder Aug 4, 2026 •

edited

Loading