Move "update related object after commit" to a separate worker in the task queue

Summary:
Depends on D20462. Ref T13276. Currently, the "Message" parser also updates related tasks and revisions when a commit is published.

For PHI1165, which ran into a race with message parsing, I originally believed we needed to separate this logic and lock + yield to avoid the race. D20462 provides what is probably a better approach for avoiding the race.

Still, I think separating these "update related revisions" and "updated related tasks" chunks into separate workers is a net improvement. There may still be some value in doing lock + yield in the future to deal with other issues, and when we occasionally run into problems with pulling a diff out of the repository to update the revision (usually because the diff is too big) this isolates the problem better and allows the commit to import.

I think the only thing to watch out for here is that Herald may now run before the revision and commit are attached to one another. This is fine for all current Herald rules, we just need to be mindful in implementing new rules.

Test Plan: Used `bin/repository reparse --message` on various commits, including commits that close revisions and close tasks.

Reviewers: amckinley

Reviewed By: amckinley

Maniphest Tasks: T13276

Differential Revision: https://secure.phabricator.com/D20463

src/__phutil_library_map__.php

@@ -1020,6 +1020,7 @@ phutil_register_library_map(array(
     'DiffusionURIEditor' => 'applications/diffusion/editor/DiffusionURIEditor.php',
     'DiffusionURITestCase' => 'applications/diffusion/request/__tests__/DiffusionURITestCase.php',
     'DiffusionUpdateCoverageConduitAPIMethod' => 'applications/diffusion/conduit/DiffusionUpdateCoverageConduitAPIMethod.php',
+    'DiffusionUpdateObjectAfterCommitWorker' => 'applications/diffusion/worker/DiffusionUpdateObjectAfterCommitWorker.php',
     'DiffusionView' => 'applications/diffusion/view/DiffusionView.php',
     'DivinerArticleAtomizer' => 'applications/diviner/atomizer/DivinerArticleAtomizer.php',
     'DivinerAtom' => 'applications/diviner/atom/DivinerAtom.php',
@@ -6685,6 +6686,7 @@ phutil_register_library_map(array(
     'DiffusionURIEditor' => 'PhabricatorApplicationTransactionEditor',
     'DiffusionURITestCase' => 'PhutilTestCase',
     'DiffusionUpdateCoverageConduitAPIMethod' => 'DiffusionConduitAPIMethod',
+    'DiffusionUpdateObjectAfterCommitWorker' => 'PhabricatorWorker',
     'DiffusionView' => 'AphrontView',
     'DivinerArticleAtomizer' => 'DivinerAtomizer',
     'DivinerAtom' => 'Phobject',

src/applications/differential/engine/DifferentialDiffExtractionEngine.php

@@ -315,6 +315,7 @@ final class DifferentialDiffExtractionEngine extends Phobject {
     $editor = id(new DifferentialTransactionEditor())
       ->setActor($viewer)
       ->setContinueOnMissingFields(true)
+      ->setContinueOnNoEffect(true)
       ->setContentSource($content_source)
       ->setChangedPriorToCommitURI($changed_uri)
       ->setIsCloseByCommit(true);
@@ -324,14 +325,7 @@ final class DifferentialDiffExtractionEngine extends Phobject {
       $editor->setActingAsPHID($author_phid);
     }
 
-    try {
-      $editor->applyTransactions($revision, $xactions);
-    } catch (PhabricatorApplicationTransactionNoEffectException $ex) {
-      // NOTE: We've marked transactions other than the CLOSE transaction
-      // as ignored when they don't have an effect, so this means that we
-      // lost a race to close the revision. That's perfectly fine, we can
-      // just continue normally.
-    }
+    $editor->applyTransactions($revision, $xactions);
   }
 
   private function loadConcerningBuilds(DifferentialRevision $revision) {

src/applications/diffusion/worker/DiffusionUpdateObjectAfterCommitWorker.php

@@ -0,0 +1,214 @@
+<?php
+
+final class DiffusionUpdateObjectAfterCommitWorker
+  extends PhabricatorWorker {
+
+  private $properties;
+
+  protected function getViewer() {
+    return PhabricatorUser::getOmnipotentUser();
+  }
+
+  protected function doWork() {
+    $viewer = $this->getViewer();
+    $data = $this->getTaskData();
+
+    $commit_phid = idx($data, 'commitPHID');
+    if (!$commit_phid) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht('No "commitPHID" in task data.'));
+    }
+
+    $commit = id(new DiffusionCommitQuery())
+      ->setViewer($viewer)
+      ->withPHIDs(array($commit_phid))
+      ->needIdentities(true)
+      ->executeOne();
+    if (!$commit) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht(
+          'Unable to load commit "%s".',
+          $commit_phid));
+    }
+
+    $object_phid = idx($data, 'objectPHID');
+    if (!$object_phid) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht('No "objectPHID" in task data.'));
+    }
+
+    $object = id(new PhabricatorObjectQuery())
+      ->setViewer($viewer)
+      ->withPHIDs(array($object_phid))
+      ->executeOne();
+    if (!$object) {
+      throw new PhabricatorWorkerPermanentFailureException(
+        pht(
+          'Unable to load object "%s".',
+          $object_phid));
+    }
+
+    $properties = idx($data, 'properties', array());
+    $this->properties = $properties;
+
+    if ($object instanceof ManiphestTask) {
+      $this->updateTask($commit, $object);
+    } else if ($object instanceof DifferentialRevision) {
+      $this->updateRevision($commit, $object);
+    }
+  }
+
+  protected function getUpdateProperty($key, $default = null) {
+    return idx($this->properties, $key, $default);
+  }
+
+  protected function getActingPHID(PhabricatorRepositoryCommit $commit) {
+    if ($commit->hasCommitterIdentity()) {
+      return $commit->getCommitterIdentity()->getIdentityDisplayPHID();
+    }
+
+    if ($commit->hasAuthorIdentity()) {
+      return $commit->getAuthorIdentity()->getIdentityDisplayPHID();
+    }
+
+    return id(new PhabricatorDiffusionApplication())->getPHID();
+  }
+
+  protected function loadActingUser($acting_phid) {
+    // If we we were able to identify an author or committer for the commit, we
+    // try to act as that user when affecting other objects, like tasks marked
+    // with "Fixes Txxx".
+
+    // This helps to prevent mistakes where a user accidentally writes the
+    // wrong task IDs and affects tasks they can't see (and thus can't undo the
+    // status changes for).
+
+    // This is just a guard rail, not a security measure. An attacker can still
+    // forge another user's identity trivially by forging author or committer
+    // email addresses.
+
+    // We also let commits with unrecognized authors act on any task to make
+    // behavior less confusing for new installs, and any user can craft a
+    // commit with an unrecognized author and committer.
+
+    $viewer = $this->getViewer();
+
+    $user_type = PhabricatorPeopleUserPHIDType::TYPECONST;
+    if (phid_get_type($acting_phid) === $user_type) {
+      $acting_user = id(new PhabricatorPeopleQuery())
+        ->setViewer($viewer)
+        ->withPHIDs(array($acting_phid))
+        ->executeOne();
+      if ($acting_user) {
+        return $acting_user;
+      }
+    }
+
+    return $viewer;
+  }
+
+  private function updateTask(
+    PhabricatorRepositoryCommit $commit,
+    ManiphestTask $task) {
+
+    $acting_phid = $this->getActingPHID($commit);
+    $acting_user = $this->loadActingUser($acting_phid);
+
+    $commit_phid = $commit->getPHID();
+
+    $xactions = array();
+
+    $xactions[] = $this->newEdgeTransaction(
+      $task,
+      $commit,
+      ManiphestTaskHasCommitEdgeType::EDGECONST);
+
+    $status = $this->getUpdateProperty('status');
+    if ($status) {
+      $xactions[] = $task->getApplicationTransactionTemplate()
+        ->setTransactionType(ManiphestTaskStatusTransaction::TRANSACTIONTYPE)
+        ->setMetadataValue('commitPHID', $commit_phid)
+        ->setNewValue($status);
+    }
+
+    $content_source = $this->newContentSource();
+
+    $editor = $task->getApplicationTransactionEditor()
+      ->setActor($acting_user)
+      ->setActingAsPHID($acting_phid)
+      ->setContentSource($content_source)
+      ->setContinueOnNoEffect(true)
+      ->setContinueOnMissingFields(true)
+      ->setUnmentionablePHIDMap(
+        array(
+          $commit_phid => $commit_phid,
+        ));
+
+    $editor->applyTransactions($task, $xactions);
+  }
+
+  private function updateRevision(
+    PhabricatorRepositoryCommit $commit,
+    DifferentialRevision $revision) {
+
+    // Reload the revision to get the active diff, which is currently required
+    // by "updateRevisionWithCommit()".
+    $revision = id(new DifferentialRevisionQuery())
+      ->setViewer($this->getViewer())
+      ->withIDs(array($revision->getID()))
+      ->needActiveDiffs(true)
+      ->executeOne();
+
+    $acting_phid = $this->getActingPHID($commit);
+    $acting_user = $this->loadActingUser($acting_phid);
+
+    $xactions = array();
+
+    $xactions[] = $this->newEdgeTransaction(
+      $revision,
+      $commit,
+      DiffusionCommitHasRevisionEdgeType::EDGECONST);
+
+    $match_data = $this->getUpdateProperty('revisionMatchData');
+
+    $type_close = DifferentialRevisionCloseTransaction::TRANSACTIONTYPE;
+    $xactions[] = $revision->getApplicationTransactionTemplate()
+      ->setTransactionType($type_close)
+      ->setNewValue(true)
+      ->setMetadataValue('isCommitClose', true)
+      ->setMetadataValue('revisionMatchData', $match_data)
+      ->setMetadataValue('commitPHID', $commit->getPHID());
+
+    $extraction_engine = id(new DifferentialDiffExtractionEngine())
+      ->setViewer($acting_user)
+      ->setAuthorPHID($acting_phid);
+
+    $content_source = $this->newContentSource();
+
+    $extraction_engine->updateRevisionWithCommit(
+      $revision,
+      $commit,
+      $xactions,
+      $content_source);
+  }
+
+  private function newEdgeTransaction(
+    $object,
+    PhabricatorRepositoryCommit $commit,
+    $edge_type) {
+
+    $commit_phid = $commit->getPHID();
+
+    return $object->getApplicationTransactionTemplate()
+      ->setTransactionType(PhabricatorTransactions::TYPE_EDGE)
+      ->setMetadataValue('edge:type', $edge_type)
+      ->setNewValue(
+        array(
+          '+' => array(
+            $commit_phid => $commit_phid,
+          ),
+        ));
+  }
+
+
+}

