Update major RefCursor callsites to work properly with RefPosition

Summary:
Ref T11823. This is the meaty part of the change, and updates `RefEngine` to use separate RefCursor (for names) and RefPosition (for actual commit positions) tables.

I'll hold this whole series until after the release cut so it has some time to bake on `secure` to look for issues. It's also not a huge problem if there are bugs here since these tables are just caches anyway, although they do feed into some other things, and obviously it's never good to have bugs.

Test Plan:
  - This logic can be invoked directly with `bin/repository refs <repository> --trace --verbose`.
  - Ran that on unchanged repositories, new branches, removed branches, and modified branches. Saw appropriate output and cursor positions.
  - Ran on a mercurial repository to test the close/open logic, saw it correct open/closed state of incorrect positions.
  - Browed around Diffusion in various repositories.

Reviewers: amckinley

Reviewed By: amckinley

Maniphest Tasks: T11823

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

src/applications/differential/controller/DifferentialRevisionOperationController.php

@@ -137,19 +137,9 @@ final class DifferentialRevisionOperationController
       return null;
     }
 
-    // NOTE: See PHI68. This is a workaround to make "Land Revision" work
-    // until T11823 is fixed properly. If we find multiple refs with the same
-    // name (normally, duplicate "master" refs), just pick the first one.
-
-    $refs = $this->newRefQuery($repository)
+    return $this->newRefQuery($repository)
       ->withRefNames(array($default_name))
