Conversation
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Unresolved cleanup data-retention, delegated-reference, and transaction-notification issues block approval.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 2
Open (3)
What changed in this PR
Adds optional delegation support with status synchronization, event handling, database tracking, and cleanup integration.
Changes:
- Adds delegation routes and background reporting.
- Propagates submission status notifications.
- Adds external-task migrations and dependency updates.
- Refactors cleanup and extension APIs.
Review findings:
- Critical (4 votes): Cleanup can delete submissions still referenced by delegated tasks.
- Moderate (1 vote): Chunk cleanup leaves completed and failed chunk rows.
- Moderate (3 votes): Submission cleanup leaves terminal chunk payloads.
- Critical (2 votes): Delegation notifications may be broadcast before transactions commit.
| File | Summary |
|---|---|
workspace-hack/Cargo.toml |
Updates shared dependency features. |
opsqueue/src/server.rs |
Registers delegation routes and status channels. |
opsqueue/src/producer/server.rs |
Emits delegation status changes. |
opsqueue/src/lib.rs |
Exposes the delegation module. |
opsqueue/src/delegation/server.rs |
Implements delegation APIs and background reporting. |
opsqueue/src/delegation/mod.rs |
Registers the delegation server module. |
opsqueue/src/db/mod.rs |
Adds a test database-pool helper. |
opsqueue/src/consumer/server/mod.rs |
Emits status changes on chunk completion and failure. |
opsqueue/src/config.rs |
Adds delegation server URL configuration. |
opsqueue/src/common/submission.rs |
Adds status notifications and cleanup changes. |
opsqueue/src/common/mod.rs |
Registers extension support. |
opsqueue/src/common/extension.rs |
Adds extension and core APIs. |
opsqueue/src/common/chunk.rs |
Adds status notifications and chunk deletion helpers. |
opsqueue/migrations/20260811150000_submissions_external_task.up.sql |
Creates external-task storage. |
opsqueue/migrations/20260811150000_submissions_external_task.down.sql |
Removes external-task storage. |
opsqueue/Cargo.toml |
Adds delegation and test dependencies. |
opsqueue/app/main.rs |
Updates cleanup task initialization. |
Cargo.toml |
Bumps the workspace version. |
Cargo.lock |
Updates locked dependencies and package versions. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
a40eb46 to
7d44741
Compare
7d44741 to
fd4e096
Compare
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Unimplemented event handling, incorrect terminal-status recovery, and duplicate background workers can produce incorrect external state.
Get a fresh assessment by requesting another Copilot review.
Review effort: Balanced
Findings: 3
Open (5)
Retrying router construction spawns duplicate delegation workers · New Return events are accepted without updating submission state · New Missing submissions are incorrectly reported as successful · New Cancellation lacks a transaction-enforced type constraint · New Cleanup holds the writer connection across the entire scan · New
| self.core_api.clone(), | ||
| self.delegation_server_url.clone(), | ||
| ) | ||
| .run_background() |
| WorkerDelegationEvent::Return(_task_ids) => { | ||
| tracing::info!( | ||
| "Received 'return' delegation event, which is not yet implemented; ignoring." | ||
| ); |
| None => { | ||
| let task_id = &task.task_id; | ||
| let submission_id = &task.submission_id; | ||
| tracing::warn!(%submission_id, %task_id, "Got external task but could not find its submission. This could be a resubmitted external task, where we already cleaned up the corresponding submission after failure/completion. Assuming it completed successfully."); | ||
| DelegatedJobStatus::Completed |
| pub(crate) async fn cancel_submission_notx( | ||
| id: SubmissionId, | ||
| mut conn: impl WriterConnection<Transaction = True>, | ||
| mut conn: impl WriterConnection, |
| .execute(tx.get_inner()) | ||
| .await?.rows_affected(); | ||
| let mut read_conn = db.reader_conn().await?; | ||
| let mut write_conn = db.writer_conn().await?; |


No description provided.