src/applications/repository/management/PhabricatorRepositoryManagementReparseWorkflow.php

@@ -311,19 +311,22 @@ final class PhabricatorRepositoryManagementReparseWorkflow
       );
 
       if ($all_from_repo && !$force_local) {
-        foreach ($classes as $class) {
-          PhabricatorWorker::scheduleTask(
-            $class,
-            $spec,
-            array(
-              'priority' => PhabricatorWorker::PRIORITY_IMPORT,
-            ));
-        }
+        $background = true;
       } else {
-        foreach ($classes as $class) {
-          $worker = newv($class, array($spec));
-          $worker->executeTask();
-        }
+        $background = false;
+      }
+
+      if (!$background) {
+        PhabricatorWorker::setRunAllTasksInProcess(true);
+      }
+
+      foreach ($classes as $class) {
+        PhabricatorWorker::scheduleTask(
+          $class,
+          $spec,
+          array(
+            'priority' => PhabricatorWorker::PRIORITY_IMPORT,
+          ));
       }
 
       $progress->update(1);

src/applications/repository/worker/commitmessageparser/PhabricatorRepositoryCommitMessageParserWorker.php

@@ -148,25 +148,6 @@ abstract class PhabricatorRepositoryCommitMessageParserWorker
         $author_phid);
     }
 
