Implement pausing for submissions

[SemMulder]

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]

TODO: How to deal with reserved chunks when pausing?

[SemMulder]

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]

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

[SemMulder]

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

[SemMulder]

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]

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]

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

Loading

[SemMulder]

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]

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

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

[SemMulder]

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.

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

[ReinierMaas]

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

[ReinierMaas]

LGTM! Thanks for the update!

Can we easily replace paused: Bool with paused: (Paused|Started/Running) enum? If not due to the Pyo3 wrapping we can leave it as is.

[SemMulder]

@OpsBotPrime merge and tag

[OpsBotPrime]

Rebased as 4642e65, waiting for CI …

[OpsBotPrime]

CI job 🟡 started.

[OpsBotPrime]

@SemMulder I tagged your PR with v63. Please wait for the build of 4642e65 to pass and don't forget to deploy it!