Repository navigation
Keep the first SUCCESS in the database backend - #10920
Conversation
The key-value backends ignore any update once a task has a SUCCESS. The database backend has no such check. With `acks_late`, a task runs again when its worker is killed after it saved the result but before the ack. The second run then writes over the stored SUCCESS. With `task_track_started`, readers see STARTED, and then a new result or a FAILURE. Return early when the stored state is SUCCESS, as the key-value backends do. Any later write for that task id is ignored, including a FAILURE that the same run stores when `on_success` raises. Also write the status column every time. SQLAlchemy only writes columns that changed since it read the row. So a STARTED write could race with another run's SUCCESS and leave that SUCCESS next to the STARTED result, which the new check would then keep. Fixes celery#10912
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #10920 +/- ##
==========================================
- Coverage 90.72% 90.71% -0.02%
==========================================
Files 153 153
Lines 21520 21529 +9
Branches 2649 2652 +3
==========================================
+ Hits 19524 19530 +6
- Misses 1684 1685 +1
- Partials 312 314 +2
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. |
MehrazRumman
left a comment
MehrazRumman
left a comment
There was a problem hiding this comment.
Checked at 52120ef. t/unit/backends/test_database.py: 111 passed. All five new tests fail on main.
- The early return matches
BaseKeyValueStoreBackend._store_result, which skips any write once the stored status is SUCCESS. The two backends now behave the same for a task that runs again after a crash, which is the case in #10912. flag_modified(task, 'status')is needed, andtest_store_result_racing_a_success_writes_its_stateshows why: without it, main leaves the row with SUCCESS next to the STARTED result, and the new SUCCESS check would then freeze it like that. Good catch.- The return happens before
_update_result, so the extended columns (worker,retries, ...) keep the values from the run that succeeded.test_store_result_after_success_keeps_extended_fieldscovers that.
On the row lock
On @auvipy's side of the open question in the description: the lock is small. I tried it locally:
def _query_task(self, session, task_id, for_update=False):
"""Query task by id, falling back to deferring missing columns from database."""
try:
query = session.query(self.task_cls).filter(self.task_cls.task_id == task_id)
if for_update:
query = query.with_for_update()
tasks = list(query)
return tasks and tasks[0]_store_result then calls self._query_task(session, task_id, for_update=True). The reads (_get_task_meta_for) stay as they are.
- SQLAlchemy emits
... FOR UPDATEfor PostgreSQL and MySQL and drops it for SQLite, so SQLite behaves as it does now. - The lock is held until
session.commit(), or untilsession_cleanupcloses the session on the early return. - With it, the database tests pass except
test_store_result_racing_a_success_writes_its_state. That fails only because its_query_taskstub takes two arguments. With the lock, that interleaving can't happen on PostgreSQL or MySQL anyway: the second run waits for the SUCCESS commit and then hits the new check. So the test would need to change with it.
I agree it can be a follow-up. The sequential case is the one #10912 reproduces, and this PR fixes it. I'd still lean towards adding the lock here, since #10912 reports 7 of 150 results changing in the Postgres fault test, and that test exercises the concurrent case too.
Smaller point
The MongoDB and Cassandra backends have the same gap, as the description says. It would be good to open an issue for them so they aren't forgotten.
i agree with this @ChristianMurphy |
When a task ran twice at once, both runs could read the row before either one wrote SUCCESS. The later write then replaced the first SUCCESS. Read the row again with SELECT ... FOR UPDATE before the check. The second run then waits for the first and sees its SUCCESS. Only a row that exists is locked. On MySQL, locking a missing row takes a gap lock, and concurrent first writes for new tasks then deadlock. SQLite has no row locks, and the status column is still always written for it.
|
Thanks. I added the lock in f4054df, with one change: I measured it with 16 processes storing STARTED and then SUCCESS for new tasks (3,200 writes), and with two runs storing SUCCESS for the same task at once (150 rounds):
Postgres had no errors in any version. The MongoDB and Cassandra gap is now #10929. |
Note: Before submitting this pull request, please review our contributing
guidelines.
Description
Fixes #10912.
The key-value backends ignore any write once a task has a SUCCESS. The database backend had no such check. So when a task runs again under
acks_late, because its worker was killed after saving the result and before the ack, the second run writes over the stored SUCCESS.This returns early in
DatabaseBackend._store_resultwhen the stored state is SUCCESS.It also always writes the
statuscolumn. SQLAlchemy only writes columns whose values changed since it read the row. If another run saved a SUCCESS after that read, a STARTED write used to update the result and skip the status. The row then held SUCCESS with the STARTED result. The new check would then ignore every later write to that row, including that run's own SUCCESS.When the row exists, it's read again with
SELECT ... FOR UPDATEbefore the check. When a task runs twice at once, the second run waits for the first and then sees its SUCCESS. A missing row isn't locked. On MySQL, that takes a gap lock, and concurrent first writes for new tasks then deadlock. With 16 processes writing new tasks, locking every read gave 8 to 20 deadlocks per 3,200 writes on MySQL 8.4, and this version gave none. SQLite has no row locks, and the status column is still always written for it.Other effects, all as on the key-value backends:
on_successraises, andAbortableAsyncResult.abort()on a task that already succeeded.AsyncResult.forget()or the dailycelery.backend_cleanuptask deletes the row. That task only runs with Celery beat. The FAQ says reusing an id is undefined.worker_deduplicate_successful_taskscan skip a copy that runs again, but it is off by default. The MongoDB and Cassandra backends have the same gap, in #10929.The new tests check that STARTED, FAILURE and SUCCESS writes after a SUCCESS are ignored, and that the extended columns stay the same. They also check that a row can still go from STARTED to SUCCESS. When another run saves SUCCESS during a write, that SUCCESS is kept, and without row locks the row isn't left mixed. The crash script in #10912 now keeps the first SUCCESS.