-    $differential_app = 'PhabricatorDifferentialApplication';
-    $revision_id = null;
-    $low_level_query = null;
-    if (PhabricatorApplication::isClassInstalled($differential_app)) {
-      $low_level_query = id(new DiffusionLowLevelCommitFieldsQuery())
-        ->setRepository($repository)
-        ->withCommitRef($ref);
-      $field_values = $low_level_query->execute();
-      $revision_id = idx($field_values, 'revisionID');
-
-      if (!empty($field_values['reviewedByPHIDs'])) {
-        $data->setCommitDetail(
-          'reviewerPHID',
-          reset($field_values['reviewedByPHIDs']));
-      }
-
-      $data->setCommitDetail('differential.revisionID', $revision_id);
-    }
-
     if ($author_phid != $commit->getAuthorPHID()) {
       $commit->setAuthorPHID($author_phid);
     }
@@ -191,92 +172,15 @@ abstract class PhabricatorRepositoryCommitMessageParserWorker
     $should_autoclose = $force_autoclose ||
                         $repository->shouldAutocloseCommit($commit);
 
-
     // When updating related objects, we'll act under an omnipotent user to
     // ensure we can see them, but take actions as either the committer or
     // author (if we recognize their accounts) or the Diffusion application
     // (if we do not).
 
