Implement pausing for submissions#132
Conversation
a9f5556 to
bcd4161
Compare
|
TODO: How to deal with reserved chunks when pausing? |
b09f93b to
7e7052d
Compare
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:
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 ;). |
Introduce `submissions_paused` and `chunks_paused` tables
(alongside the existing `submissions_{completed,failed,cancelled}` and
`chunks_{completed,failed}` tables).
Pausing a submission atomically moves it from `submissions` →
`submissions_paused` and its remaining chunks from `chunks` →
`chunks_paused`. Because paused chunks are no longer in the `chunks`
table, the consumer dispatcher naturally skips them without any changes
to the dispatch query.
Unpausing reverses the move and notifies waiting consumers.
Paused submissions can also be inserted directly in the paused state
(via the new `paused: bool` field on `InsertSubmission`); in that
case `notify_on_insert` is intentionally not fired.
Paused submissions are cancellable; `cancel_submission` now handles
the case where the submission is found in `submissions_paused`.
…ly completed, failed, or paused chunks
990a5d2 to
71ace4e
Compare
| pass | ||
|
|
||
|
|
||
| class ChunkNotFoundError(IncorrectUsageError): |
There was a problem hiding this comment.
Note that ChunkNotFoundError can be dropped since it is (and was) dead code.
| ) | ||
| .await | ||
| }) | ||
| .await; |
There was a problem hiding this comment.
Note that we were silently dropping errors here before, which went unnoticed because of the underscore prefix of _chunk_size.
| }) | ||
| .await; | ||
|
|
||
| counter!(crate::prometheus::SUBMISSIONS_TOTAL_COUNTER).increment(1); |
There was a problem hiding this comment.
TODO: Increment SUBMISSIONS_PAUSED_COUNTER.
| #[sqlx::test] | ||
| /// Test that a submission inserted directly in the paused state stays paused | ||
| /// (unlike empty non-paused submissions which are auto-completed). | ||
| pub async fn insert_paused_submission_stays_paused(db: sqlx::SqlitePool) { | ||
| let db = WriterPool::new(db); | ||
| let mut conn = db.writer_conn().await.unwrap(); | ||
| insert_submission_from_chunks( | ||
| None, | ||
| vec![Some("foo".into()), Some("bar".into())], | ||
| None, | ||
| StrategicMetadataMap::default(), | ||
| ChunkSize::default(), | ||
| true, | ||
| &mut conn, | ||
| ) | ||
| .await | ||
| .expect("insertion failed"); | ||
|
|
||
| assert_matches!(count_submissions(&mut conn).await, Ok(0)); | ||
| assert_matches!(count_submissions_paused(&mut conn).await, Ok(1)); | ||
| assert_matches!(count_submissions_completed(&mut conn).await, Ok(0)); | ||
| } |
There was a problem hiding this comment.
TODO: Drop this test, it doesn't add much compared to the tests above.
1 participant
Introduce
submissions_pausedandchunks_pausedtables (alongside the existingsubmissions_{completed,failed,cancelled}andchunks_{completed,failed}tables).Pausing a submission atomically moves it from
submissions→submissions_pausedand its remaining chunks fromchunks→chunks_paused. Because paused chunks are no longer in thechunkstable, the consumer dispatcher naturally skips them without any changes to the dispatch query.Unpausing reverses the move and notifies waiting consumers.
Paused submissions can also be inserted directly in the paused state (via the new
paused: boolfield onInsertSubmission); in that casenotify_on_insertis intentionally not fired.Paused submissions are cancellable;
cancel_submissionnow handles the case where the submission is found insubmissions_paused.