Add job delegation - #176
Draft
vincevannoort wants to merge 11 commits into
Draft
Conversation
u63 forces us to wrap literals in `u63::new`, and we need to convert to u64 at actual usage sites anyway.
…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.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Adds paused submissions, Python timeout support, and job delegation with status reporting.
Changes:
- Adds paused submission persistence, APIs, metrics, and Python bindings.
- Adds delegation endpoints and background status synchronization.
- Adds blocking-operation timeouts and removes obsolete chunk errors.
Reviewed changes
Copilot reviewed 30 out of 32 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
workspace-hack/Cargo.toml |
Updates generated dependency features. |
opsqueue/src/server/interface.rs |
Adds integration-facing submission controls. |
opsqueue/src/server.rs |
Wires delegation into the router. |
opsqueue/src/prometheus.rs |
Adds pause/unpause metrics. |
opsqueue/src/producer/server.rs |
Supports paused insertion and unpausing. |
opsqueue/src/producer/common.rs |
Adds the paused request field. |
opsqueue/src/producer/client.rs |
Adds the Rust unpause client. |
opsqueue/src/lib.rs |
Exposes delegation. |
opsqueue/src/delegation/server.rs |
Implements delegation endpoints and reporting. |
opsqueue/src/delegation/mod.rs |
Defines the delegation module. |
opsqueue/src/db/mod.rs |
Adds split test-pool construction. |
opsqueue/src/consumer/strategy.rs |
Updates submission test calls. |
opsqueue/src/consumer/server/mod.rs |
Emits terminal status notifications. |
opsqueue/src/consumer/client.rs |
Updates submission test calls. |
opsqueue/src/config.rs |
Adds delegation server configuration. |
opsqueue/src/common/submission.rs |
Implements paused submission persistence and status. |
opsqueue/src/common/errors.rs |
Removes the obsolete chunk error. |
opsqueue/src/common/chunk.rs |
Adds paused chunk transitions and idempotency. |
opsqueue/migrations/20260811150000_submissions_external_task.up.sql |
Creates external task mappings. |
opsqueue/migrations/20260811150000_submissions_external_task.down.sql |
Reverts task mappings. |
opsqueue/migrations/20260805143000_pausing.up.sql |
Creates paused submission tables. |
opsqueue/migrations/20260805143000_pausing.down.sql |
Reverts paused tables. |
opsqueue/Cargo.toml |
Adds delegation test dependencies. |
libs/opsqueue_python/tests/test_roundtrip.py |
Tests pausing and timeouts. |
libs/opsqueue_python/src/producer.rs |
Adds FFI unpause and timeout support. |
libs/opsqueue_python/src/lib.rs |
Registers paused submissions. |
libs/opsqueue_python/src/errors.rs |
Maps timeout errors to Python. |
libs/opsqueue_python/src/common.rs |
Adds the Python paused status type. |
libs/opsqueue_python/python/opsqueue/producer.py |
Exposes pausing, unpausing, and timeouts. |
libs/opsqueue_python/python/opsqueue/exceptions.py |
Removes ChunkNotFoundError. |
Cargo.lock |
Locks added dependencies. |
Suppressed comments (1)
opsqueue/src/delegation/server.rs:394
- Only the current batch's completions were sent, but this deletes all terminal external tasks. If sending a later batch fails, its mappings have already been removed and those completions are permanently lost. Delete only terminal tasks from the current batch.
out_of_date_tasks
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| tracing::error!("DB error acquiring writer connection: {e:?}"); | ||
| StatusCode::INTERNAL_SERVER_ERROR | ||
| })?; | ||
| insert_external_task(&mut conn, job.payload.submission_id, &job.task_id) |
| ) -> Result<StatusCode, StatusCode> { | ||
| tracing::info!("Received 'return' delegation event, which is not yet implemented; ignoring."); | ||
|
|
||
| Ok(StatusCode::ACCEPTED) |
| let conn = state.interface.pool.writer_conn().await?; | ||
| update_last_status_sent( | ||
| conn, | ||
| out_of_date_tasks |
Comment on lines
+1490
to
+1491
| // TODO(delegation): Prevent deletion if it is still referenced in | ||
| // `submissions_external_task`. |
Comment on lines
+238
to
+242
| .retry(retry_policy()) | ||
| .when(|e| match e { | ||
| L(_) => false, | ||
| R(client_err) => client_err.is_ephemeral(), | ||
| }) |
Comment on lines
+121
to
+123
| for id in ids { | ||
| statuses.push(submission::db::submission_status(id, &mut conn).await?); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
This pull request adds support for pausing and unpausing submissions in the opsqueue Python client, introduces timeout handling for blocking submission operations, and removes the unused
ChunkNotFoundError. It also updates the FFI layer to expose the new paused status and error handling to Python. The changes enhance the flexibility and robustness of submission management in the client.Closes: https://github.com/channable/infra-requests/issues/56#issuecomment-5453471722