Use write connection and transactions in SQL patches

Summary:
Patches often read from slaves (possibly stale data) and use that information to write on master.
It causes problems when applying more patches quickly after each other because data created in previous patch may not be replicated yet.

Test Plan:
  $ bin/storage upgrade

Reviewers: epriestley

Reviewed By: epriestley

CC: aran, Korvin

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

resources/sql/patches/079.nametokenindex.php

@@ -1,14 +1,18 @@
 <?php
 
-$conn = $schema_conn;
-
 echo "Indexing username tokens for typeaheads...\n";
 
-$users = id(new PhabricatorUser())->loadAll();
+$table = new PhabricatorUser();
+$table->openTransaction();
+$table->beginReadLocking();
+
+$users = $table->loadAll();
 echo count($users)." users to index";
 foreach ($users as $user) {
   $user->updateNameTokens();
   echo ".";
 }
 
+$table->endReadLocking();
+$table->saveTransaction();
 echo "\nDone.\n";

resources/sql/patches/081.filekeys.php

@@ -1,15 +1,22 @@
 <?php
 
 echo "Generating file keys...\n";
-$files = id(new PhabricatorFile())->loadAllWhere('secretKey IS NULL');
+$table = new PhabricatorFile();
+$table->openTransaction();
+$table->beginReadLocking();
+
+$files = $table->loadAllWhere('secretKey IS NULL');
 echo count($files).' files to generate keys for';
 foreach ($files as $file) {
   queryfx(
-    $file->establishConnection('r'),
+    $file->establishConnection('w'),
     'UPDATE %T SET secretKey = %s WHERE id = %d',
     $file->getTableName(),
     $file->generateSecretKey(),
     $file->getID());
   echo '.';
 }
+
+$table->endReadLocking();
+$table->saveTransaction();
 echo "\nDone.\n";

resources/sql/patches/090.forceuniqueprojectnames.php

@@ -1,7 +1,11 @@
 <?php
 
 echo "Ensuring project names are unique enough...\n";
-$projects = id(new PhabricatorProject())->loadAll();
+$table = new PhabricatorProject();
+$table->openTransaction();
+$table->beginReadLocking();
+
+$projects = $table->loadAll();
 
 $slug_map = array();
 
@@ -66,6 +70,8 @@ while ($update) {
   }
 }
 
+$table->endReadLocking();
+$table->saveTransaction();
 echo "Done.\n";
 
 

resources/sql/patches/093.gitremotes.php

@@ -3,11 +3,12 @@
 echo "Stripping remotes from repository default branches...\n";
 
 $table = new PhabricatorRepository();
+$table->openTransaction();
 $conn_w = $table->establishConnection('w');
 
 $repos = queryfx_all(
   $conn_w,
-  'SELECT id, name, details FROM %T WHERE versionControlSystem = %s',
+  'SELECT id, name, details FROM %T WHERE versionControlSystem = %s FOR UPDATE',
   $table->getTableName(),
   'git');
 
@@ -39,4 +40,5 @@ foreach ($repos as $repo) {
     $id);
 }
 
+$table->saveTransaction();
 echo "Done.\n";

resources/sql/patches/098.heraldruletypemigration.php

@@ -1,8 +1,11 @@
 <?php
 
 echo "Checking for rules that can be converted to 'personal'. ";
+$table = new HeraldRule();
+$table->openTransaction();
+$table->beginReadLocking();
 