-    $actor = PhabricatorUser::getOmnipotentUser();
-    $acting_as_phid = nonempty(
-      $committer_phid,
-      $author_phid,
-      id(new PhabricatorDiffusionApplication())->getPHID());
-
-    $acting_user = $this->loadActingUser($actor, $acting_as_phid);
-
-    $conn_w = id(new DifferentialRevision())->establishConnection('w');
-
-    // NOTE: The `differential_commit` table has a unique ID on `commitPHID`,
-    // preventing more than one revision from being associated with a commit.
-    // Generally this is good and desirable, but with the advent of hash
-    // tracking we may end up in a situation where we match several different
-    // revisions. We just kind of ignore this and pick one, we might want to
-    // revisit this and do something differently. (If we match several revisions
-    // someone probably did something very silly, though.)
-
-    $revision = null;
-    if ($revision_id) {
-      $revision_query = id(new DifferentialRevisionQuery())
-        ->withIDs(array($revision_id))
-        ->setViewer($actor)
-        ->needReviewers(true)
-        ->needActiveDiffs(true);
-
-      $revision = $revision_query->executeOne();
-
-      if ($revision) {
-        $commit_drev = DiffusionCommitHasRevisionEdgeType::EDGECONST;
-        id(new PhabricatorEdgeEditor())
-          ->addEdge($commit->getPHID(), $commit_drev, $revision->getPHID())
-          ->save();
-
-        $should_close = !$revision->isPublished() && $should_autoclose;
-        if ($should_close) {
-          $type_close = DifferentialRevisionCloseTransaction::TRANSACTIONTYPE;
-
-          $commit_close_xaction = id(new DifferentialTransaction())
-            ->setTransactionType($type_close)
-            ->setNewValue(true);
-
-          $commit_close_xaction->setMetadataValue(
-            'commitPHID',
-            $commit->getPHID());
-
-          if ($low_level_query) {
-            $commit_close_xaction->setMetadataValue(
-              'revisionMatchData',
-              $low_level_query->getRevisionMatchData());
-            $data->setCommitDetail(
-              'revisionMatchData',
-              $low_level_query->getRevisionMatchData());
-          }
-
-          $extraction_engine = id(new DifferentialDiffExtractionEngine())
-            ->setViewer($actor)
-            ->setAuthorPHID($acting_as_phid);
-
-          $content_source = $this->newContentSource();
-
-          $extraction_engine->updateRevisionWithCommit(
-            $revision,
-            $commit,
-            array(
-              $commit_close_xaction,
-            ),
-            $content_source);
-        }
-      }
-    }
-
     if ($should_autoclose) {
-      $this->closeTasks(
-        $actor,
-        $acting_as_phid,
-        $repository,
-        $commit,
-        $message,
-        $acting_user);
+      $actor = PhabricatorUser::getOmnipotentUser();
+      $this->closeRevisions($actor, $ref, $commit, $data);
+      $this->closeTasks($actor, $ref, $commit, $data);
     }
 
     $data->save();