-      ->execute();
-
-    if ($refs) {
-      return head($refs);
-    }
-
-    return null;
+      ->executeOne();
   }
 
   private function getDefaultRefName(

src/applications/diffusion/query/DiffusionCachedResolveRefsQuery.php

@@ -106,9 +106,11 @@ final class DiffusionCachedResolveRefsQuery
 
     $cursors = queryfx_all(
       $conn_r,
-      'SELECT refNameHash, refType, commitIdentifier, isClosed FROM %T
-        WHERE repositoryPHID = %s AND refNameHash IN (%Ls)',
+      'SELECT c.refNameHash, c.refType, p.commitIdentifier, p.isClosed
+        FROM %T c JOIN %T p ON p.cursorID = c.id
+        WHERE c.repositoryPHID = %s AND c.refNameHash IN (%Ls)',
       id(new PhabricatorRepositoryRefCursor())->getTableName(),
+      id(new PhabricatorRepositoryRefPosition())->getTableName(),
       $repository->getPHID(),
       array_keys($name_hashes));
 

src/applications/repository/engine/PhabricatorRepositoryRefEngine.php

@@ -7,17 +7,18 @@
 final class PhabricatorRepositoryRefEngine
   extends PhabricatorRepositoryEngine {
 
-  private $newRefs = array();
-  private $deadRefs = array();
+  private $newPositions = array();
+  private $deadPositions = array();
   private $closeCommits = array();
   private $hasNoCursors;
 
   public function updateRefs() {
-    $this->newRefs = array();
-    $this->deadRefs = array();
+    $this->newPositions = array();
+    $this->deadPositions = array();
     $this->closeCommits = array();
 
     $repository = $this->getRepository();
+    $viewer = $this->getViewer();
 
     $branches_may_close = false;
 
@@ -53,8 +54,9 @@ final class PhabricatorRepositoryRefEngine
     );
 
     $all_cursors = id(new PhabricatorRepositoryRefCursorQuery())
-      ->setViewer(PhabricatorUser::getOmnipotentUser())
+      ->setViewer($viewer)
       ->withRepositoryPHIDs(array($repository->getPHID()))
+      ->needPositions(true)
       ->execute();
     $cursor_groups = mgroup($all_cursors, 'getRefType');
 
@@ -63,8 +65,15 @@ final class PhabricatorRepositoryRefEngine
     // Find all the heads of closing refs.
     $all_closing_heads = array();
     foreach ($all_cursors as $cursor) {
-      if ($this->shouldCloseRef($cursor->getRefType(), $cursor->getRefName())) {
-        $all_closing_heads[] = $cursor->getCommitIdentifier();
+      $should_close = $this->shouldCloseRef(
+        $cursor->getRefType(),
+        $cursor->getRefName());
+      if (!$should_close) {
+        continue;
+      }
+
+      foreach ($cursor->getPositionIdentifiers() as $identifier) {
+        $all_closing_heads[] = $identifier;
       }
     }
     $all_closing_heads = array_unique($all_closing_heads);
@@ -79,25 +88,13 @@ final class PhabricatorRepositoryRefEngine
       $this->setCloseFlagOnCommits($this->closeCommits);
     }
 
-    if ($this->newRefs || $this->deadRefs) {
+    if ($this->newPositions || $this->deadPositions) {
       $repository->openTransaction();
-        foreach ($this->newRefs as $ref) {
-          $ref->save();
-        }
-        foreach ($this->deadRefs as $ref) {
-          // Shove this ref into the old refs table so the discovery engine
-          // can check if any commits have been rendered unreachable.
-          id(new PhabricatorRepositoryOldRef())
-            ->setRepositoryPHID($repository->getPHID())
-            ->setCommitIdentifier($ref->getCommitIdentifier())
-            ->save();
 
-          $ref->delete();
-        }
+        $this->saveNewPositions();
+        $this->deleteDeadPositions();
+
       $repository->saveTransaction();
-
-      $this->newRefs = array();
-      $this->deadRefs = array();
     }
 
     $branches = $maps[PhabricatorRepositoryRefCursor::TYPE_BRANCH];
@@ -111,10 +108,12 @@ final class PhabricatorRepositoryRefEngine
     array $branches) {
 
     assert_instances_of($branches, 'DiffusionRepositoryRef');
+    $viewer = $this->getViewer();
 
     $all_cursors = id(new PhabricatorRepositoryRefCursorQuery())
-      ->setViewer(PhabricatorUser::getOmnipotentUser())
+      ->setViewer($viewer)
       ->withRepositoryPHIDs(array($repository->getPHID()))
+      ->needPositions(true)
       ->execute();
 
     $state_map = array();
@@ -124,36 +123,57 @@ final class PhabricatorRepositoryRefEngine
         continue;
       }
       $raw_name = $cursor->getRefNameRaw();
-      $hash = $cursor->getCommitIdentifier();
 
-      $state_map[$raw_name][$hash] = $cursor;
+      foreach ($cursor->getPositions() as $position) {
+        $hash = $position->getCommitIdentifier();
+        $state_map[$raw_name][$hash] = $position;
+      }
     }
 
+    $updates = array();
     foreach ($branches as $branch) {
-      $cursor = idx($state_map, $branch->getShortName(), array());
-      $cursor = idx($cursor, $branch->getCommitIdentifier());
-      if (!$cursor) {
+      $position = idx($state_map, $branch->getShortName(), array());
+      $position = idx($position, $branch->getCommitIdentifier());
+      if (!$position) {
         continue;
       }
 
       $fields = $branch->getRawFields();
 
-      $cursor_state = (bool)$cursor->getIsClosed();
+      $position_state = (bool)$position->getIsClosed();
       $branch_state = (bool)idx($fields, 'closed');
 
-      if ($cursor_state != $branch_state) {
-        $cursor->setIsClosed((int)$branch_state)->save();
-      }
+      if ($position_state != $branch_state) {
+        $updates[$position->getID()] = (int)$branch_state;
       }
     }
 
-  private function markRefNew(PhabricatorRepositoryRefCursor $cursor) {
-    $this->newRefs[] = $cursor;
+    if ($updates) {
+      $position_table = id(new PhabricatorRepositoryRefPosition());
+      $conn = $position_table->establishConnection('w');
+
+      $position_table->openTransaction();
+        foreach ($updates as $position_id => $branch_state) {
+          queryfx(
+            $conn,
+            'UPDATE %T SET isClosed = %d WHERE id = %d',
+            $position_table->getTableName(),
+            $branch_state,
+            $position_id);
+        }
+      $position_table->saveTransaction();
+    }
+  }
+
+  private function markPositionNew(
+    PhabricatorRepositoryRefPosition $position) {
+    $this->newPositions[] = $position;
     return $this;
   }
 
-  private function markRefDead(PhabricatorRepositoryRefCursor $cursor) {
-    $this->deadRefs[] = $cursor;
+  private function markPositionDead(
+    PhabricatorRepositoryRefPosition $position) {
+    $this->deadPositions[] = $position;
     return $this;
   }
 
@@ -203,10 +223,7 @@ final class PhabricatorRepositoryRefEngine
     // NOTE: Mercurial branches may have multiple branch heads; this logic
     // is complex primarily to account for that.
 
-    // Group all the cursors by their ref name, like "master". Since Mercurial
-    // branches may have multiple heads, there could be several cursors with
-    // the same name.
-    $cursor_groups = mgroup($cursors, 'getRefNameRaw');
+    $cursors = mpull($cursors, null, 'getRefNameRaw');
 
     // Group all the new ref values by their name. As above, these groups may
     // have multiple members in Mercurial.
@@ -215,38 +232,47 @@ final class PhabricatorRepositoryRefEngine
     foreach ($ref_groups as $name => $refs) {
       $new_commits = mpull($refs, 'getCommitIdentifier', 'getCommitIdentifier');
 
-      $ref_cursors = idx($cursor_groups, $name, array());
-      $old_commits = mpull($ref_cursors, null, 'getCommitIdentifier');
+      $ref_cursor = idx($cursors, $name);
+      if ($ref_cursor) {
+        $old_positions = $ref_cursor->getPositions();
+      } else {
+        $old_positions = array();
+      }
 
       // We're going to delete all the cursors pointing at commits which are
       // no longer associated with the refs. This primarily makes the Mercurial
       // multiple head case easier, and means that when we update a ref we
       // delete the old one and write a new one.
-      foreach ($ref_cursors as $cursor) {
-        if (isset($new_commits[$cursor->getCommitIdentifier()])) {
+      foreach ($old_positions as $old_position) {
+        $hash = $old_position->getCommitIdentifier();
+        if (isset($new_commits[$hash])) {
           // This ref previously pointed at this commit, and still does.
           $this->log(
             pht(
               'Ref %s "%s" still points at %s.',
               $ref_type,
               $name,
-              $cursor->getCommitIdentifier()));
-        } else {
+              $hash));
+          continue;
+        }
+
         // This ref previously pointed at this commit, but no longer does.
         $this->log(
           pht(
             'Ref %s "%s" no longer points at %s.',
             $ref_type,
             $name,
-              $cursor->getCommitIdentifier()));
+            $hash));
 
         // Nuke the obsolete cursor.
-          $this->markRefDead($cursor);
-        }
+        $this->markPositionDead($old_position);
       }
 
       // Now, we're going to insert new cursors for all the commits which are
       // associated with this ref that don't currently have cursors.
+      $old_commits = mpull($old_positions, 'getCommitIdentifier');
+      $old_commits = array_fuse($old_commits);
+
       $added_commits = array_diff_key($new_commits, $old_commits);
       foreach ($added_commits as $identifier) {
         $this->log(
@@ -255,12 +281,24 @@ final class PhabricatorRepositoryRefEngine
             $ref_type,
             $name,
             $identifier));
-        $this->markRefNew(
-          id(new PhabricatorRepositoryRefCursor())
-            ->setRepositoryPHID($repository->getPHID())
-            ->setRefType($ref_type)
-            ->setRefName($name)
-            ->setCommitIdentifier($identifier));
+
+        if (!$ref_cursor) {
+          // If this is the first time we've seen a particular ref (for
+          // example, a new branch) we need to insert a RefCursor record
+          // for it before we can insert a RefPosition.
+
+          $ref_cursor = $this->newRefCursor(
+            $repository,
+            $ref_type,
+            $name);
+        }
+
+        $new_position = id(new PhabricatorRepositoryRefPosition())
+          ->setCursorID($ref_cursor->getID())
+          ->setCommitIdentifier($identifier)
+          ->setIsClosed(0);
+
+        $this->markPositionNew($new_position);
       }
 
       if ($this->shouldCloseRef($ref_type, $name)) {
@@ -277,16 +315,21 @@ final class PhabricatorRepositoryRefEngine
     // Find any cursors for refs which no longer exist. This happens when a
     // branch, tag or bookmark is deleted.
 
-    foreach ($cursor_groups as $name => $cursor_group) {
-      if (idx($ref_groups, $name) === null) {
-        foreach ($cursor_group as $cursor) {
+    foreach ($cursors as $name => $cursor) {
+      if (!empty($ref_groups[$name])) {
+        // This ref still has some positions, so we don't need to wipe it
+        // out. Try the next one.
+        continue;
+      }
+
+      foreach ($cursor->getPositions() as $position) {
         $this->log(
           pht(
             'Ref %s "%s" no longer exists.',
             $cursor->getRefType(),
             $cursor->getRefName()));
-          $this->markRefDead($cursor);
-        }
+
+        $this->markPositionDead($position);
       }
     }
   }
@@ -452,6 +495,81 @@ final class PhabricatorRepositoryRefEngine
     return $this;
   }
 
+  private function newRefCursor(
+    PhabricatorRepository $repository,
+    $ref_type,
+    $ref_name) {
+
+    $cursor = id(new PhabricatorRepositoryRefCursor())
+      ->setRepositoryPHID($repository->getPHID())
+      ->setRefType($ref_type)
+      ->setRefName($ref_name);
+
+    try {
+      return $cursor->save();
+    } catch (AphrontDuplicateKeyQueryException $ex) {
+      // If we raced another daemon to create this position and lost the race,
+      // load the cursor the other daemon created instead.
+    }
+
+    $viewer = $this->getViewer();
+
+    $cursor = id(new PhabricatorRepositoryRefCursorQuery())
+      ->setViewer($viewer)
+      ->withRepositoryPHIDs(array($repository->getPHID()))
+      ->withRefTypes(array($ref_type))
+      ->withRefNames(array($ref_name))
+      ->needPositions(true)
+      ->executeOne();
+    if (!$cursor) {
+      throw new Exception(
+        pht(
+          'Failed to create a new ref cursor (for "%s", of type "%s", in '.
+          'repository "%s") because it collided with an existing cursor, '.
+          'but then failed to load that cursor.',
+          $ref_name,
+          $ref_type,
+          $repository->getDisplayName()));
+    }
+
+    return $cursor;
+  }
+
+  private function saveNewPositions() {
+    $positions = $this->newPositions;
+
+    foreach ($positions as $position) {
+      try {
+        $position->save();
+      } catch (AphrontDuplicateKeyQueryException $ex) {
+        // We may race another daemon to create this position. If we do, and
+        // we lose the race, that's fine: the other daemon did our work for
+        // us and we can continue.
+      }
+    }
+
+    $this->newPositions = array();
+  }
+
+  private function deleteDeadPositions() {
+    $positions = $this->deadPositions;
+    $repository = $this->getRepository();
+
+    foreach ($positions as $position) {
+      // Shove this ref into the old refs table so the discovery engine
+      // can check if any commits have been rendered unreachable.
+      id(new PhabricatorRepositoryOldRef())
+        ->setRepositoryPHID($repository->getPHID())
+        ->setCommitIdentifier($position->getCommitIdentifier())
+        ->save();
+
+      $position->delete();
+    }
+
+    $this->deadPositions = array();
+  }
+
+
 
 /* -(  Updating Git Refs  )-------------------------------------------------- */
 

src/applications/repository/query/PhabricatorRepositoryRefCursorQuery.php

@@ -9,6 +9,7 @@ final class PhabricatorRepositoryRefCursorQuery
   private $refTypes;
   private $refNames;
   private $datasourceQuery;
+  private $needPositions;
 
   public function withIDs(array $ids) {
     $this->ids = $ids;
@@ -40,6 +41,11 @@ final class PhabricatorRepositoryRefCursorQuery
     return $this;
   }
 
+  public function needPositions($need) {
+    $this->needPositions = $need;
+    return $this;
+  }
+
   public function newResultObject() {
     return new PhabricatorRepositoryRefCursor();
   }
@@ -68,6 +74,22 @@ final class PhabricatorRepositoryRefCursorQuery
       $ref->attachRepository($repository);
     }
 