-$rules = id(new HeraldRule())->loadAll();
+$rules = $table->loadAll();
 
 foreach ($rules as $rule) {
   if ($rule->getRuleType() !== HeraldRuleTypeConfig::RULE_TYPE_PERSONAL) {
@@ -43,4 +46,6 @@ foreach ($rules as $rule) {
   }
 }
 
-echo "Done. ";
+$table->endReadLocking();
+$table->saveTransaction();
+echo "Done.\n";

resources/sql/patches/102.heraldcleanup.php

@@ -1,8 +1,11 @@
 <?php
 
 echo "Cleaning up old Herald rule applied rows...\n";
+$table = new HeraldRule();
+$table->openTransaction();
+$table->beginReadLocking();
 
-$rules = id(new HeraldRule())->loadAll();
+$rules = $table->loadAll();
 foreach ($rules as $key => $rule) {
   $first_policy = HeraldRepetitionPolicyConfig::toInt(
     HeraldRepetitionPolicyConfig::FIRST);
@@ -11,7 +14,7 @@ foreach ($rules as $key => $rule) {
   }
 }
 
-$conn_w = id(new HeraldRule())->establishConnection('w');
+$conn_w = $table->establishConnection('w');
 
 $clause = '';
 if ($rules) {
@@ -31,5 +34,6 @@ do {
   echo ".";
 } while ($conn_w->getAffectedRows());
 
-echo "\n";
+$table->endReadLocking();
-echo "Done.\n";
+$table->saveTransaction();
+echo "\nDone.\n";

resources/sql/patches/111.commitauditmigration.php

@@ -1,15 +1,17 @@
 <?php
 
 echo "Updating old commit authors...\n";
-
 $table = new PhabricatorRepositoryCommit();
+$table->openTransaction();
+
 $conn = $table->establishConnection('w');
 $data = new PhabricatorRepositoryCommitData();
 $commits = queryfx_all(
   $conn,
   'SELECT c.id id, c.authorPHID authorPHID, d.commitDetails details
     FROM %T c JOIN %T d ON d.commitID = c.id
-    WHERE c.authorPHID IS NULL',
+    WHERE c.authorPHID IS NULL
+    FOR UPDATE',
   $table->getTableName(),
   $data->getTableName());
 
@@ -28,16 +30,16 @@ foreach ($commits as $commit) {
   }
 }
 
+$table->saveTransaction();
 echo "Done.\n";
 
 
 echo "Updating old commit mailKeys...\n";
+$table->openTransaction();
 
-$table = new PhabricatorRepositoryCommit();
-$conn = $table->establishConnection('w');
 $commits = queryfx_all(
   $conn,
-  'SELECT id FROM %T WHERE mailKey = %s',
+  'SELECT id FROM %T WHERE mailKey = %s FOR UPDATE',
   $table->getTableName(),
   '');
 
@@ -52,4 +54,5 @@ foreach ($commits as $commit) {
   echo "#{$id}\n";
 }
 
+$table->saveTransaction();
 echo "Done.\n";

resources/sql/patches/131.migraterevisionquery.php

@@ -1,12 +1,13 @@
 <?php
 
 $table = new DifferentialRevision();
+$table->openTransaction();
+$table->beginReadLocking();
 $conn_w = $table->establishConnection('w');
 
 echo "Migrating revisions";
 do {
-  $revisions = id(new DifferentialRevision())
+  $revisions = $table->loadAllWhere('branchName IS NULL LIMIT 1000');
-    ->loadAllWhere('branchName IS NULL LIMIT 1000');
 
   foreach ($revisions as $revision) {
     echo ".";
@@ -28,4 +29,7 @@ do {
       $revision->getID());
   }
 } while (count($revisions) == 1000);
+
+$table->endReadLocking();
+$table->saveTransaction();
 echo "\nDone.\n";

resources/sql/patches/20121209.xmacromigrate.php

@@ -1,7 +1,10 @@
 <?php
 
 echo "Giving image macros PHIDs";
-foreach (new LiskMigrationIterator(new PhabricatorFileImageMacro()) as $macro) {
+$table = new PhabricatorFileImageMacro();
+$table->openTransaction();
+
+foreach (new LiskMigrationIterator($table) as $macro) {
   if ($macro->getPHID()) {
     continue;
   }
@@ -9,10 +12,12 @@ foreach (new LiskMigrationIterator(new PhabricatorFileImageMacro()) as $macro) {
   echo ".";
 
   queryfx(
-    $macro->establishConnection('r'),
+    $macro->establishConnection('w'),
     'UPDATE %T SET phid = %s WHERE id = %d',
     $macro->getTableName(),
     $macro->generatePHID(),
     $macro->getID());
 }
+
+$table->saveTransaction();
 echo "\nDone.\n";

resources/sql/patches/emailtableport.php

@@ -3,16 +3,16 @@
 echo "Migrating user emails...\n";
 
 $table  = new PhabricatorUser();
-$conn   = $table->establishConnection('r');
+$table->openTransaction();
+$conn   = $table->establishConnection('w');
 
 $emails = queryfx_all(
   $conn,
-  'SELECT phid, email FROM %T',
+  'SELECT phid, email FROM %T LOCK IN SHARE MODE',
   $table->getTableName());
 $emails = ipull($emails, 'email', 'phid');
 
 $etable = new PhabricatorUserEmail();
-$econn  = $etable->establishConnection('w');
 
 foreach ($emails as $phid => $email) {
 
@@ -21,7 +21,7 @@ foreach ($emails as $phid => $email) {
 
   echo "Migrating '{$phid}'...\n";
   queryfx(
-    $econn,
+    $conn,
     'INSERT INTO %T (userPHID, address, verificationCode, isVerified, isPrimary)
       VALUES (%s, %s, %s, 1, 1)',
     $etable->getTableName(),
@@ -30,4 +30,5 @@ foreach ($emails as $phid => $email) {
     Filesystem::readRandomCharacters(24));
 }
 
+$table->saveTransaction();
 echo "Done.\n";

resources/sql/patches/migrate-differential-dependencies.php

@@ -1,7 +1,10 @@
 <?php
 
 echo "Migrating differential dependencies to edges...\n";
-foreach (new LiskMigrationIterator(new DifferentialRevision()) as $rev) {
+$table = new DifferentialRevision();
+$table->openTransaction();
+
+foreach (new LiskMigrationIterator($table) as $rev) {
   $id = $rev->getID();
   echo "Revision {$id}: ";
 
@@ -23,4 +26,5 @@ foreach (new LiskMigrationIterator(new DifferentialRevision()) as $rev) {
   echo "OKAY\n";
 }
 
+$table->saveTransaction();
 echo "Done.\n";

resources/sql/patches/migrate-maniphest-dependencies.php

@@ -1,7 +1,10 @@
 <?php
 
 echo "Migrating task dependencies to edges...\n";
-foreach (new LiskMigrationIterator(new ManiphestTask()) as $task) {
+$table = new ManiphestTask();
+$table->openTransaction();
+
+foreach (new LiskMigrationIterator($table) as $task) {
   $id = $task->getID();
   echo "Task {$id}: ";
 
@@ -23,4 +26,5 @@ foreach (new LiskMigrationIterator(new ManiphestTask()) as $task) {
   echo "OKAY\n";
 }
 
+$table->saveTransaction();
 echo "Done.\n";

resources/sql/patches/migrate-maniphest-revisions.php

@@ -1,7 +1,10 @@
 <?php
 
 echo "Migrating task revisions to edges...\n";
-foreach (new LiskMigrationIterator(new ManiphestTask()) as $task) {
+$table = new ManiphestTask();
+$table->establishConnection('w');
+
+foreach (new LiskMigrationIterator($table) as $task) {
   $id = $task->getID();
   echo "Task {$id}: ";
 

resources/sql/patches/migrate-project-edges.php

@@ -1,12 +1,15 @@
 <?php
 
 echo "Migrating project members to edges...\n";
-foreach (new LiskMigrationIterator(new PhabricatorProject()) as $proj) {
+$table = new PhabricatorProject();
+$table->establishConnection('w');
+
+foreach (new LiskMigrationIterator($table) as $proj) {
   $id = $proj->getID();
   echo "Project {$id}: ";
 
   $members = queryfx_all(
-    $proj->establishConnection('r'),
+    $proj->establishConnection('w'),
     'SELECT userPHID FROM %T WHERE projectPHID = %s',
     'project_affiliation',
     $proj->getPHID());

resources/sql/patches/ponder-mailkey-populate.php

@@ -1,7 +1,10 @@
 <?php
 
 echo "Populating Questions with mail keys...\n";
-foreach (new LiskMigrationIterator(new PonderQuestion()) as $question) {
+$table = new PonderQuestion();
+$table->openTransaction();
+
+foreach (new LiskMigrationIterator($table) as $question) {
   $id = $question->getID();
 
   echo "Question {$id}: ";
@@ -18,4 +21,5 @@ foreach (new LiskMigrationIterator(new PonderQuestion()) as $question) {
   }
 }
 
+$table->saveTransaction();
 echo "Done.\n";