Change concurrent update handling to be correct

Author: JelteF

Diff for: pgcommitfest/commitfest/views.py

@@ -830,18 +830,16 @@ def cfbot_ingest(message):
         # not it doesn't contain the newest patches that the CFBot knows about.
         return
 
-    # We don't simply fetch this from the patch.cfbot_branch, because we want
-    # to get it as select_for_update. That way we can use this row as a lock to
-    # ensure only one cfbot HTTP request can change a patch its Branch and
-    # Tasks at the same time. (see also transaction.atomic at the top of this
-    # function)
-    old_branch = CfbotBranch.objects.select_for_update().get(pk=patch_id)
-    if old_branch and old_branch.branch_id != branch_id and old_branch.created.replace(tzinfo=timezone.utc) > created:
-        # This is a message for an old branch, ignore it.
-        return
-
     # Every message should have a branch_status, which we will INSERT
     # or UPDATE.  We do this first, because cfbot_task refers to it.
+    # Due to the way messages are sent/queued by cfbot it's possible that it
+    # sends the messages out-of-order. To handle this we we only update in two
+    # cases:
+    # 1. The created time of the branch is newer than the one in our database:
+    #    This is a newer branch
+    # 2. If it's the same branch that we already have, but the modified time is
+    #    newer: This is a status update for the current branch that we received
+    #    in-order.
     cursor.execute("""INSERT INTO commitfest_cfbotbranch (patch_id, branch_id,
                                                 branch_name, commit_id,
                                                 apply_url, status,
@@ -855,7 +853,9 @@ def cfbot_ingest(message):
                             commit_id = EXCLUDED.commit_id,
                             apply_url = EXCLUDED.apply_url,
                             created = EXCLUDED.created
-                        WHERE commitfest_cfbotbranch.modified < EXCLUDED.modified
+                        WHERE commitfest_cfbotbranch.created < EXCLUDED.created
+                            OR (commitfest_cfbotbranch.branch_id = EXCLUDED.branch_id
+                                AND commitfest_cfbotbranch.modified < EXCLUDED.modified)
                         """,
                    (
                        patch_id,
@@ -868,6 +868,14 @@ def cfbot_ingest(message):
                        branch_status["modified"])
                    )
 
+    # Now we check what we have in our database. If that contains a different
+    # branch_id than what we just tried to insert, then apparently this is a
+    # status update for an old branch and we don't care about any of the
+    # contents of this message.
+    branch_in_db = CfbotBranch.objects.get(pk=patch_id)
+    if branch_in_db.branch_id != branch_id:
+        return
+
     # Most messages have a task_status.  It might be missing in rare cases, like
     # when cfbot decides that a whole branch has timed out.  We INSERT or
     # UPDATE.
@@ -892,6 +900,14 @@ def cfbot_ingest(message):
                            task_status["modified"])
                        )
 
+    # Remove any old tasks that are not related to this branch. These should
+    # only be left over when we just updated the branch_id. Knowing if we just
+    # updated the branch_id was is not trivial though, because INSERT ON
+    # CONFLICT does not allow us to easily return the old value of the row.
+    # So instead we always delete all tasks that are not related to this
+    # branch. This is fine, because doing so is very cheap in the no-op case
+    # because we have an index on patch_id and there's only a handful of tasks
+    # per patch.
     cursor.execute("DELETE FROM commitfest_cfbottask WHERE patch_id=%s AND branch_id != %s", (patch_id, branch_id))