+    if (!$refs) {
+      return $refs;
+    }
+
+    if ($this->needPositions) {
+      $positions = id(new PhabricatorRepositoryRefPosition())->loadAllWhere(
+        'cursorID IN (%Ld)',
+        mpull($refs, 'getID'));
+      $positions = mgroup($positions, 'getCursorID');
+
+      foreach ($refs as $key => $ref) {
+        $ref_positions = idx($positions, $ref->getID(), array());
+        $ref->attachPositions($ref_positions);
+      }
+    }
+
     return $refs;
   }
 

src/applications/repository/storage/PhabricatorRepositoryRefCursor.php

@@ -21,6 +21,7 @@ final class PhabricatorRepositoryRefCursor
   protected $refNameEncoding;
 
   private $repository = self::ATTACHABLE;
+  private $positions = self::ATTACHABLE;
 
   protected function getConfiguration() {
     return array(
@@ -71,6 +72,20 @@ final class PhabricatorRepositoryRefCursor
     return $this->assertAttached($this->repository);
   }
 
+  public function attachPositions(array $positions) {
+    assert_instances_of($positions, 'PhabricatorRepositoryRefPosition');
+    $this->positions = $positions;
+    return $this;
+  }
+
+  public function getPositions() {
+    return $this->assertAttached($this->positions);
+  }
+
+  public function getPositionIdentifiers() {
+    return mpull($this->getPositions(), 'getCommitIdentifier');
+  }
+
 
 /* -(  PhabricatorPolicyInterface  )----------------------------------------- */