@@ -295,27 +199,63 @@ abstract class PhabricatorRepositoryCommitMessageParserWorker
       ->execute();
   }
 
+  private function closeRevisions(
+    PhabricatorUser $actor,
+    DiffusionCommitRef $ref,
+    PhabricatorRepositoryCommit $commit,
+    PhabricatorRepositoryCommitData $data) {
+
+    $differential = 'PhabricatorDifferentialApplication';
+    if (!PhabricatorApplication::isClassInstalled($differential)) {
+      return;
+    }
+
+    $repository = $commit->getRepository();
+
+    $field_query = id(new DiffusionLowLevelCommitFieldsQuery())
+      ->setRepository($repository)
+      ->withCommitRef($ref);
+
+    $field_values = $field_query->execute();
+
+    $revision_id = idx($field_values, 'revisionID');
+    if (!$revision_id) {
+      return;
+    }
+
+    $revision = id(new DifferentialRevisionQuery())
+      ->setViewer($actor)
+      ->withIDs(array($revision_id))
+      ->executeOne();
+    if (!$revision) {
+      return;
+    }
+
+    // NOTE: This is very old code from when revisions had a single reviewer.
+    // It still powers the "Reviewer (Deprecated)" field in Herald, but should
+    // be removed.
+    if (!empty($field_values['reviewedByPHIDs'])) {
+      $data->setCommitDetail(
+        'reviewerPHID',
+        head($field_values['reviewedByPHIDs']));
+    }
+
+    $match_data = $field_query->getRevisionMatchData();
+
+    $data->setCommitDetail('differential.revisionID', $revision_id);
+    $data->setCommitDetail('revisionMatchData', $match_data);
+
+    $properties = array(
+      'revisionMatchData' => $match_data,
+    );
+    $this->queueObjectUpdate($commit, $revision, $properties);
+  }
+
   private function closeTasks(
     PhabricatorUser $actor,
-    $acting_as,
-    PhabricatorRepository $repository,
+    DiffusionCommitRef $ref,
     PhabricatorRepositoryCommit $commit,
-    $message,
-    PhabricatorUser $acting_user = null) {
-
-    // If we we were able to identify an author for the commit, we try to act
-    // as that user when loading tasks marked with "Fixes Txxx". This prevents
-    // mistakes where a user accidentally writes the wrong task IDs and affects
-    // tasks they can't see (and thus can't undo the status changes for).
-
-    // This is just a guard rail, not a security measure. An attacker can still
-    // forge another user's identity trivially by forging author or committer
-    // emails. We also let commits with unrecognized authors act on any task to
-    // make behavior less confusing for new installs.
-
-    if (!$acting_user) {
-      $acting_user = $actor;
-    }
+    PhabricatorRepositoryCommitData $data) {
 
     $maniphest = 'PhabricatorManiphestApplication';
     if (!PhabricatorApplication::isClassInstalled($maniphest)) {
@@ -324,11 +264,12 @@ abstract class PhabricatorRepositoryCommitMessageParserWorker
 
     $prefixes = ManiphestTaskStatus::getStatusPrefixMap();
     $suffixes = ManiphestTaskStatus::getStatusSuffixMap();
+    $message = $data->getCommitMessage();
 
     $matches = id(new ManiphestCustomFieldStatusParser())
       ->parseCorpus($message);
 
-    $task_statuses = array();
+    $task_map = array();
     foreach ($matches as $match) {
       $prefix = phutil_utf8_strtolower($match['prefix']);
       $suffix = phutil_utf8_strtolower($match['suffix']);
@@ -340,89 +281,44 @@ abstract class PhabricatorRepositoryCommitMessageParserWorker
 
       foreach ($match['monograms'] as $task_monogram) {
         $task_id = (int)trim($task_monogram, 'tT');
-        $task_statuses[$task_id] = $status;
+        $task_map[$task_id] = $status;
       }
     }
 
-    if (!$task_statuses) {
+    if (!$task_map) {
       return;
     }
 
     $tasks = id(new ManiphestTaskQuery())
-      ->setViewer($acting_user)
-      ->withIDs(array_keys($task_statuses))
-      ->needProjectPHIDs(true)
-      ->requireCapabilities(
-        array(
-          PhabricatorPolicyCapability::CAN_VIEW,
-          PhabricatorPolicyCapability::CAN_EDIT,
-        ))
+      ->setViewer($actor)
+      ->withIDs(array_keys($task_map))
       ->execute();
-
     foreach ($tasks as $task_id => $task) {
-      $xactions = array();
+      $status = $task_map[$task_id];
 
-      $edge_type = ManiphestTaskHasCommitEdgeType::EDGECONST;
-      $edge_xaction = id(new ManiphestTransaction())
-        ->setTransactionType(PhabricatorTransactions::TYPE_EDGE)
-        ->setMetadataValue('edge:type', $edge_type)
-        ->setNewValue(
-          array(
-            '+' => array(
-              $commit->getPHID() => $commit->getPHID(),
-            ),
-          ));
+      $properties = array(
+        'status' => $status,
+      );
 
-      $status = $task_statuses[$task_id];
-      if ($status) {
-        if ($task->getStatus() != $status) {
-          $xactions[] = id(new ManiphestTransaction())
-            ->setTransactionType(
-              ManiphestTaskStatusTransaction::TRANSACTIONTYPE)
-            ->setMetadataValue('commitPHID', $commit->getPHID())
-            ->setNewValue($status);
-
-          $edge_xaction->setMetadataValue('commitPHID', $commit->getPHID());
-        }
-      }
-
-      $xactions[] = $edge_xaction;
-
-      $content_source = $this->newContentSource();
-
-      $editor = id(new ManiphestTransactionEditor())
-        ->setActor($actor)
-        ->setActingAsPHID($acting_as)
-        ->setContinueOnNoEffect(true)
-        ->setContinueOnMissingFields(true)
-        ->setUnmentionablePHIDMap(
-          array($commit->getPHID() => $commit->getPHID()))
-        ->setContentSource($content_source);
-
-      $editor->applyTransactions($task, $xactions);
+      $this->queueObjectUpdate($commit, $task, $properties);
     }
   }
 
-  private function loadActingUser(PhabricatorUser $viewer, $user_phid) {
-    if (!$user_phid) {
-      return null;
-    }
+  private function queueObjectUpdate(
+    PhabricatorRepositoryCommit $commit,
+    $object,
+    array $properties) {
 
-    $user_type = PhabricatorPeopleUserPHIDType::TYPECONST;
-    if (phid_get_type($user_phid) != $user_type) {
-      return null;
-    }
-
-    $user = id(new PhabricatorPeopleQuery())
-      ->setViewer($viewer)
-      ->withPHIDs(array($user_phid))
-      ->executeOne();
-    if (!$user) {
-      return null;
-    }
-
-    return $user;
+    $this->queueTask(
+      'DiffusionUpdateObjectAfterCommitWorker',
+      array(
+        'commitPHID' => $commit->getPHID(),
+        'objectPHID' => $object->getPHID(),
+        'properties' => $properties,
+      ),
+      array(
+        'priority' => PhabricatorWorker::PRIORITY_DEFAULT,
+      ));
   }
 
-
 }