diff --git a/scripts/sql/manage_storage.php b/scripts/sql/manage_storage.php
index 521e26938..db21fc23f 100755
--- a/scripts/sql/manage_storage.php
+++ b/scripts/sql/manage_storage.php
@@ -1,213 +1,246 @@
 #!/usr/bin/env php
 <?php
 
 $root = dirname(dirname(dirname(__FILE__)));
 require_once $root.'/scripts/init/init-setup.php';
 
 $args = new PhutilArgumentParser($argv);
 $args->setTagline(pht('manage Phabricator storage and schemata'));
 $args->setSynopsis(<<<EOHELP
 **storage** __workflow__ [__options__]
 Manage Phabricator database storage and schema versioning.
 
 **storage** upgrade
 Initialize or upgrade Phabricator storage.
 
 **storage** upgrade --user __root__ --password __hunter2__
 Use administrative credentials for schema changes.
 EOHELP
 );
 $args->parseStandardArguments();
 
 $default_namespace  = PhabricatorLiskDAO::getDefaultStorageNamespace();
 
 try {
   $args->parsePartial(
     array(
       array(
         'name'    => 'force',
         'short'   => 'f',
         'help'    => pht(
           'Do not prompt before performing dangerous operations.'),
       ),
       array(
         'name' => 'host',
         'param' => 'hostname',
         'help' => pht(
-          'Connect to __host__ instead of the default host.'),
+          'Operate on the database server identified by __hostname__.'),
+      ),
+      array(
+        'name' => 'ref',
+        'param' => 'ref',
+        'help' => pht(
+          'Operate on the database identified by __ref__.'),
       ),
       array(
         'name'    => 'user',
         'short'   => 'u',
         'param'   => 'username',
         'help'    => pht(
           'Connect with __username__ instead of the configured default.'),
       ),
       array(
         'name'    => 'password',
         'short'   => 'p',
         'param'   => 'password',
         'help'    => pht('Use __password__ instead of the configured default.'),
       ),
       array(
         'name'    => 'namespace',
         'param'   => 'name',
         'default' => $default_namespace,
         'help'    => pht(
           "Use namespace __namespace__ instead of the configured ".
           "default ('%s'). This is an advanced feature used by unit tests; ".
           "you should not normally use this flag.",
           $default_namespace),
       ),
       array(
         'name'    => 'dryrun',
         'help'    => pht(
           'Do not actually change anything, just show what would be changed.'),
       ),
       array(
         'name'    => 'disable-utf8mb4',
         'help'    => pht(
           'Disable %s, even if the database supports it. This is an '.
           'advanced feature used for testing changes to Phabricator; you '.
           'should not normally use this flag.',
           'utf8mb4'),
       ),
     ));
 } catch (PhutilArgumentUsageException $ex) {
   $args->printUsageException($ex);
   exit(77);
 }
 
 // First, test that the Phabricator configuration is set up correctly. After
 // we know this works we'll test any administrative credentials specifically.
 
-$host = $args->getArg('host');
-if (strlen($host)) {
-  $ref = null;
-
-  $refs = PhabricatorDatabaseRef::getLiveRefs();
+$refs = PhabricatorDatabaseRef::getActiveDatabaseRefs();
+if (!$refs) {
+  throw new PhutilArgumentUsageException(
+    pht('No databases are configured.'));
+}
 
-  // Include the master in case the user is just specifying a redundant
-  // "--host" flag for no reason and does not actually have a database
-  // cluster configured.
-  foreach (PhabricatorDatabaseRef::getMasterDatabaseRefs() as $master_ref) {
-    $refs[] = $master_ref;
+$host = $args->getArg('host');
+$ref_key = $args->getArg('ref');
+if (strlen($host) || strlen($ref_key)) {
+  if ($host && $ref_key) {
+    throw new PhutilArgumentUsageException(
+      pht(
+        'Use "--host" or "--ref" to select a database, but not both.'));
   }
 
+  $refs = PhabricatorDatabaseRef::getActiveDatabaseRefs();
+
+  $possible_refs = array();
   foreach ($refs as $possible_ref) {
-    if ($possible_ref->getHost() == $host) {
-      $ref = $possible_ref;
+    if ($host && ($possible_ref->getHost() == $host)) {
+      $possible_refs[] = $possible_ref;
+      break;
+    }
+    if ($ref_key && ($possible_ref->getRefKey() == $ref_key)) {
+      $possible_refs[] = $possible_ref;
       break;
     }
   }
 
-  if (!$ref) {
+  if (!$possible_refs) {
+    if ($host) {
+      throw new PhutilArgumentUsageException(
+        pht(
+          'There is no configured database on host "%s". This command can '.
+          'only interact with configured databases.',
+          $host));
+    } else {
+      throw new PhutilArgumentUsageException(
+        pht(
+          'There is no configured database with ref "%s". This command can '.
+          'only interact with configured databases.',
+          $ref_key));
+    }
+  }
+
+  if (count($possible_refs) > 1) {
     throw new PhutilArgumentUsageException(
       pht(
-        'There is no configured database on host "%s". This command can '.
-        'only interact with configured databases.',
+        'Host "%s" identifies more than one database. Use "--ref" to select '.
+        'a specific database.',
         $host));
   }
-} else {
-  $ref = PhabricatorDatabaseRef::getMasterDatabaseRef();
-  if (!$ref) {
-    throw new Exception(
-      pht('No database master is configured.'));
-  }
-}
-
-$default_user = $ref->getUser();
-$default_host = $ref->getHost();
-$default_port = $ref->getPort();
 
-$test_api = id(new PhabricatorStorageManagementAPI())
-  ->setUser($default_user)
-  ->setHost($default_host)
-  ->setPort($default_port)
-  ->setPassword($ref->getPass())
-  ->setNamespace($args->getArg('namespace'));
-
-try {
-  queryfx(
-    $test_api->getConn(null),
-    'SELECT 1');
-} catch (AphrontQueryException $ex) {
-  $message = phutil_console_format(
-    "**%s**\n\n%s\n\n%s\n\n%s\n\n**%s**: %s\n",
-    pht('MySQL Credentials Not Configured'),
-    pht(
-      'Unable to connect to MySQL using the configured credentials. '.
-      'You must configure standard credentials before you can upgrade '.
-      'storage. Run these commands to set up credentials:'),
-    "  phabricator/ $ ./bin/config set mysql.host __host__\n".
-    "  phabricator/ $ ./bin/config set mysql.user __username__\n".
-    "  phabricator/ $ ./bin/config set mysql.pass __password__",
-    pht(
-      'These standard credentials are separate from any administrative '.
-      'credentials provided to this command with __%s__ or '.
-      '__%s__, and must be configured correctly before you can proceed.',
-      '--user',
-      '--password'),
-    pht('Raw MySQL Error'),
-    $ex->getMessage());
-  echo phutil_console_wrap($message);
-  exit(1);
+  $refs = $possible_refs;
 }
 
-if ($args->getArg('password') === null) {
-  // This is already a PhutilOpaqueEnvelope.
-  $password = $ref->getPass();
-} else {
-  // Put this in a PhutilOpaqueEnvelope.
-  $password = new PhutilOpaqueEnvelope($args->getArg('password'));
-  PhabricatorEnv::overrideConfig('mysql.pass', $args->getArg('password'));
-}
+$apis = array();
+foreach ($refs as $ref) {
+  $default_user = $ref->getUser();
+  $default_host = $ref->getHost();
+  $default_port = $ref->getPort();
+
+  $test_api = id(new PhabricatorStorageManagementAPI())
+    ->setUser($default_user)
+    ->setHost($default_host)
+    ->setPort($default_port)
+    ->setPassword($ref->getPass())
+    ->setNamespace($args->getArg('namespace'));
+
+  try {
+    queryfx(
+      $test_api->getConn(null),
+      'SELECT 1');
+  } catch (AphrontQueryException $ex) {
+    $message = phutil_console_format(
+      "**%s**\n\n%s\n\n%s\n\n%s\n\n**%s**: %s\n",
+      pht('MySQL Credentials Not Configured'),
+      pht(
+        'Unable to connect to MySQL using the configured credentials. '.
+        'You must configure standard credentials before you can upgrade '.
+        'storage. Run these commands to set up credentials:'),
+      "  phabricator/ $ ./bin/config set mysql.host __host__\n".
+      "  phabricator/ $ ./bin/config set mysql.user __username__\n".
+      "  phabricator/ $ ./bin/config set mysql.pass __password__",
+      pht(
+        'These standard credentials are separate from any administrative '.
+        'credentials provided to this command with __%s__ or '.
+        '__%s__, and must be configured correctly before you can proceed.',
+        '--user',
+        '--password'),
+      pht('Raw MySQL Error'),
+      $ex->getMessage());
+    echo phutil_console_wrap($message);
+    exit(1);
+  }
 
-$selected_user = $args->getArg('user');
-if ($selected_user === null) {
-  $selected_user = $default_user;
-}
+  if ($args->getArg('password') === null) {
+    // This is already a PhutilOpaqueEnvelope.
+    $password = $ref->getPass();
+  } else {
+    // Put this in a PhutilOpaqueEnvelope.
+    $password = new PhutilOpaqueEnvelope($args->getArg('password'));
+    PhabricatorEnv::overrideConfig('mysql.pass', $args->getArg('password'));
+  }
 
-$api = id(new PhabricatorStorageManagementAPI())
-  ->setUser($selected_user)
-  ->setHost($default_host)
-  ->setPort($default_port)
-  ->setPassword($password)
-  ->setNamespace($args->getArg('namespace'))
-  ->setDisableUTF8MB4($args->getArg('disable-utf8mb4'));
-PhabricatorEnv::overrideConfig('mysql.user', $api->getUser());
+  $selected_user = $args->getArg('user');
+  if ($selected_user === null) {
+    $selected_user = $default_user;
+  }
 
-try {
-  queryfx(
-    $api->getConn(null),
-    'SELECT 1');
-} catch (AphrontQueryException $ex) {
-  $message = phutil_console_format(
-    "**%s**\n\n%s\n\n**%s**: %s\n",
-    pht('Bad Administrative Credentials'),
-    pht(
-      'Unable to connect to MySQL using the administrative credentials '.
-      'provided with the __%s__ and __%s__ flags. Check that '.
-      'you have entered them correctly.',
-      '--user',
-      '--password'),
-    pht('Raw MySQL Error'),
-    $ex->getMessage());
-  echo phutil_console_wrap($message);
-  exit(1);
+  $api = id(new PhabricatorStorageManagementAPI())
+    ->setUser($selected_user)
+    ->setHost($default_host)
+    ->setPort($default_port)
+    ->setPassword($password)
+    ->setNamespace($args->getArg('namespace'))
+    ->setDisableUTF8MB4($args->getArg('disable-utf8mb4'));
+  PhabricatorEnv::overrideConfig('mysql.user', $api->getUser());
+
+  try {
+    queryfx(
+      $api->getConn(null),
+      'SELECT 1');
+  } catch (AphrontQueryException $ex) {
+    $message = phutil_console_format(
+      "**%s**\n\n%s\n\n**%s**: %s\n",
+      pht('Bad Administrative Credentials'),
+      pht(
+        'Unable to connect to MySQL using the administrative credentials '.
+        'provided with the __%s__ and __%s__ flags. Check that '.
+        'you have entered them correctly.',
+        '--user',
+        '--password'),
+      pht('Raw MySQL Error'),
+      $ex->getMessage());
+    echo phutil_console_wrap($message);
+    exit(1);
+  }
+
+  $api->setRef($ref);
+  $apis[] = $api;
 }
 
 $workflows = id(new PhutilClassMapQuery())
   ->setAncestorClass('PhabricatorStorageManagementWorkflow')
   ->execute();
 
 $patches = PhabricatorSQLPatchList::buildAllPatches();
 
 foreach ($workflows as $workflow) {
-  $workflow->setAPI($api);
+  $workflow->setAPIs($apis);
   $workflow->setPatches($patches);
 }
 
 $workflows[] = new PhutilHelpArgumentWorkflow();
 
 $args->parseWorkflows($workflows);
diff --git a/src/applications/config/schema/PhabricatorConfigSchemaQuery.php b/src/applications/config/schema/PhabricatorConfigSchemaQuery.php
index 0ef5e484b..2b5fc79f6 100644
--- a/src/applications/config/schema/PhabricatorConfigSchemaQuery.php
+++ b/src/applications/config/schema/PhabricatorConfigSchemaQuery.php
@@ -1,347 +1,377 @@
 <?php
 
 final class PhabricatorConfigSchemaQuery extends Phobject {
 
+  private $refs;
+  private $apis;
+
+  public function setRefs(array $refs) {
+    $this->refs = $refs;
+    return $this;
+  }
+
+  public function getRefs() {
+    if (!$this->refs) {
+      return PhabricatorDatabaseRef::getMasterDatabaseRefs();
+    }
+    return $this->refs;
+  }
+
+  public function setAPIs(array $apis) {
+    $map = array();
+    foreach ($apis as $api) {
+      $map[$api->getRef()->getRefKey()] = $api;
+    }
+    $this->apis = $map;
+    return $this;
+  }
+
   private function getDatabaseNames(PhabricatorDatabaseRef $ref) {
     $api = $this->getAPI($ref);
     $patches = PhabricatorSQLPatchList::buildAllPatches();
     return $api->getDatabaseList(
       $patches,
       $only_living = true);
   }
 
   private function getAPI(PhabricatorDatabaseRef $ref) {
+    $key = $ref->getRefKey();
+
+    if (isset($this->apis[$key])) {
+      return $this->apis[$key];
+    }
+
     return id(new PhabricatorStorageManagementAPI())
       ->setUser($ref->getUser())
       ->setHost($ref->getHost())
       ->setPort($ref->getPort())
       ->setNamespace(PhabricatorLiskDAO::getDefaultStorageNamespace())
       ->setPassword($ref->getPass());
   }
 
   public function loadActualSchemata() {
-    $refs = PhabricatorDatabaseRef::getMasterDatabaseRefs();
+    $refs = $this->getRefs();
 
     $schemata = array();
     foreach ($refs as $ref) {
       $schema = $this->loadActualSchemaForServer($ref);
       $schemata[$schema->getRef()->getRefKey()] = $schema;
     }
 
     return $schemata;
   }
 
   private function loadActualSchemaForServer(PhabricatorDatabaseRef $ref) {
     $databases = $this->getDatabaseNames($ref);
 
     $conn = $ref->newManagementConnection();
 
     $tables = queryfx_all(
       $conn,
       'SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COLLATION
         FROM INFORMATION_SCHEMA.TABLES
         WHERE TABLE_SCHEMA IN (%Ls)',
       $databases);
 
     $database_info = queryfx_all(
       $conn,
       'SELECT SCHEMA_NAME, DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME
         FROM INFORMATION_SCHEMA.SCHEMATA
         WHERE SCHEMA_NAME IN (%Ls)',
       $databases);
     $database_info = ipull($database_info, null, 'SCHEMA_NAME');
 
     // Find databases which exist, but which the user does not have permission
     // to see.
     $invisible_databases = array();
     foreach ($databases as $database_name) {
       if (isset($database_info[$database_name])) {
         continue;
       }
 
       try {
         queryfx($conn, 'SHOW TABLES IN %T', $database_name);
       } catch (AphrontAccessDeniedQueryException $ex) {
         // This database exists, the user just doesn't have permission to
         // see it.
         $invisible_databases[] = $database_name;
       } catch (AphrontSchemaQueryException $ex) {
         // This database is legitimately missing.
       }
     }
 
     $sql = array();
     foreach ($tables as $table) {
       $sql[] = qsprintf(
         $conn,
         '(TABLE_SCHEMA = %s AND TABLE_NAME = %s)',
         $table['TABLE_SCHEMA'],
         $table['TABLE_NAME']);
     }
 
     if ($sql) {
       $column_info = queryfx_all(
         $conn,
         'SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, CHARACTER_SET_NAME,
             COLLATION_NAME, COLUMN_TYPE, IS_NULLABLE, EXTRA
           FROM INFORMATION_SCHEMA.COLUMNS
           WHERE (%Q)',
         '('.implode(') OR (', $sql).')');
       $column_info = igroup($column_info, 'TABLE_SCHEMA');
     } else {
       $column_info = array();
     }
 
     // NOTE: Tables like KEY_COLUMN_USAGE and TABLE_CONSTRAINTS only contain
     // primary, unique, and foreign keys, so we can't use them here. We pull
     // indexes later on using SHOW INDEXES.
 
     $server_schema = id(new PhabricatorConfigServerSchema())
       ->setRef($ref);
 
     $tables = igroup($tables, 'TABLE_SCHEMA');
     foreach ($tables as $database_name => $database_tables) {
       $info = $database_info[$database_name];
 
       $database_schema = id(new PhabricatorConfigDatabaseSchema())
         ->setName($database_name)
         ->setCharacterSet($info['DEFAULT_CHARACTER_SET_NAME'])
         ->setCollation($info['DEFAULT_COLLATION_NAME']);
 
       $database_column_info = idx($column_info, $database_name, array());
       $database_column_info = igroup($database_column_info, 'TABLE_NAME');
 
       foreach ($database_tables as $table) {
         $table_name = $table['TABLE_NAME'];
 
         $table_schema = id(new PhabricatorConfigTableSchema())
           ->setName($table_name)
           ->setCollation($table['TABLE_COLLATION']);
 
         $columns = idx($database_column_info, $table_name, array());
         foreach ($columns as $column) {
           if (strpos($column['EXTRA'], 'auto_increment') === false) {
             $auto_increment = false;
           } else {
             $auto_increment = true;
           }
 
           $column_schema = id(new PhabricatorConfigColumnSchema())
             ->setName($column['COLUMN_NAME'])
             ->setCharacterSet($column['CHARACTER_SET_NAME'])
             ->setCollation($column['COLLATION_NAME'])
             ->setColumnType($column['COLUMN_TYPE'])
             ->setNullable($column['IS_NULLABLE'] == 'YES')
             ->setAutoIncrement($auto_increment);
 
           $table_schema->addColumn($column_schema);
         }
 
         $key_parts = queryfx_all(
           $conn,
           'SHOW INDEXES FROM %T.%T',
           $database_name,
           $table_name);
         $keys = igroup($key_parts, 'Key_name');
         foreach ($keys as $key_name => $key_pieces) {
           $key_pieces = isort($key_pieces, 'Seq_in_index');
           $head = head($key_pieces);
 
           // This handles string indexes which index only a prefix of a field.
           $column_names = array();
           foreach ($key_pieces as $piece) {
             $name = $piece['Column_name'];
             if ($piece['Sub_part']) {
               $name = $name.'('.$piece['Sub_part'].')';
             }
             $column_names[] = $name;
           }
 
           $key_schema = id(new PhabricatorConfigKeySchema())
             ->setName($key_name)
             ->setColumnNames($column_names)
             ->setUnique(!$head['Non_unique'])
             ->setIndexType($head['Index_type']);
 
           $table_schema->addKey($key_schema);
         }
 
         $database_schema->addTable($table_schema);
       }
 
       $server_schema->addDatabase($database_schema);
     }
 
     foreach ($invisible_databases as $database_name) {
       $server_schema->addDatabase(
         id(new PhabricatorConfigDatabaseSchema())
           ->setName($database_name)
           ->setAccessDenied(true));
     }
 
     return $server_schema;
   }
 
   public function loadExpectedSchemata() {
-    $refs = PhabricatorDatabaseRef::getMasterDatabaseRefs();
+    $refs = $this->getRefs();
 
     $schemata = array();
     foreach ($refs as $ref) {
       $schema = $this->loadExpectedSchemaForServer($ref);
       $schemata[$schema->getRef()->getRefKey()] = $schema;
     }
 
     return $schemata;
   }
 
   public function loadExpectedSchemaForServer(PhabricatorDatabaseRef $ref) {
     $databases = $this->getDatabaseNames($ref);
     $info = $this->getAPI($ref)->getCharsetInfo();
 
     $specs = id(new PhutilClassMapQuery())
       ->setAncestorClass('PhabricatorConfigSchemaSpec')
       ->execute();
 
     $server_schema = id(new PhabricatorConfigServerSchema())
       ->setRef($ref);
 
     foreach ($specs as $spec) {
       $spec
         ->setUTF8Charset(
           $info[PhabricatorStorageManagementAPI::CHARSET_DEFAULT])
         ->setUTF8BinaryCollation(
           $info[PhabricatorStorageManagementAPI::COLLATE_TEXT])
         ->setUTF8SortingCollation(
           $info[PhabricatorStorageManagementAPI::COLLATE_SORT])
         ->setServer($server_schema)
         ->buildSchemata($server_schema);
     }
 
     return $server_schema;
   }
 
   public function buildComparisonSchemata(
     array $expect_servers,
     array $actual_servers) {
 
     $schemata = array();
     foreach ($actual_servers as $key => $actual_server) {
       $schemata[$key] = $this->buildComparisonSchemaForServer(
         $expect_servers[$key],
         $actual_server);
     }
 
     return $schemata;
   }
 
   private function buildComparisonSchemaForServer(
     PhabricatorConfigServerSchema $expect,
     PhabricatorConfigServerSchema $actual) {
 
     $comp_server = $actual->newEmptyClone();
 
     $all_databases = $actual->getDatabases() + $expect->getDatabases();
     foreach ($all_databases as $database_name => $database_template) {
       $actual_database = $actual->getDatabase($database_name);
       $expect_database = $expect->getDatabase($database_name);
 
       $issues = $this->compareSchemata($expect_database, $actual_database);
 
       $comp_database = $database_template->newEmptyClone()
         ->setIssues($issues);
 
       if (!$actual_database) {
         $actual_database = $expect_database->newEmptyClone();
       }
 
       if (!$expect_database) {
         $expect_database = $actual_database->newEmptyClone();
       }
 
       $all_tables =
         $actual_database->getTables() +
         $expect_database->getTables();
       foreach ($all_tables as $table_name => $table_template) {
         $actual_table = $actual_database->getTable($table_name);
         $expect_table = $expect_database->getTable($table_name);
 
         $issues = $this->compareSchemata($expect_table, $actual_table);
 
         $comp_table = $table_template->newEmptyClone()
           ->setIssues($issues);
 
         if (!$actual_table) {
           $actual_table = $expect_table->newEmptyClone();
         }
         if (!$expect_table) {
           $expect_table = $actual_table->newEmptyClone();
         }
 
         $all_columns =
           $actual_table->getColumns() +
           $expect_table->getColumns();
         foreach ($all_columns as $column_name => $column_template) {
           $actual_column = $actual_table->getColumn($column_name);
           $expect_column = $expect_table->getColumn($column_name);
 
           $issues = $this->compareSchemata($expect_column, $actual_column);
 
           $comp_column = $column_template->newEmptyClone()
             ->setIssues($issues);
 
           $comp_table->addColumn($comp_column);
         }
 
         $all_keys =
           $actual_table->getKeys() +
           $expect_table->getKeys();
         foreach ($all_keys as $key_name => $key_template) {
           $actual_key = $actual_table->getKey($key_name);
           $expect_key = $expect_table->getKey($key_name);
 
           $issues = $this->compareSchemata($expect_key, $actual_key);
 
           $comp_key = $key_template->newEmptyClone()
             ->setIssues($issues);
 
           $comp_table->addKey($comp_key);
         }
 
         $comp_database->addTable($comp_table);
       }
       $comp_server->addDatabase($comp_database);
     }
 
     return $comp_server;
   }
 
   private function compareSchemata(
     PhabricatorConfigStorageSchema $expect = null,
     PhabricatorConfigStorageSchema $actual = null) {
 
     $expect_is_key = ($expect instanceof PhabricatorConfigKeySchema);
     $actual_is_key = ($actual instanceof PhabricatorConfigKeySchema);
 
     if ($expect_is_key || $actual_is_key) {
       $missing_issue = PhabricatorConfigStorageSchema::ISSUE_MISSINGKEY;
       $surplus_issue = PhabricatorConfigStorageSchema::ISSUE_SURPLUSKEY;
     } else {
       $missing_issue = PhabricatorConfigStorageSchema::ISSUE_MISSING;
       $surplus_issue = PhabricatorConfigStorageSchema::ISSUE_SURPLUS;
     }
 
     if (!$expect && !$actual) {
       throw new Exception(pht('Can not compare two missing schemata!'));
     } else if ($expect && !$actual) {
       $issues = array($missing_issue);
     } else if ($actual && !$expect) {
       $issues = array($surplus_issue);
     } else {
       $issues = $actual->compareTo($expect);
     }
 
     return $issues;
   }
 
 
 }
diff --git a/src/infrastructure/cluster/PhabricatorDatabaseRef.php b/src/infrastructure/cluster/PhabricatorDatabaseRef.php
index 9033ba99e..95d1cc95f 100644
--- a/src/infrastructure/cluster/PhabricatorDatabaseRef.php
+++ b/src/infrastructure/cluster/PhabricatorDatabaseRef.php
@@ -1,574 +1,581 @@
 <?php
 
 final class PhabricatorDatabaseRef
   extends Phobject {
 
   const STATUS_OKAY = 'okay';
   const STATUS_FAIL = 'fail';
   const STATUS_AUTH = 'auth';
   const STATUS_REPLICATION_CLIENT = 'replication-client';
 
   const REPLICATION_OKAY = 'okay';
   const REPLICATION_MASTER_REPLICA = 'master-replica';
   const REPLICATION_REPLICA_NONE = 'replica-none';
   const REPLICATION_SLOW = 'replica-slow';
   const REPLICATION_NOT_REPLICATING = 'not-replicating';
 
   const KEY_REFS = 'cluster.db.refs';
   const KEY_INDIVIDUAL = 'cluster.db.individual';
 
   private $host;
   private $port;
   private $user;
   private $pass;
   private $disabled;
   private $isMaster;
   private $isIndividual;
 
   private $connectionLatency;
   private $connectionStatus;
   private $connectionMessage;
 
   private $replicaStatus;
   private $replicaMessage;
   private $replicaDelay;
 
   private $healthRecord;
   private $didFailToConnect;
 
   public function setHost($host) {
     $this->host = $host;
     return $this;
   }
 
   public function getHost() {
     return $this->host;
   }
 
   public function setPort($port) {
     $this->port = $port;
     return $this;
   }
 
   public function getPort() {
     return $this->port;
   }
 
   public function setUser($user) {
     $this->user = $user;
     return $this;
   }
 
   public function getUser() {
     return $this->user;
   }
 
   public function setPass(PhutilOpaqueEnvelope $pass) {
     $this->pass = $pass;
     return $this;
   }
 
   public function getPass() {
     return $this->pass;
   }
 
   public function setIsMaster($is_master) {
     $this->isMaster = $is_master;
     return $this;
   }
 
   public function getIsMaster() {
     return $this->isMaster;
   }
 
   public function setDisabled($disabled) {
     $this->disabled = $disabled;
     return $this;
   }
 
   public function getDisabled() {
     return $this->disabled;
   }
 
   public function setConnectionLatency($connection_latency) {
     $this->connectionLatency = $connection_latency;
     return $this;
   }
 
   public function getConnectionLatency() {
     return $this->connectionLatency;
   }
 
   public function setConnectionStatus($connection_status) {
     $this->connectionStatus = $connection_status;
     return $this;
   }
 
   public function getConnectionStatus() {
     if ($this->connectionStatus === null) {
       throw new PhutilInvalidStateException('queryAll');
     }
 
     return $this->connectionStatus;
   }
 
   public function setConnectionMessage($connection_message) {
     $this->connectionMessage = $connection_message;
     return $this;
   }
 
   public function getConnectionMessage() {
     return $this->connectionMessage;
   }
 
   public function setReplicaStatus($replica_status) {
     $this->replicaStatus = $replica_status;
     return $this;
   }
 
   public function getReplicaStatus() {
     return $this->replicaStatus;
   }
 
   public function setReplicaMessage($replica_message) {
     $this->replicaMessage = $replica_message;
     return $this;
   }
 
   public function getReplicaMessage() {
     return $this->replicaMessage;
   }
 
   public function setReplicaDelay($replica_delay) {
     $this->replicaDelay = $replica_delay;
     return $this;
   }
 
   public function getReplicaDelay() {
     return $this->replicaDelay;
   }
 
   public function setIsIndividual($is_individual) {
     $this->isIndividual = $is_individual;
     return $this;
   }
 
   public function getIsIndividual() {
     return $this->isIndividual;
   }
 
   public function getRefKey() {
     $host = $this->getHost();
 
     $port = $this->getPort();
     if (strlen($port)) {
       return "{$host}:{$port}";
     }
 
     return $host;
   }
 
   public static function getConnectionStatusMap() {
     return array(
       self::STATUS_OKAY => array(
         'icon' => 'fa-exchange',
         'color' => 'green',
         'label' => pht('Okay'),
       ),
       self::STATUS_FAIL => array(
         'icon' => 'fa-times',
         'color' => 'red',
         'label' => pht('Failed'),
       ),
       self::STATUS_AUTH => array(
         'icon' => 'fa-key',
         'color' => 'red',
         'label' => pht('Invalid Credentials'),
       ),
       self::STATUS_REPLICATION_CLIENT => array(
         'icon' => 'fa-eye-slash',
         'color' => 'yellow',
         'label' => pht('Missing Permission'),
       ),
     );
   }
 
   public static function getReplicaStatusMap() {
     return array(
       self::REPLICATION_OKAY => array(
         'icon' => 'fa-download',
         'color' => 'green',
         'label' => pht('Okay'),
       ),
       self::REPLICATION_MASTER_REPLICA => array(
         'icon' => 'fa-database',
         'color' => 'red',
         'label' => pht('Replicating Master'),
       ),
       self::REPLICATION_REPLICA_NONE => array(
         'icon' => 'fa-download',
         'color' => 'red',
         'label' => pht('Not A Replica'),
       ),
       self::REPLICATION_SLOW => array(
         'icon' => 'fa-hourglass',
         'color' => 'red',
         'label' => pht('Slow Replication'),
       ),
       self::REPLICATION_NOT_REPLICATING => array(
         'icon' => 'fa-exclamation-triangle',
         'color' => 'red',
         'label' => pht('Not Replicating'),
       ),
     );
   }
 
-  public static function getLiveRefs() {
+  public static function getClusterRefs() {
     $cache = PhabricatorCaches::getRequestCache();
 
     $refs = $cache->getKey(self::KEY_REFS);
     if (!$refs) {
       $refs = self::newRefs();
       $cache->setKey(self::KEY_REFS, $refs);
     }
 
     return $refs;
   }
 
   public static function getLiveIndividualRef() {
     $cache = PhabricatorCaches::getRequestCache();
 
     $ref = $cache->getKey(self::KEY_INDIVIDUAL);
     if (!$ref) {
       $ref = self::newIndividualRef();
       $cache->setKey(self::KEY_INDIVIDUAL, $ref);
     }
 
     return $ref;
   }
 
   public static function newRefs() {
     $refs = array();
 
     $default_port = PhabricatorEnv::getEnvConfig('mysql.port');
     $default_port = nonempty($default_port, 3306);
 
     $default_user = PhabricatorEnv::getEnvConfig('mysql.user');
 
     $default_pass = PhabricatorEnv::getEnvConfig('mysql.pass');
     $default_pass = new PhutilOpaqueEnvelope($default_pass);
 
     $config = PhabricatorEnv::getEnvConfig('cluster.databases');
     foreach ($config as $server) {
       $host = $server['host'];
       $port = idx($server, 'port', $default_port);
       $user = idx($server, 'user', $default_user);
       $disabled = idx($server, 'disabled', false);
 
       $pass = idx($server, 'pass');
       if ($pass) {
         $pass = new PhutilOpaqueEnvelope($pass);
       } else {
         $pass = clone $default_pass;
       }
 
       $role = $server['role'];
 
       $ref = id(new self())
         ->setHost($host)
         ->setPort($port)
         ->setUser($user)
         ->setPass($pass)
         ->setDisabled($disabled)
         ->setIsMaster(($role == 'master'));
 
       $refs[] = $ref;
     }
 
     return $refs;
   }
 
   public static function queryAll() {
     $refs = self::newRefs();
 
     foreach ($refs as $ref) {
       if ($ref->getDisabled()) {
         continue;
       }
 
       $conn = $ref->newManagementConnection();
 
       $t_start = microtime(true);
       $replica_status = false;
       try {
         $replica_status = queryfx_one($conn, 'SHOW SLAVE STATUS');
         $ref->setConnectionStatus(self::STATUS_OKAY);
       } catch (AphrontAccessDeniedQueryException $ex) {
         $ref->setConnectionStatus(self::STATUS_REPLICATION_CLIENT);
         $ref->setConnectionMessage(
           pht(
             'No permission to run "SHOW SLAVE STATUS". Grant this user '.
             '"REPLICATION CLIENT" permission to allow Phabricator to '.
             'monitor replica health.'));
       } catch (AphrontInvalidCredentialsQueryException $ex) {
         $ref->setConnectionStatus(self::STATUS_AUTH);
         $ref->setConnectionMessage($ex->getMessage());
       } catch (AphrontQueryException $ex) {
         $ref->setConnectionStatus(self::STATUS_FAIL);
 
         $class = get_class($ex);
         $message = $ex->getMessage();
         $ref->setConnectionMessage(
           pht(
             '%s: %s',
             get_class($ex),
             $ex->getMessage()));
       }
       $t_end = microtime(true);
       $ref->setConnectionLatency($t_end - $t_start);
 
       if ($replica_status !== false) {
         $is_replica = (bool)$replica_status;
         if ($ref->getIsMaster() && $is_replica) {
           $ref->setReplicaStatus(self::REPLICATION_MASTER_REPLICA);
           $ref->setReplicaMessage(
             pht(
               'This host has a "master" role, but is replicating data from '.
               'another host ("%s")!',
               idx($replica_status, 'Master_Host')));
         } else if (!$ref->getIsMaster() && !$is_replica) {
           $ref->setReplicaStatus(self::REPLICATION_REPLICA_NONE);
           $ref->setReplicaMessage(
             pht(
               'This host has a "replica" role, but is not replicating data '.
               'from a master (no output from "SHOW SLAVE STATUS").'));
         } else {
           $ref->setReplicaStatus(self::REPLICATION_OKAY);
         }
 
         if ($is_replica) {
           $latency = idx($replica_status, 'Seconds_Behind_Master');
           if (!strlen($latency)) {
             $ref->setReplicaStatus(self::REPLICATION_NOT_REPLICATING);
           } else {
             $latency = (int)$latency;
             $ref->setReplicaDelay($latency);
             if ($latency > 30) {
               $ref->setReplicaStatus(self::REPLICATION_SLOW);
               $ref->setReplicaMessage(
                 pht(
                   'This replica is lagging far behind the master. Data is at '.
                   'risk!'));
             }
           }
         }
       }
     }
 
     return $refs;
   }
 
   public function newManagementConnection() {
     return $this->newConnection(
       array(
         'retries' => 0,
         'timeout' => 2,
       ));
   }
 
   public function newApplicationConnection($database) {
     return $this->newConnection(
       array(
         'database' => $database,
       ));
   }
 
   public function isSevered() {
     // If we only have an individual database, never sever our connection to
     // it, at least for now. It's possible that using the same severing rules
     // might eventually make sense to help alleviate load-related failures,
     // but we should wait for all the cluster stuff to stabilize first.
     if ($this->getIsIndividual()) {
       return false;
     }
 
     if ($this->didFailToConnect) {
       return true;
     }
 
     $record = $this->getHealthRecord();
     $is_healthy = $record->getIsHealthy();
     if (!$is_healthy) {
       return true;
     }
 
     return false;
   }
 
   public function isReachable(AphrontDatabaseConnection $connection) {
     $record = $this->getHealthRecord();
     $should_check = $record->getShouldCheck();
 
     if ($this->isSevered() && !$should_check) {
       return false;
     }
 
     try {
       $connection->openConnection();
       $reachable = true;
     } catch (AphrontSchemaQueryException $ex) {
       // We get one of these if the database we're trying to select does not
       // exist. In this case, just re-throw the exception. This is expected
       // during first-time setup, when databases like "config" will not exist
       // yet.
       throw $ex;
     } catch (Exception $ex) {
       $reachable = false;
     }
 
     if ($should_check) {
       $record->didHealthCheck($reachable);
     }
 
     if (!$reachable) {
       $this->didFailToConnect = true;
     }
 
     return $reachable;
   }
 
   public function checkHealth() {
     $health = $this->getHealthRecord();
 
     $should_check = $health->getShouldCheck();
     if ($should_check) {
       // This does an implicit health update.
       $connection = $this->newManagementConnection();
       $this->isReachable($connection);
     }
 
     return $this;
   }
 
   public function getHealthRecord() {
     if (!$this->healthRecord) {
       $this->healthRecord = new PhabricatorDatabaseHealthRecord($this);
     }
     return $this->healthRecord;
   }
 
+  public static function getActiveDatabaseRefs() {
+    $refs = array();
+
+    foreach (self::getMasterDatabaseRefs() as $ref) {
+      $refs[] = $ref;
+    }
+
+    foreach (self::getReplicaDatabaseRefs() as $ref) {
+      $refs[] = $ref;
+    }
+
+    return $refs;
+  }
+
   public static function getMasterDatabaseRefs() {
-    $refs = self::getLiveRefs();
+    $refs = self::getClusterRefs();
 
     if (!$refs) {
       return array(self::getLiveIndividualRef());
     }
 
     $masters = array();
     foreach ($refs as $ref) {
       if ($ref->getDisabled()) {
         continue;
       }
       if ($ref->getIsMaster()) {
         $masters[] = $ref;
       }
     }
 
     return $masters;
   }
 
-  public static function getMasterDatabaseRef() {
-    // TODO: Remove this method; it no longer makes sense with application
-    // partitioning.
-
-    return head(self::getMasterDatabaseRefs());
-  }
-
   public static function getMasterDatabaseRefForDatabase($database) {
     $masters = self::getMasterDatabaseRefs();
 
     // TODO: Actually implement this.
 
     return head($masters);
   }
 
   public static function newIndividualRef() {
     $conf = PhabricatorEnv::newObjectFromConfig(
       'mysql.configuration-provider',
       array(null, 'w', null));
 
     return id(new self())
       ->setHost($conf->getHost())
       ->setPort($conf->getPort())
       ->setUser($conf->getUser())
       ->setPass($conf->getPassword())
       ->setIsIndividual(true)
       ->setIsMaster(true);
   }
 
   public static function getReplicaDatabaseRefs() {
-    $refs = self::getLiveRefs();
+    $refs = self::getClusterRefs();
 
     if (!$refs) {
       return array();
     }
 
     $replicas = array();
     foreach ($refs as $ref) {
       if ($ref->getDisabled()) {
         continue;
       }
       if ($ref->getIsMaster()) {
         continue;
       }
 
       $replicas[] = $ref;
     }
 
     return $replicas;
   }
 
   public static function getReplicaDatabaseRefForDatabase($database) {
     $replicas = self::getReplicaDatabaseRefs();
 
     // TODO: Actually implement this.
 
     // TODO: We may have multiple replicas to choose from, and could make
     // more of an effort to pick the "best" one here instead of always
     // picking the first one. Once we've picked one, we should try to use
     // the same replica for the rest of the request, though.
 
     return head($replicas);
   }
 
   private function newConnection(array $options) {
     // If we believe the database is unhealthy, don't spend as much time
     // trying to connect to it, since it's likely to continue to fail and
     // hammering it can only make the problem worse.
     $record = $this->getHealthRecord();
     if ($record->getIsHealthy()) {
       $default_retries = 3;
       $default_timeout = 10;
     } else {
       $default_retries = 0;
       $default_timeout = 2;
     }
 
     $spec = $options + array(
       'user' => $this->getUser(),
       'pass' => $this->getPass(),
       'host' => $this->getHost(),
       'port' => $this->getPort(),
       'database' => null,
       'retries' => $default_retries,
       'timeout' => $default_timeout,
     );
 
     return PhabricatorEnv::newObjectFromConfig(
       'mysql.implementation',
       array(
         $spec,
       ));
   }
 
 }
diff --git a/src/infrastructure/storage/management/PhabricatorStorageManagementAPI.php b/src/infrastructure/storage/management/PhabricatorStorageManagementAPI.php
index 87cb079c7..15b46300f 100644
--- a/src/infrastructure/storage/management/PhabricatorStorageManagementAPI.php
+++ b/src/infrastructure/storage/management/PhabricatorStorageManagementAPI.php
@@ -1,355 +1,365 @@
 <?php
 
 final class PhabricatorStorageManagementAPI extends Phobject {
 
+  private $ref;
   private $host;
   private $user;
   private $port;
   private $password;
   private $namespace;
   private $conns = array();
   private $disableUTF8MB4;
 
   const CHARSET_DEFAULT = 'CHARSET';
   const CHARSET_SORT = 'CHARSET_SORT';
   const CHARSET_FULLTEXT = 'CHARSET_FULLTEXT';
   const COLLATE_TEXT = 'COLLATE_TEXT';
   const COLLATE_SORT = 'COLLATE_SORT';
   const COLLATE_FULLTEXT = 'COLLATE_FULLTEXT';
 
   const TABLE_STATUS = 'patch_status';
 
   public function setDisableUTF8MB4($disable_utf8_mb4) {
     $this->disableUTF8MB4 = $disable_utf8_mb4;
     return $this;
   }
 
   public function getDisableUTF8MB4() {
     return $this->disableUTF8MB4;
   }
 
   public function setNamespace($namespace) {
     $this->namespace = $namespace;
     PhabricatorLiskDAO::pushStorageNamespace($namespace);
     return $this;
   }
 
   public function getNamespace() {
     return $this->namespace;
   }
 
   public function setUser($user) {
     $this->user = $user;
     return $this;
   }
 
   public function getUser() {
     return $this->user;
   }
 
   public function setPassword($password) {
     $this->password = $password;
     return $this;
   }
 
   public function getPassword() {
     return $this->password;
   }
 
   public function setHost($host) {
     $this->host = $host;
     return $this;
   }
 
   public function getHost() {
     return $this->host;
   }
 
   public function setPort($port) {
     $this->port = $port;
     return $this;
   }
 
   public function getPort() {
     return $this->port;
   }
 
+  public function setRef(PhabricatorDatabaseRef $ref) {
+    $this->ref = $ref;
+    return $this;
+  }
+
+  public function getRef() {
+    return $this->ref;
+  }
+
   public function getDatabaseName($fragment) {
     return $this->namespace.'_'.$fragment;
   }
 
   public function getDatabaseList(array $patches, $only_living = false) {
     assert_instances_of($patches, 'PhabricatorStoragePatch');
 
     $list = array();
 
     foreach ($patches as $patch) {
       if ($patch->getType() == 'db') {
         if ($only_living && $patch->isDead()) {
           continue;
         }
         $list[] = $this->getDatabaseName($patch->getName());
       }
     }
 
     return $list;
   }
 
   public function getConn($fragment) {
     $database = $this->getDatabaseName($fragment);
     $return = &$this->conns[$this->host][$this->user][$database];
     if (!$return) {
       $return = PhabricatorEnv::newObjectFromConfig(
       'mysql.implementation',
       array(
         array(
           'user'      => $this->user,
           'pass'      => $this->password,
           'host'      => $this->host,
           'port'      => $this->port,
           'database'  => $fragment
             ? $database
             : null,
         ),
       ));
     }
     return $return;
   }
 
   public function getAppliedPatches() {
     try {
       $applied = queryfx_all(
         $this->getConn('meta_data'),
         'SELECT patch FROM %T',
         self::TABLE_STATUS);
       return ipull($applied, 'patch');
     } catch (AphrontAccessDeniedQueryException $ex) {
       throw new PhutilProxyException(
         pht(
           'Failed while trying to read schema status: the database "%s" '.
           'exists, but the current user ("%s") does not have permission to '.
           'access it. GRANT the current user more permissions, or use a '.
           'different user.',
           $this->getDatabaseName('meta_data'),
           $this->getUser()),
         $ex);
     } catch (AphrontQueryException $ex) {
       return null;
     }
   }
 
   public function getPatchDurations() {
     try {
       $rows = queryfx_all(
         $this->getConn('meta_data'),
         'SELECT patch, duration FROM %T WHERE duration IS NOT NULL',
         self::TABLE_STATUS);
       return ipull($rows, 'duration', 'patch');
     } catch (AphrontQueryException $ex) {
       return array();
     }
   }
 
   public function createDatabase($fragment) {
     $info = $this->getCharsetInfo();
 
     queryfx(
       $this->getConn(null),
       'CREATE DATABASE IF NOT EXISTS %T COLLATE %T',
       $this->getDatabaseName($fragment),
       $info[self::COLLATE_TEXT]);
   }
 
   public function createTable($fragment, $table, array $cols) {
     queryfx(
       $this->getConn($fragment),
       'CREATE TABLE IF NOT EXISTS %T.%T (%Q) '.
       'ENGINE=InnoDB, COLLATE utf8_general_ci',
       $this->getDatabaseName($fragment),
       $table,
       implode(', ', $cols));
   }
 
   public function getLegacyPatches(array $patches) {
     assert_instances_of($patches, 'PhabricatorStoragePatch');
 
     try {
       $row = queryfx_one(
         $this->getConn('meta_data'),
         'SELECT version FROM %T',
         'schema_version');
       $version = $row['version'];
     } catch (AphrontQueryException $ex) {
       return array();
     }
 
     $legacy = array();
     foreach ($patches as $key => $patch) {
       if ($patch->getLegacy() !== false && $patch->getLegacy() <= $version) {
         $legacy[] = $key;
       }
     }
 
     return $legacy;
   }
 
   public function markPatchApplied($patch, $duration = null) {
     $conn = $this->getConn('meta_data');
 
     queryfx(
       $conn,
       'INSERT INTO %T (patch, applied) VALUES (%s, %d)',
       self::TABLE_STATUS,
       $patch,
       time());
 
     // We didn't add this column for a long time, so it may not exist yet.
     if ($duration !== null) {
       try {
         queryfx(
           $conn,
           'UPDATE %T SET duration = %d WHERE patch = %s',
           self::TABLE_STATUS,
           (int)floor($duration * 1000000),
           $patch);
       } catch (AphrontQueryException $ex) {
         // Just ignore this, as it almost certainly indicates that we just
         // don't have the column yet.
       }
     }
   }
 
   public function applyPatch(PhabricatorStoragePatch $patch) {
     $type = $patch->getType();
     $name = $patch->getName();
     switch ($type) {
       case 'db':
         $this->createDatabase($name);
         break;
       case 'sql':
         $this->applyPatchSQL($name);
         break;
       case 'php':
         $this->applyPatchPHP($name);
         break;
       default:
         throw new Exception(pht("Unable to apply patch of type '%s'.", $type));
     }
   }
 
   public function applyPatchSQL($sql) {
     $sql = Filesystem::readFile($sql);
     $queries = preg_split('/;\s+/', $sql);
     $queries = array_filter($queries);
 
     $conn = $this->getConn(null);
 
     $charset_info = $this->getCharsetInfo();
     foreach ($charset_info as $key => $value) {
       $charset_info[$key] = qsprintf($conn, '%T', $value);
     }
 
     foreach ($queries as $query) {
       $query = str_replace('{$NAMESPACE}', $this->namespace, $query);
 
       foreach ($charset_info as $key => $value) {
         $query = str_replace('{$'.$key.'}', $value, $query);
       }
 
       try {
         queryfx($conn, '%Q', $query);
       } catch (AphrontAccessDeniedQueryException $ex) {
         throw new PhutilProxyException(
           pht(
             'Unable to access a required database or table. This almost '.
             'always means that the user you are connecting with ("%s") does '.
             'not have sufficient permissions granted in MySQL. You can '.
             'use `bin/storage databases` to get a list of all databases '.
             'permission is required on.',
             $this->getUser()),
           $ex);
       }
     }
   }
 
   public function applyPatchPHP($script) {
     $schema_conn = $this->getConn(null);
     require_once $script;
   }
 
   public function isCharacterSetAvailable($character_set) {
     if ($character_set == 'utf8mb4') {
       if ($this->getDisableUTF8MB4()) {
         return false;
       }
     }
 
     $conn = $this->getConn(null);
     return self::isCharacterSetAvailableOnConnection($character_set, $conn);
   }
 
   public static function isCharacterSetAvailableOnConnection(
     $character_set,
     AphrontDatabaseConnection $conn) {
     $result = queryfx_one(
       $conn,
       'SELECT CHARACTER_SET_NAME FROM INFORMATION_SCHEMA.CHARACTER_SETS
         WHERE CHARACTER_SET_NAME = %s',
       $character_set);
 
     return (bool)$result;
   }
 
   public function getCharsetInfo() {
     if ($this->isCharacterSetAvailable('utf8mb4')) {
       // If utf8mb4 is available, we use it with the utf8mb4_unicode_ci
       // collation. This is most correct, and will sort properly.
 
       $charset = 'utf8mb4';
       $charset_sort = 'utf8mb4';
       $charset_full = 'utf8mb4';
       $collate_text = 'utf8mb4_bin';
       $collate_sort = 'utf8mb4_unicode_ci';
       $collate_full = 'utf8mb4_unicode_ci';
     } else {
       // If utf8mb4 is not available, we use binary for most data. This allows
       // us to store 4-byte unicode characters.
       //
       // It's possible that strings will be truncated in the middle of a
       // character on insert. We encourage users to set STRICT_ALL_TABLES
       // to prevent this.
       //
       // For "fulltext" and "sort" columns, we don't use binary.
       //
       // With "fulltext", we can not use binary because MySQL won't let us.
       // We use 3-byte utf8 instead and accept being unable to index 4-byte
       // characters.
       //
       // With "sort", if we use binary we lose case insensitivity (for
       // example, "ALincoln@logcabin.com" and "alincoln@logcabin.com" would no
       // longer be identified as the same email address). This can be very
       // confusing and is far worse overall than not supporting 4-byte unicode
       // characters, so we use 3-byte utf8 and accept limited 4-byte support as
       // a tradeoff to get sensible collation behavior. Many columns where
       // collation is important rarely contain 4-byte characters anyway, so we
       // are not giving up too much.
 
       $charset = 'binary';
       $charset_sort = 'utf8';
       $charset_full = 'utf8';
       $collate_text = 'binary';
       $collate_sort = 'utf8_general_ci';
       $collate_full = 'utf8_general_ci';
     }
 
     return array(
       self::CHARSET_DEFAULT => $charset,
       self::CHARSET_SORT => $charset_sort,
       self::CHARSET_FULLTEXT => $charset_full,
       self::COLLATE_TEXT => $collate_text,
       self::COLLATE_SORT => $collate_sort,
       self::COLLATE_FULLTEXT => $collate_full,
     );
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementAdjustWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementAdjustWorkflow.php
index 465f8728c..cd1939d92 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementAdjustWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementAdjustWorkflow.php
@@ -1,64 +1,71 @@
 <?php
 
 final class PhabricatorStorageManagementAdjustWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('adjust')
       ->setExamples('**adjust** [__options__]')
       ->setSynopsis(
         pht(
           'Make schemata adjustments to correct issues with characters sets, '.
           'collations, and keys.'))
       ->setArguments(
         array(
           array(
             'name' => 'unsafe',
             'help' => pht(
               'Permit adjustments which truncate data. This option may '.
               'destroy some data, but the lost data is usually not '.
               'important (most commonly, the ends of very long object '.
               'titles).'),
           ),
         ));
   }
 
   public function didExecute(PhutilArgumentParser $args) {
     $unsafe = $args->getArg('unsafe');
 
-    $this->requireAllPatchesApplied();
-    return $this->adjustSchemata($unsafe);
+    foreach ($this->getMasterAPIs() as $api) {
+      $this->requireAllPatchesApplied($api);
+      $err = $this->adjustSchemata($api, $unsafe);
+      if ($err) {
+        return $err;
+      }
+    }
+
+    return 0;
   }
 
-  private function requireAllPatchesApplied() {
-    $api = $this->getAPI();
+  private function requireAllPatchesApplied(
+    PhabricatorStorageManagementAPI $api) {
     $applied = $api->getAppliedPatches();
 
     if ($applied === null) {
       throw new PhutilArgumentUsageException(
         pht(
           'You have not initialized the database yet. You must initialize '.
           'the database before you can adjust schemata. Run `%s` '.
           'to initialize the database.',
           'storage upgrade'));
     }
 
     $applied = array_fuse($applied);
 
     $patches = $this->getPatches();
     $patches = mpull($patches, null, 'getFullKey');
     $missing = array_diff_key($patches, $applied);
 
     if ($missing) {
       throw new PhutilArgumentUsageException(
         pht(
           'You have not applied all available storage patches yet. You must '.
           'apply all available patches before you can adjust schemata. '.
           'Run `%s` to show patch status, and `%s` to apply missing patches.',
           'storage status',
           'storage upgrade'));
     }
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDatabasesWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDatabasesWorkflow.php
index e29dd60e3..ef6735ec8 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDatabasesWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDatabasesWorkflow.php
@@ -1,26 +1,27 @@
 <?php
 
 final class PhabricatorStorageManagementDatabasesWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('databases')
       ->setExamples('**databases** [__options__]')
       ->setSynopsis(pht('List Phabricator databases.'));
   }
 
   protected function isReadOnlyWorkflow() {
     return true;
   }
 
   public function didExecute(PhutilArgumentParser $args) {
-    $api     = $this->getAPI();
+    $api = $this->getAnyAPI();
+
     $patches = $this->getPatches();
 
     $databases = $api->getDatabaseList($patches, true);
     echo implode("\n", $databases)."\n";
     return 0;
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDestroyWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDestroyWorkflow.php
index 9bf590614..b9cf82b2c 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDestroyWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDestroyWorkflow.php
@@ -1,89 +1,90 @@
 <?php
 
 final class PhabricatorStorageManagementDestroyWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('destroy')
       ->setExamples('**destroy** [__options__]')
       ->setSynopsis(pht('Permanently destroy all storage and data.'))
       ->setArguments(
         array(
           array(
             'name'  => 'unittest-fixtures',
             'help'  => pht(
               'Restrict **destroy** operations to databases created '.
               'by %s test fixtures.',
               'PhabricatorTestCase'),
           ),
         ));
   }
 
   public function didExecute(PhutilArgumentParser $args) {
     $console = PhutilConsole::getConsole();
 
+    $api = $this->getSingleAPI();
+
     if (!$this->isDryRun() && !$this->isForce()) {
       $console->writeOut(
         phutil_console_wrap(
           pht(
             'Are you completely sure you really want to permanently destroy '.
             'all storage for Phabricator data? This operation can not be '.
             'undone and your data will not be recoverable if you proceed.')));
 
       if (!phutil_console_confirm(pht('Permanently destroy all data?'))) {
         $console->writeOut("%s\n", pht('Cancelled.'));
         exit(1);
       }
 
       if (!phutil_console_confirm(pht('Really destroy all data forever?'))) {
         $console->writeOut("%s\n", pht('Cancelled.'));
         exit(1);
       }
     }
 
-    $api     = $this->getAPI();
     $patches = $this->getPatches();
 
     if ($args->getArg('unittest-fixtures')) {
       $conn = $api->getConn(null);
       $databases = queryfx_all(
         $conn,
         'SELECT DISTINCT(TABLE_SCHEMA) AS db '.
         'FROM INFORMATION_SCHEMA.TABLES '.
         'WHERE TABLE_SCHEMA LIKE %>',
         PhabricatorTestCase::NAMESPACE_PREFIX);
       $databases = ipull($databases, 'db');
     } else {
       $databases   = $api->getDatabaseList($patches);
       $databases[] = $api->getDatabaseName('meta_data');
 
       // These are legacy databases that were dropped long ago. See T2237.
       $databases[] = $api->getDatabaseName('phid');
       $databases[] = $api->getDatabaseName('directory');
     }
 
     foreach ($databases as $database) {
       if ($this->isDryRun()) {
         $console->writeOut(
           "%s\n",
           pht("DRYRUN: Would drop database '%s'.", $database));
       } else {
         $console->writeOut(
           "%s\n",
           pht("Dropping database '%s'...", $database));
         queryfx(
           $api->getConn(null),
           'DROP DATABASE IF EXISTS %T',
           $database);
       }
     }
 
     if (!$this->isDryRun()) {
       $console->writeOut("%s\n", pht('Storage was destroyed.'));
     }
 
     return 0;
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDumpWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDumpWorkflow.php
index 6a481d470..7195e735c 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDumpWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementDumpWorkflow.php
@@ -1,213 +1,213 @@
 <?php
 
 final class PhabricatorStorageManagementDumpWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('dump')
       ->setExamples('**dump** [__options__]')
       ->setSynopsis(pht('Dump all data in storage to stdout.'))
       ->setArguments(
         array(
           array(
             'name' => 'for-replica',
             'help' => pht(
               'Add __--master-data__ to the __mysqldump__ command, '.
               'generating a CHANGE MASTER statement in the output.'),
           ),
           array(
             'name' => 'output',
             'param' => 'file',
             'help' => pht(
               'Write output directly to disk. This handles errors better '.
               'than using pipes. Use with __--compress__ to gzip the '.
               'output.'),
           ),
           array(
             'name' => 'compress',
             'help' => pht(
               'With __--output__, write a compressed file to disk instead '.
               'of a plaintext file.'),
           ),
           array(
             'name' => 'overwrite',
             'help' => pht(
               'With __--output__, overwrite the output file if it already '.
               'exists.'),
           ),
         ));
   }
 
   protected function isReadOnlyWorkflow() {
     return true;
   }
 
   public function didExecute(PhutilArgumentParser $args) {
-    $api = $this->getAPI();
+    $api = $this->getSingleAPI();
     $patches = $this->getPatches();
 
     $console = PhutilConsole::getConsole();
 
     $applied = $api->getAppliedPatches();
     if ($applied === null) {
       $namespace = $api->getNamespace();
       $console->writeErr(
         pht(
           '**Storage Not Initialized**: There is no database storage '.
           'initialized in this storage namespace ("%s"). Use '.
           '**%s** to initialize storage.',
           $namespace,
           './bin/storage upgrade'));
       return 1;
     }
 
     $databases = $api->getDatabaseList($patches, true);
 
     list($host, $port) = $this->getBareHostAndPort($api->getHost());
 
     $has_password = false;
 
     $password = $api->getPassword();
     if ($password) {
       if (strlen($password->openEnvelope())) {
         $has_password = true;
       }
     }
 
     $output_file = $args->getArg('output');
     $is_compress = $args->getArg('compress');
     $is_overwrite = $args->getArg('overwrite');
 
     if ($is_compress) {
       if ($output_file === null) {
         throw new PhutilArgumentUsageException(
           pht(
             'The "--compress" flag can only be used alongside "--output".'));
       }
     }
 
     if ($is_overwrite) {
       if ($output_file === null) {
         throw new PhutilArgumentUsageException(
           pht(
             'The "--overwrite" flag can only be used alongside "--output".'));
       }
     }
 
     if ($output_file !== null) {
       if (Filesystem::pathExists($output_file)) {
         if (!$is_overwrite) {
           throw new PhutilArgumentUsageException(
             pht(
               'Output file "%s" already exists. Use "--overwrite" '.
               'to overwrite.',
               $output_file));
         }
       }
     }
 
     $argv = array();
     $argv[] = '--hex-blob';
     $argv[] = '--single-transaction';
     $argv[] = '--default-character-set=utf8';
 
     if ($args->getArg('for-replica')) {
       $argv[] = '--master-data';
     }
 
     $argv[] = '-u';
     $argv[] = $api->getUser();
     $argv[] = '-h';
     $argv[] = $host;
 
     if ($port) {
       $argv[] = '--port';
       $argv[] = $port;
     }
 
     $argv[] = '--databases';
     foreach ($databases as $database) {
       $argv[] = $database;
     }
 
 
     if ($has_password) {
       $command = csprintf('mysqldump -p%P %Ls', $password, $argv);
     } else {
       $command = csprintf('mysqldump %Ls', $argv);
     }
 
     // If we aren't writing to a file, just passthru the command.
     if ($output_file === null) {
       return phutil_passthru('%C', $command);
     }
 
     // If we are writing to a file, stream the command output to disk. This
     // mode makes sure the whole command fails if there's an error (commonly,
     // a full disk). See T6996 for discussion.
 
     if ($is_compress) {
       $file = gzopen($output_file, 'wb');
     } else {
       $file = fopen($output_file, 'wb');
     }
 
     if (!$file) {
       throw new Exception(
         pht(
           'Failed to open file "%s" for writing.',
           $file));
     }
 
     $future = new ExecFuture('%C', $command);
 
     $lines = new LinesOfALargeExecFuture($future);
 
     try {
       foreach ($lines as $line) {
         $line = $line."\n";
         if ($is_compress) {
           $ok = gzwrite($file, $line);
         } else {
           $ok = fwrite($file, $line);
         }
 
         if ($ok !== strlen($line)) {
           throw new Exception(
             pht(
               'Failed to write %d byte(s) to file "%s".',
               new PhutilNumber(strlen($line)),
               $output_file));
         }
       }
 
       if ($is_compress) {
         $ok = gzclose($file);
       } else {
         $ok = fclose($file);
       }
 
       if ($ok !== true) {
         throw new Exception(
           pht(
             'Failed to close file "%s".',
             $output_file));
       }
     } catch (Exception $ex) {
       // If we might have written a partial file to disk, try to remove it so
       // we don't leave any confusing artifacts laying around.
 
       try {
         Filesystem::remove($output_file);
       } catch (Exception $ex) {
         // Ignore any errors we hit.
       }
 
       throw $ex;
     }
 
     return 0;
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementProbeWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementProbeWorkflow.php
index c45146e6d..a75afe69e 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementProbeWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementProbeWorkflow.php
@@ -1,105 +1,106 @@
 <?php
 
 final class PhabricatorStorageManagementProbeWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('probe')
       ->setExamples('**probe**')
       ->setSynopsis(pht('Show approximate table sizes.'));
   }
 
   protected function isReadOnlyWorkflow() {
     return true;
   }
 
   public function didExecute(PhutilArgumentParser $args) {
+    $api = $this->getSingleAPI();
+
     $console = PhutilConsole::getConsole();
     $console->writeErr(
       "%s\n",
       pht('Analyzing table sizes (this may take a moment)...'));
 
-    $api       = $this->getAPI();
-    $patches   = $this->getPatches();
+    $patches = $this->getPatches();
     $databases = $api->getDatabaseList($patches, true);
 
     $conn_r = $api->getConn(null);
 
     $data = array();
     foreach ($databases as $database) {
       queryfx($conn_r, 'USE %C', $database);
       $tables = queryfx_all(
         $conn_r,
         'SHOW TABLE STATUS');
       $tables = ipull($tables, null, 'Name');
       $data[$database] = $tables;
     }
 
     $totals = array_fill_keys(array_keys($data), 0);
     $overall = 0;
 
     foreach ($data as $db => $tables) {
       foreach ($tables as $table => $info) {
         $table_size = $info['Data_length'] + $info['Index_length'];
 
         $data[$db][$table]['_totalSize'] = $table_size;
         $totals[$db] += $table_size;
         $overall += $table_size;
       }
     }
 
     asort($totals);
 
     $table = id(new PhutilConsoleTable())
       ->setShowHeader(false)
       ->setPadding(2)
       ->addColumn('name', array('title' => pht('Database / Table')))
       ->addColumn('size', array('title' => pht('Size')))
       ->addColumn('percentage', array('title' => pht('Percentage')));
 
     foreach ($totals as $db => $size) {
       list($database_size, $database_percentage) = $this->formatSize(
         $totals[$db],
         $overall);
 
       $table->addRow(array(
         'name' => tsprintf('**%s**', $db),
         'size' => tsprintf('**%s**', $database_size),
         'percentage' => tsprintf('**%s**', $database_percentage),
       ));
       $data[$db] = isort($data[$db], '_totalSize');
       foreach ($data[$db] as $table_name => $info) {
         list($table_size, $table_percentage) = $this->formatSize(
           $info['_totalSize'],
           $overall);
 
         $table->addRow(array(
           'name' => '    '.$table_name,
           'size' => $table_size,
           'percentage' => $table_percentage,
         ));
       }
     }
 
     list($overall_size, $overall_percentage) = $this->formatSize(
       $overall,
       $overall);
     $table->addRow(array(
       'name' => tsprintf('**%s**', pht('TOTAL')),
       'size' => tsprintf('**%s**', $overall_size),
       'percentage' => tsprintf('**%s**', $overall_percentage),
     ));
 
     $table->draw();
     return 0;
   }
 
   private function formatSize($n, $o) {
     return array(
       sprintf('%8.8s MB', number_format($n / (1024 * 1024), 1)),
       sprintf('%3.1f%%', 100 * ($n / $o)),
     );
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementQuickstartWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementQuickstartWorkflow.php
index 2d67d97e4..e181063ac 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementQuickstartWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementQuickstartWorkflow.php
@@ -1,169 +1,180 @@
 <?php
 
 final class PhabricatorStorageManagementQuickstartWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('quickstart')
       ->setExamples('**quickstart** [__options__]')
       ->setSynopsis(
         pht(
           'Generate a new quickstart database dump. This command is mostly '.
           'useful when developing Phabricator.'))
       ->setArguments(
         array(
           array(
             'name'  => 'output',
             'param' => 'file',
             'help'  => pht('Specify output file to write.'),
           ),
         ));
   }
 
   public function execute(PhutilArgumentParser $args) {
     parent::execute($args);
 
     $output = $args->getArg('output');
     if (!$output) {
       throw new PhutilArgumentUsageException(
         pht(
           'Specify a file to write with `%s`.',
           '--output'));
     }
 
     $namespace = 'phabricator_quickstart_'.Filesystem::readRandomCharacters(8);
 
     $bin = dirname(phutil_get_library_root('phabricator')).'/bin/storage';
 
-    if (!$this->getAPI()->isCharacterSetAvailable('utf8mb4')) {
+    // We don't care which database we're using to generate a quickstart file,
+    // since all of the schemata should be identical.
+    $api = $this->getAnyAPI();
+
+    $ref = $api->getRef();
+    $ref_key = $ref->getRefKey();
+
+    if (!$api->isCharacterSetAvailable('utf8mb4')) {
       throw new PhutilArgumentUsageException(
         pht(
           'You can only generate a new quickstart file if MySQL supports '.
           'the %s character set (available in MySQL 5.5 and newer). The '.
           'configured server does not support %s.',
           'utf8mb4',
           'utf8mb4'));
     }
 
     $err = phutil_passthru(
-      '%s upgrade --force --no-quickstart --namespace %s',
+      '%s upgrade --force --no-quickstart --namespace %s --ref %s',
       $bin,
-      $namespace);
+      $namespace,
+      $ref_key);
     if ($err) {
       return $err;
     }
 
     $err = phutil_passthru(
-      '%s adjust --force --namespace %s',
+      '%s adjust --force --namespace %s --ref %s',
       $bin,
-      $namespace);
+      $namespace,
+      $ref_key);
     if ($err) {
       return $err;
     }
 
     $tmp = new TempFile();
     $err = phutil_passthru(
-      '%s dump --namespace %s > %s',
+      '%s dump --namespace %s --ref %s > %s',
       $bin,
       $namespace,
+      $ref_key,
       $tmp);
     if ($err) {
       return $err;
     }
 
     $err = phutil_passthru(
-      '%s destroy --force --namespace %s',
+      '%s destroy --force --namespace %s --ref %s',
       $bin,
-      $namespace);
+      $namespace,
+      $ref_key);
     if ($err) {
       return $err;
     }
 
     $dump = Filesystem::readFile($tmp);
 
     $dump = str_replace(
       $namespace,
       '{$NAMESPACE}',
       $dump);
 
     // NOTE: This is a hack. We can not use `binary` for these columns, because
     // they are a part of a fulltext index. This regex is avoiding matching a
     // possible NOT NULL at the end of the line.
     $old = $dump;
     $dump = preg_replace(
       '/`corpus` longtext CHARACTER SET .*? COLLATE [^\s,]+/mi',
       '`corpus` longtext CHARACTER SET {$CHARSET_FULLTEXT} '.
         'COLLATE {$COLLATE_FULLTEXT}',
       $dump);
     if ($dump == $old) {
       // If we didn't make any changes, yell about it. We'll produce an invalid
       // dump otherwise.
       throw new PhutilArgumentUsageException(
         pht(
           'Failed to apply hack to adjust %s search column!',
           'FULLTEXT'));
     }
 
     $dump = str_replace(
       'utf8mb4_bin',
       '{$COLLATE_TEXT}',
       $dump);
 
     $dump = str_replace(
       'utf8mb4_unicode_ci',
       '{$COLLATE_SORT}',
       $dump);
 
     $dump = str_replace(
       'utf8mb4',
       '{$CHARSET}',
       $dump);
 
     $old = $dump;
     $dump = preg_replace(
       '/CHARACTER SET {\$CHARSET} COLLATE {\$COLLATE_SORT}/mi',
       'CHARACTER SET {$CHARSET_SORT} COLLATE {$COLLATE_SORT}',
       $dump);
     if ($dump == $old) {
       throw new PhutilArgumentUsageException(
         pht('Failed to adjust SORT columns!'));
     }
 
     // Strip out a bunch of unnecessary commands which make the dump harder
     // to handle and slower to import.
 
     // Remove character set adjustments and key disables.
     $dump = preg_replace(
       '(^/\*.*\*/;$)m',
       '',
       $dump);
 
     // Remove comments.
     $dump = preg_replace('/^--.*$/m', '', $dump);
 
     // Remove table drops, locks, and unlocks. These are never relevant when
     // performing a quickstart.
     $dump = preg_replace(
       '/^(DROP TABLE|LOCK TABLES|UNLOCK TABLES).*$/m',
       '',
       $dump);
 
     // Collapse adjacent newlines.
     $dump = preg_replace('/\n\s*\n/', "\n", $dump);
 
     $dump = str_replace(';', ";\n", $dump);
     $dump = trim($dump)."\n";
 
     Filesystem::writeFile($output, $dump);
 
     $console = PhutilConsole::getConsole();
     $console->writeOut(
       "**<bg:green> %s </bg>** %s\n",
       pht('SUCCESS'),
       pht('Wrote fresh quickstart SQL.'));
 
     return 0;
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementRenamespaceWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementRenamespaceWorkflow.php
index 51d3dd148..0540bb3eb 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementRenamespaceWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementRenamespaceWorkflow.php
@@ -1,212 +1,216 @@
 <?php
 
 final class PhabricatorStorageManagementRenamespaceWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('renamespace')
       ->setExamples(
         '**renamespace** [__options__] '.
         '--input __dump.sql__ --from __old__ --to __new__ > __out.sql__')
       ->setSynopsis(pht('Change the database namespace of a .sql dump file.'))
       ->setArguments(
         array(
           array(
             'name' => 'input',
             'param' => 'file',
             'help' => pht('SQL dumpfile to process.'),
           ),
           array(
             'name' => 'live',
             'help' => pht(
               'Generate a live dump instead of processing a file on disk.'),
           ),
           array(
             'name' => 'from',
             'param' => 'namespace',
             'help' => pht('Current database namespace used by dumpfile.'),
           ),
           array(
             'name' => 'to',
             'param' => 'namespace',
             'help' => pht('Desired database namespace for output.'),
           ),
           array(
             'name' => 'output',
             'param' => 'file',
             'help' => pht('Write output directly to a file on disk.'),
           ),
           array(
             'name' => 'compress',
             'help' => pht('Emit gzipped output instead of plain text.'),
           ),
           array(
             'name' => 'overwrite',
             'help' => pht(
               'With __--output__, write to disk even if the file already '.
               'exists.'),
           ),
         ));
   }
 
   protected function isReadOnlyWorkflow() {
     return true;
   }
 
   public function didExecute(PhutilArgumentParser $args) {
     $console = PhutilConsole::getConsole();
 
     $input = $args->getArg('input');
     $is_live = $args->getArg('live');
     if (!strlen($input) && !$is_live) {
       throw new PhutilArgumentUsageException(
         pht(
-          'Specify the dumpfile to read with "--in", or use "--live" to '.
+          'Specify the dumpfile to read with "--input", or use "--live" to '.
           'generate one automatically.'));
     }
 
     $from = $args->getArg('from');
     if (!strlen($from)) {
       throw new PhutilArgumentUsageException(
         pht(
           'Specify namespace to rename from with %s.',
           '--from'));
     }
 
     $to = $args->getArg('to');
     if (!strlen($to)) {
       throw new PhutilArgumentUsageException(
         pht(
           'Specify namespace to rename to with %s.',
           '--to'));
     }
 
 
     $output_file = $args->getArg('output');
     $is_overwrite = $args->getArg('overwrite');
     $is_compress = $args->getArg('compress');
 
     if ($is_overwrite) {
       if ($output_file === null) {
         throw new PhutilArgumentUsageException(
           pht(
             'The "--overwrite" flag can only be used alongside "--output".'));
       }
     }
 
     if ($output_file !== null) {
       if (Filesystem::pathExists($output_file)) {
         if (!$is_overwrite) {
           throw new PhutilArgumentUsageException(
             pht(
               'Output file "%s" already exists. Use "--overwrite" '.
               'to overwrite.',
               $output_file));
         }
       }
     }
 
     if ($is_live) {
+      $api = $this->getSingleAPI();
+      $ref_key = $api->getRef()->getRefKey();
+
       $root = dirname(phutil_get_library_root('phabricator'));
 
       $future = new ExecFuture(
-        '%R dump',
-        $root.'/bin/storage');
+        '%R dump --ref %s',
+        $root.'/bin/storage',
+        $ref_key);
 
       $lines = new LinesOfALargeExecFuture($future);
     } else {
       $lines = new LinesOfALargeFile($input);
     }
 
     if ($output_file === null) {
       $file = fopen('php://stdout', 'wb');
       $output_name = pht('stdout');
     } else {
       if ($is_compress) {
         $file = gzopen($output_file, 'wb');
       } else {
         $file = fopen($output_file, 'wb');
       }
       $output_name = $output_file;
     }
 
     if (!$file) {
       throw new Exception(
         pht(
           'Failed to open output file "%s" for writing.',
           $output_name));
     }
 
     $name_pattern = preg_quote($from, '@');
 
     $patterns = array(
       'use' => '@^(USE `)('.$name_pattern.')(_.*)$@',
       'create' => '@^(CREATE DATABASE /\*.*?\*/ `)('.$name_pattern.')(_.*)$@',
     );
 
     $found = array_fill_keys(array_keys($patterns), 0);
 
     try {
       $matches = null;
       foreach ($lines as $line) {
 
         foreach ($patterns as $key => $pattern) {
           if (preg_match($pattern, $line, $matches)) {
             $line = $matches[1].$to.$matches[3];
             $found[$key]++;
           }
         }
 
         $data = $line."\n";
 
         if ($is_compress) {
           $bytes = gzwrite($file, $data);
         } else {
           $bytes = fwrite($file, $data);
         }
 
         if ($bytes !== strlen($data)) {
           throw new Exception(
             pht(
               'Failed to write %d byte(s) to "%s".',
               new PhutilNumber(strlen($data)),
               $output_name));
         }
       }
 
       if ($is_compress) {
         $ok = gzclose($file);
       } else {
         $ok = fclose($file);
       }
 
       if ($ok !== true) {
         throw new Exception(
           pht(
             'Failed to close file "%s".',
             $output_file));
       }
     } catch (Exception $ex) {
       try {
         if ($output_file !== null) {
           Filesystem::remove($output_file);
         }
       } catch (Exception $ex) {
         // Ignore any exception.
       }
 
       throw $ex;
     }
 
     // Give the user a chance to catch things if the results are crazy.
     $console->writeErr(
       pht(
         'Adjusted **%s** create statements and **%s** use statements.',
         new PhutilNumber($found['create']),
         new PhutilNumber($found['use']))."\n");
 
     return 0;
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementShellWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementShellWorkflow.php
index 84e2ed548..0bf185a08 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementShellWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementShellWorkflow.php
@@ -1,42 +1,42 @@
 <?php
 
 final class PhabricatorStorageManagementShellWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('shell')
       ->setExamples('**shell** [__options__]')
       ->setSynopsis(pht('Launch an interactive shell.'));
   }
 
   protected function isReadOnlyWorkflow() {
     return true;
   }
 
   public function execute(PhutilArgumentParser $args) {
-    $api = $this->getAPI();
+    $api = $this->getSingleAPI();
     list($host, $port) = $this->getBareHostAndPort($api->getHost());
 
     $flag_port = $port
       ? csprintf('--port %d', $port)
       : '';
 
     $flag_password = '';
     $password = $api->getPassword();
     if ($password) {
       if (strlen($password->openEnvelope())) {
         $flag_password = csprintf('--password=%P', $password);
       }
     }
 
     return phutil_passthru(
       'mysql --protocol=TCP --default-character-set=utf8mb4 '.
       '-u %s %C -h %s %C',
       $api->getUser(),
       $flag_password,
       $host,
       $flag_port);
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementStatusWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementStatusWorkflow.php
index f9e91427d..88174d22a 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementStatusWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementStatusWorkflow.php
@@ -1,65 +1,69 @@
 <?php
 
 final class PhabricatorStorageManagementStatusWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('status')
       ->setExamples('**status** [__options__]')
       ->setSynopsis(pht('Show patch application status.'));
   }
 
   protected function isReadOnlyWorkflow() {
     return true;
   }
 
   public function didExecute(PhutilArgumentParser $args) {
-    $api     = $this->getAPI();
-    $patches = $this->getPatches();
+    foreach ($this->getAPIs() as $api) {
+      $patches = $this->getPatches();
+      $applied = $api->getAppliedPatches();
 
-    $applied = $api->getAppliedPatches();
+      if ($applied === null) {
+        echo phutil_console_format(
+          "**%s**: %s\n",
+          pht('Database Not Initialized'),
+          pht('Run **%s** to initialize.', './bin/storage upgrade'));
 
-    if ($applied === null) {
-      echo phutil_console_format(
-        "**%s**: %s\n",
-        pht('Database Not Initialized'),
-        pht('Run **%s** to initialize.', './bin/storage upgrade'));
+        return 1;
+      }
 
-      return 1;
-    }
+      $ref = $api->getRef();
 
-    $table = id(new PhutilConsoleTable())
-      ->setShowHeader(false)
-      ->addColumn('id',       array('title' => pht('ID')))
-      ->addColumn('status',   array('title' => pht('Status')))
-      ->addColumn('duration', array('title' => pht('Duration')))
-      ->addColumn('type',     array('title' => pht('Type')))
-      ->addColumn('name',     array('title' => pht('Name')));
+      $table = id(new PhutilConsoleTable())
+        ->setShowHeader(false)
+        ->addColumn('id', array('title' => pht('ID')))
+        ->addColumn('host', array('title' => pht('Host')))
+        ->addColumn('status', array('title' => pht('Status')))
+        ->addColumn('duration', array('title' => pht('Duration')))
+        ->addColumn('type', array('title' => pht('Type')))
+        ->addColumn('name', array('title' => pht('Name')));
 
-    $durations = $api->getPatchDurations();
+      $durations = $api->getPatchDurations();
 
-    foreach ($patches as $patch) {
-      $duration = idx($durations, $patch->getFullKey());
-      if ($duration === null) {
-        $duration = '-';
-      } else {
-        $duration = pht('%s us', new PhutilNumber($duration));
+      foreach ($patches as $patch) {
+        $duration = idx($durations, $patch->getFullKey());
+        if ($duration === null) {
+          $duration = '-';
+        } else {
+          $duration = pht('%s us', new PhutilNumber($duration));
+        }
+
+        $table->addRow(array(
+          'id' => $patch->getFullKey(),
+          'host' => $ref->getRefKey(),
+          'status' => in_array($patch->getFullKey(), $applied)
+            ? pht('Applied')
+            : pht('Not Applied'),
+          'duration' => $duration,
+          'type' => $patch->getType(),
+          'name' => $patch->getName(),
+        ));
       }
 
-      $table->addRow(array(
-        'id' => $patch->getFullKey(),
-        'status' => in_array($patch->getFullKey(), $applied)
-          ? pht('Applied')
-          : pht('Not Applied'),
-        'duration' => $duration,
-        'type' => $patch->getType(),
-        'name' => $patch->getName(),
-      ));
+      $table->draw();
     }
-
-    $table->draw();
     return 0;
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementUpgradeWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementUpgradeWorkflow.php
index 70dc90fdf..ce019f6cc 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementUpgradeWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementUpgradeWorkflow.php
@@ -1,88 +1,96 @@
 <?php
 
 final class PhabricatorStorageManagementUpgradeWorkflow
   extends PhabricatorStorageManagementWorkflow {
 
   protected function didConstruct() {
     $this
       ->setName('upgrade')
       ->setExamples('**upgrade** [__options__]')
       ->setSynopsis(pht('Upgrade database schemata.'))
       ->setArguments(
         array(
           array(
             'name'  => 'apply',
             'param' => 'patch',
             'help'  => pht(
               'Apply __patch__ explicitly. This is an advanced feature for '.
               'development and debugging; you should not normally use this '.
               'flag. This skips adjustment.'),
           ),
           array(
             'name'  => 'no-quickstart',
             'help'  => pht(
               'Build storage patch-by-patch from scratch, even if it could '.
               'be loaded from the quickstart template.'),
           ),
           array(
             'name'  => 'init-only',
             'help'  => pht(
               'Initialize storage only; do not apply patches or adjustments.'),
           ),
           array(
             'name' => 'no-adjust',
             'help' => pht(
               'Do not apply storage adjustments after storage upgrades.'),
           ),
         ));
   }
 
   public function didExecute(PhutilArgumentParser $args) {
     $console = PhutilConsole::getConsole();
     $patches = $this->getPatches();
 
     if (!$this->isDryRun() && !$this->isForce()) {
       $console->writeOut(
         phutil_console_wrap(
           pht(
             'Before running storage upgrades, you should take down the '.
             'Phabricator web interface and stop any running Phabricator '.
             'daemons (you can disable this warning with %s).',
             '--force')));
 
       if (!phutil_console_confirm(pht('Are you ready to continue?'))) {
         $console->writeOut("%s\n", pht('Cancelled.'));
         return 1;
       }
     }
 
     $apply_only = $args->getArg('apply');
     if ($apply_only) {
       if (empty($patches[$apply_only])) {
         throw new PhutilArgumentUsageException(
           pht(
             "%s argument '%s' is not a valid patch. ".
             "Use '%s' to show patch status.",
             '--apply',
             $apply_only,
             './bin/storage status'));
       }
     }
 
     $no_quickstart = $args->getArg('no-quickstart');
     $init_only     = $args->getArg('init-only');
     $no_adjust     = $args->getArg('no-adjust');
 
-    $this->upgradeSchemata($apply_only, $no_quickstart, $init_only);
+    $apis = $this->getMasterAPIs();
 
-    if ($no_adjust || $init_only || $apply_only) {
-      $console->writeOut(
-        "%s\n",
-        pht('Declining to apply storage adjustments.'));
-      return 0;
-    } else {
-      return $this->adjustSchemata(false);
+    foreach ($apis as $api) {
+      $this->upgradeSchemata($api, $apply_only, $no_quickstart, $init_only);
+
+      if ($no_adjust || $init_only || $apply_only) {
+        $console->writeOut(
+          "%s\n",
+          pht('Declining to apply storage adjustments.'));
+      } else {
+        $err = $this->adjustSchemata($api, false);
+        if ($err) {
+          return $err;
+        }
+      }
     }
+
+    return 0;
   }
 
 }
diff --git a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementWorkflow.php b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementWorkflow.php
index 7bfb4d873..2404918b5 100644
--- a/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementWorkflow.php
+++ b/src/infrastructure/storage/management/workflow/PhabricatorStorageManagementWorkflow.php
@@ -1,964 +1,1019 @@
 <?php
 
 abstract class PhabricatorStorageManagementWorkflow
   extends PhabricatorManagementWorkflow {
 
-  private $api;
+  private $apis = array();
   private $dryRun;
   private $force;
   private $patches;
 
   private $didInitialize;
 
-  final public function getAPI() {
-    return $this->api;
+  final public function setAPIs(array $apis) {
+    $this->apis = $apis;
+    return $this;
   }
 
-  final public function setAPI(PhabricatorStorageManagementAPI $api) {
-    $this->api = $api;
-    return $this;
+  final public function getAnyAPI() {
+    return head($this->getAPIs());
+  }
+
+  final public function getMasterAPIs() {
+    $apis = $this->getAPIs();
+
+    $results = array();
+    foreach ($apis as $api) {
+      if ($api->getRef()->getIsMaster()) {
+        $results[] = $api;
+      }
+    }
+
+    if (!$results) {
+      throw new PhutilArgumentUsageException(
+        pht(
+          'This command only operates on database masters, but the selected '.
+          'database hosts do not include any masters.'));
+    }
+
+    return $results;
+  }
+
+  final public function getSingleAPI() {
+    $apis = $this->getAPIs();
+    if (count($apis) == 1) {
+      return head($apis);
+    }
+
+    throw new PhutilArgumentUsageException(
+      pht(
+        'Phabricator is configured in cluster mode, with multiple database '.
+        'hosts. Use "--host" to specify which host you want to operate on.'));
+  }
+
+  final public function getAPIs() {
+    return $this->apis;
   }
 
   final protected function isDryRun() {
     return $this->dryRun;
   }
 
   final protected function setDryRun($dry_run) {
     $this->dryRun = $dry_run;
     return $this;
   }
 
   final protected function isForce() {
     return $this->force;
   }
 
   final protected function setForce($force) {
     $this->force = $force;
     return $this;
   }
 
   public function getPatches() {
     return $this->patches;
   }
 
   public function setPatches(array $patches) {
     assert_instances_of($patches, 'PhabricatorStoragePatch');
     $this->patches = $patches;
     return $this;
   }
 
   protected function isReadOnlyWorkflow() {
     return false;
   }
 
   public function execute(PhutilArgumentParser $args) {
     $this->setDryRun($args->getArg('dryrun'));
     $this->setForce($args->getArg('force'));
 
     if (!$this->isReadOnlyWorkflow()) {
       if (PhabricatorEnv::isReadOnly()) {
         if ($this->isForce()) {
           PhabricatorEnv::setReadOnly(false, null);
         } else {
           throw new PhutilArgumentUsageException(
             pht(
               'Phabricator is currently in read-only mode. Use --force to '.
               'override this mode.'));
         }
       }
     }
 
     return $this->didExecute($args);
   }
 
   public function didExecute(PhutilArgumentParser $args) {}
 
-  private function loadSchemata() {
-    $query = id(new PhabricatorConfigSchemaQuery())
-      ->setAPI($this->getAPI());
+  private function loadSchemata(PhabricatorStorageManagementAPI $api) {
+    $query = id(new PhabricatorConfigSchemaQuery());
+
+    $ref = $api->getRef();
+    $ref_key = $ref->getRefKey();
 
-    $actual = $query->loadActualSchema();
-    $expect = $query->loadExpectedSchema();
-    $comp = $query->buildComparisonSchema($expect, $actual);
+    $query->setAPIs(array($api));
+    $query->setRefs(array($ref));
 
-    return array($comp, $expect, $actual);
+    $actual = $query->loadActualSchemata();
+    $expect = $query->loadExpectedSchemata();
+    $comp = $query->buildComparisonSchemata($expect, $actual);
+
+    return array(
+      $comp[$ref_key],
+      $expect[$ref_key],
+      $actual[$ref_key],
+    );
   }
 
-  final protected function adjustSchemata($unsafe) {
-    $lock = $this->lock();
+  final protected function adjustSchemata(
+    PhabricatorStorageManagementAPI $api,
+    $unsafe) {
+
+    $lock = $this->lock($api);
 
     try {
-      $err = $this->doAdjustSchemata($unsafe);
+      $err = $this->doAdjustSchemata($api, $unsafe);
     } catch (Exception $ex) {
       $lock->unlock();
       throw $ex;
     }
 
     $lock->unlock();
 
     return $err;
   }
 
-  final private function doAdjustSchemata($unsafe) {
+  final private function doAdjustSchemata(
+    PhabricatorStorageManagementAPI $api,
+    $unsafe) {
+
     $console = PhutilConsole::getConsole();
 
     $console->writeOut(
       "%s\n",
-      pht('Verifying database schemata...'));
+      pht(
+        'Verifying database schemata on "%s"...',
+        $api->getRef()->getRefKey()));
 
-    list($adjustments, $errors) = $this->findAdjustments();
-    $api = $this->getAPI();
+    list($adjustments, $errors) = $this->findAdjustments($api);
 
     if (!$adjustments) {
       $console->writeOut(
         "%s\n",
         pht('Found no adjustments for schemata.'));
 
       return $this->printErrors($errors, 0);
     }
 
     if (!$this->force && !$api->isCharacterSetAvailable('utf8mb4')) {
       $message = pht(
         "You have an old version of MySQL (older than 5.5) which does not ".
         "support the utf8mb4 character set. We strongly recomend upgrading to ".
         "5.5 or newer.\n\n".
         "If you apply adjustments now and later update MySQL to 5.5 or newer, ".
         "you'll need to apply adjustments again (and they will take a long ".
         "time).\n\n".
         "You can exit this workflow, update MySQL now, and then run this ".
         "workflow again. This is recommended, but may cause a lot of downtime ".
         "right now.\n\n".
         "You can exit this workflow, continue using Phabricator without ".
         "applying adjustments, update MySQL at a later date, and then run ".
         "this workflow again. This is also a good approach, and will let you ".
         "delay downtime until later.\n\n".
         "You can proceed with this workflow, and then optionally update ".
         "MySQL at a later date. After you do, you'll need to apply ".
         "adjustments again.\n\n".
         "For more information, see \"Managing Storage Adjustments\" in ".
         "the documentation.");
 
       $console->writeOut(
         "\n**<bg:yellow> %s </bg>**\n\n%s\n",
         pht('OLD MySQL VERSION'),
         phutil_console_wrap($message));
 
       $prompt = pht('Continue with old MySQL version?');
       if (!phutil_console_confirm($prompt, $default_no = true)) {
         return;
       }
     }
 
     $table = id(new PhutilConsoleTable())
       ->addColumn('database', array('title' => pht('Database')))
       ->addColumn('table', array('title' => pht('Table')))
       ->addColumn('name', array('title' => pht('Name')))
       ->addColumn('info', array('title' => pht('Issues')));
 
     foreach ($adjustments as $adjust) {
       $info = array();
       foreach ($adjust['issues'] as $issue) {
         $info[] = PhabricatorConfigStorageSchema::getIssueName($issue);
       }
 
       $table->addRow(array(
         'database' => $adjust['database'],
         'table' => idx($adjust, 'table'),
         'name' => idx($adjust, 'name'),
         'info' => implode(', ', $info),
       ));
     }
 
     $console->writeOut("\n\n");
 
     $table->draw();
 
     if ($this->dryRun) {
       $console->writeOut(
         "%s\n",
         pht('DRYRUN: Would apply adjustments.'));
       return 0;
     } else if ($this->didInitialize) {
       // If we just initialized the database, continue without prompting. This
       // is nicer for first-time setup and there's no reasonable reason any
       // user would ever answer "no" to the prompt against an empty schema.
     } else if (!$this->force) {
       $console->writeOut(
         "\n%s\n",
         pht(
           "Found %s adjustment(s) to apply, detailed above.\n\n".
           "You can review adjustments in more detail from the web interface, ".
           "in Config > Database Status. To better understand the adjustment ".
           "workflow, see \"Managing Storage Adjustments\" in the ".
           "documentation.\n\n".
           "MySQL needs to copy table data to make some adjustments, so these ".
           "migrations may take some time.",
           phutil_count($adjustments)));
 
       $prompt = pht('Apply these schema adjustments?');
       if (!phutil_console_confirm($prompt, $default_no = true)) {
         return 1;
       }
     }
 
     $console->writeOut(
       "%s\n",
       pht('Applying schema adjustments...'));
 
     $conn = $api->getConn(null);
 
     if ($unsafe) {
       queryfx($conn, 'SET SESSION sql_mode = %s', '');
     } else {
       queryfx($conn, 'SET SESSION sql_mode = %s', 'STRICT_ALL_TABLES');
     }
 
     $failed = array();
 
     // We make changes in several phases.
     $phases = array(
       // Drop surplus autoincrements. This allows us to drop primary keys on
       // autoincrement columns.
       'drop_auto',
 
       // Drop all keys we're going to adjust. This prevents them from
       // interfering with column changes.
       'drop_keys',
 
       // Apply all database, table, and column changes.
       'main',
 
       // Restore adjusted keys.
       'add_keys',
 
       // Add missing autoincrements.
       'add_auto',
     );
 
     $bar = id(new PhutilConsoleProgressBar())
       ->setTotal(count($adjustments) * count($phases));
 
     foreach ($phases as $phase) {
       foreach ($adjustments as $adjust) {
         try {
           switch ($adjust['kind']) {
             case 'database':
               if ($phase == 'main') {
                 queryfx(
                   $conn,
                   'ALTER DATABASE %T CHARACTER SET = %s COLLATE = %s',
                   $adjust['database'],
                   $adjust['charset'],
                   $adjust['collation']);
               }
               break;
             case 'table':
               if ($phase == 'main') {
                 queryfx(
                   $conn,
                   'ALTER TABLE %T.%T COLLATE = %s',
                   $adjust['database'],
                   $adjust['table'],
                   $adjust['collation']);
               }
               break;
             case 'column':
               $apply = false;
               $auto = false;
               $new_auto = idx($adjust, 'auto');
               if ($phase == 'drop_auto') {
                 if ($new_auto === false) {
                   $apply = true;
                   $auto = false;
                 }
               } else if ($phase == 'main') {
                 $apply = true;
                 if ($new_auto === false) {
                   $auto = false;
                 } else {
                   $auto = $adjust['is_auto'];
                 }
               } else if ($phase == 'add_auto') {
                 if ($new_auto === true) {
                   $apply = true;
                   $auto = true;
                 }
               }
 
               if ($apply) {
                 $parts = array();
 
                 if ($auto) {
                   $parts[] = qsprintf(
                     $conn,
                     'AUTO_INCREMENT');
                 }
 
                 if ($adjust['charset']) {
                   $parts[] = qsprintf(
                     $conn,
                     'CHARACTER SET %Q COLLATE %Q',
                     $adjust['charset'],
                     $adjust['collation']);
                 }
 
                 queryfx(
                   $conn,
                   'ALTER TABLE %T.%T MODIFY %T %Q %Q %Q',
                   $adjust['database'],
                   $adjust['table'],
                   $adjust['name'],
                   $adjust['type'],
                   implode(' ', $parts),
                   $adjust['nullable'] ? 'NULL' : 'NOT NULL');
               }
               break;
             case 'key':
               if (($phase == 'drop_keys') && $adjust['exists']) {
                 if ($adjust['name'] == 'PRIMARY') {
                   $key_name = 'PRIMARY KEY';
                 } else {
                   $key_name = qsprintf($conn, 'KEY %T', $adjust['name']);
                 }
 
                 queryfx(
                   $conn,
                   'ALTER TABLE %T.%T DROP %Q',
                   $adjust['database'],
                   $adjust['table'],
                   $key_name);
               }
 
               if (($phase == 'add_keys') && $adjust['keep']) {
                 // Different keys need different creation syntax. Notable
                 // special cases are primary keys and fulltext keys.
                 if ($adjust['name'] == 'PRIMARY') {
                   $key_name = 'PRIMARY KEY';
                 } else if ($adjust['indexType'] == 'FULLTEXT') {
                   $key_name = qsprintf($conn, 'FULLTEXT %T', $adjust['name']);
                 } else {
                   if ($adjust['unique']) {
                     $key_name = qsprintf(
                       $conn,
                       'UNIQUE KEY %T',
                       $adjust['name']);
                   } else {
                     $key_name = qsprintf(
                       $conn,
                       '/* NONUNIQUE */ KEY %T',
                       $adjust['name']);
                   }
                 }
 
                 queryfx(
                   $conn,
                   'ALTER TABLE %T.%T ADD %Q (%Q)',
                   $adjust['database'],
                   $adjust['table'],
                   $key_name,
                   implode(', ', $adjust['columns']));
               }
               break;
             default:
               throw new Exception(
                 pht('Unknown schema adjustment kind "%s"!', $adjust['kind']));
           }
         } catch (AphrontQueryException $ex) {
           $failed[] = array($adjust, $ex);
         }
         $bar->update(1);
       }
     }
     $bar->done();
 
     if (!$failed) {
       $console->writeOut(
         "%s\n",
         pht('Completed applying all schema adjustments.'));
 
       $err = 0;
     } else {
       $table = id(new PhutilConsoleTable())
         ->addColumn('target', array('title' => pht('Target')))
         ->addColumn('error', array('title' => pht('Error')));
 
       foreach ($failed as $failure) {
         list($adjust, $ex) = $failure;
 
         $pieces = array_select_keys(
           $adjust,
           array('database', 'table', 'name'));
         $pieces = array_filter($pieces);
         $target = implode('.', $pieces);
 
         $table->addRow(
           array(
             'target' => $target,
             'error' => $ex->getMessage(),
           ));
       }
 
       $console->writeOut("\n");
       $table->draw();
       $console->writeOut(
         "\n%s\n",
         pht('Failed to make some schema adjustments, detailed above.'));
       $console->writeOut(
         "%s\n",
         pht(
           'For help troubleshooting adjustments, see "Managing Storage '.
           'Adjustments" in the documentation.'));
 
       $err = 1;
     }
 
     return $this->printErrors($errors, $err);
   }
 
-  private function findAdjustments() {
-    list($comp, $expect, $actual) = $this->loadSchemata();
+  private function findAdjustments(
+    PhabricatorStorageManagementAPI $api) {
+    list($comp, $expect, $actual) = $this->loadSchemata($api);
 
     $issue_charset = PhabricatorConfigStorageSchema::ISSUE_CHARSET;
     $issue_collation = PhabricatorConfigStorageSchema::ISSUE_COLLATION;
     $issue_columntype = PhabricatorConfigStorageSchema::ISSUE_COLUMNTYPE;
     $issue_surpluskey = PhabricatorConfigStorageSchema::ISSUE_SURPLUSKEY;
     $issue_missingkey = PhabricatorConfigStorageSchema::ISSUE_MISSINGKEY;
     $issue_columns = PhabricatorConfigStorageSchema::ISSUE_KEYCOLUMNS;
     $issue_unique = PhabricatorConfigStorageSchema::ISSUE_UNIQUE;
     $issue_longkey = PhabricatorConfigStorageSchema::ISSUE_LONGKEY;
     $issue_auto = PhabricatorConfigStorageSchema::ISSUE_AUTOINCREMENT;
 
     $adjustments = array();
     $errors = array();
     foreach ($comp->getDatabases() as $database_name => $database) {
       foreach ($this->findErrors($database) as $issue) {
         $errors[] = array(
           'database' => $database_name,
           'issue' => $issue,
         );
       }
 
       $expect_database = $expect->getDatabase($database_name);
       $actual_database = $actual->getDatabase($database_name);
 
       if (!$expect_database || !$actual_database) {
         // If there's a real issue here, skip this stuff.
         continue;
       }
 
       if ($actual_database->getAccessDenied()) {
         // If we can't access the database, we can't access the tables either.
         continue;
       }
 
       $issues = array();
       if ($database->hasIssue($issue_charset)) {
         $issues[] = $issue_charset;
       }
       if ($database->hasIssue($issue_collation)) {
         $issues[] = $issue_collation;
       }
 
       if ($issues) {
         $adjustments[] = array(
           'kind' => 'database',
           'database' => $database_name,
           'issues' => $issues,
           'charset' => $expect_database->getCharacterSet(),
           'collation' => $expect_database->getCollation(),
         );
       }
 
       foreach ($database->getTables() as $table_name => $table) {
         foreach ($this->findErrors($table) as $issue) {
           $errors[] = array(
             'database' => $database_name,
             'table' => $table_name,
             'issue' => $issue,
           );
         }
 
         $expect_table = $expect_database->getTable($table_name);
         $actual_table = $actual_database->getTable($table_name);
 
         if (!$expect_table || !$actual_table) {
           continue;
         }
 
         $issues = array();
         if ($table->hasIssue($issue_collation)) {
           $issues[] = $issue_collation;
         }
 
         if ($issues) {
           $adjustments[] = array(
             'kind' => 'table',
             'database' => $database_name,
             'table' => $table_name,
             'issues' => $issues,
             'collation' => $expect_table->getCollation(),
           );
         }
 
         foreach ($table->getColumns() as $column_name => $column) {
           foreach ($this->findErrors($column) as $issue) {
             $errors[] = array(
               'database' => $database_name,
               'table' => $table_name,
               'name' => $column_name,
               'issue' => $issue,
             );
           }
 
           $expect_column = $expect_table->getColumn($column_name);
           $actual_column = $actual_table->getColumn($column_name);
 
           if (!$expect_column || !$actual_column) {
             continue;
           }
 
           $issues = array();
           if ($column->hasIssue($issue_collation)) {
             $issues[] = $issue_collation;
           }
           if ($column->hasIssue($issue_charset)) {
             $issues[] = $issue_charset;
           }
           if ($column->hasIssue($issue_columntype)) {
             $issues[] = $issue_columntype;
           }
           if ($column->hasIssue($issue_auto)) {
             $issues[] = $issue_auto;
           }
 
           if ($issues) {
             if ($expect_column->getCharacterSet() === null) {
               // For non-text columns, we won't be specifying a collation or
               // character set.
               $charset = null;
               $collation = null;
             } else {
               $charset = $expect_column->getCharacterSet();
               $collation = $expect_column->getCollation();
             }
 
             $adjustment = array(
               'kind' => 'column',
               'database' => $database_name,
               'table' => $table_name,
               'name' => $column_name,
               'issues' => $issues,
               'collation' => $collation,
               'charset' => $charset,
               'type' => $expect_column->getColumnType(),
 
               // NOTE: We don't adjust column nullability because it is
               // dangerous, so always use the current nullability.
               'nullable' => $actual_column->getNullable(),
 
               // NOTE: This always stores the current value, because we have
               // to make these updates separately.
               'is_auto' => $actual_column->getAutoIncrement(),
             );
 
             if ($column->hasIssue($issue_auto)) {
               $adjustment['auto'] = $expect_column->getAutoIncrement();
             }
 
             $adjustments[] = $adjustment;
           }
         }
 
         foreach ($table->getKeys() as $key_name => $key) {
           foreach ($this->findErrors($key) as $issue) {
             $errors[] = array(
               'database' => $database_name,
               'table' => $table_name,
               'name' => $key_name,
               'issue' => $issue,
             );
           }
 
           $expect_key = $expect_table->getKey($key_name);
           $actual_key = $actual_table->getKey($key_name);
 
           $issues = array();
           $keep_key = true;
           if ($key->hasIssue($issue_surpluskey)) {
             $issues[] = $issue_surpluskey;
             $keep_key = false;
           }
 
           if ($key->hasIssue($issue_missingkey)) {
             $issues[] = $issue_missingkey;
           }
 
           if ($key->hasIssue($issue_columns)) {
             $issues[] = $issue_columns;
           }
 
           if ($key->hasIssue($issue_unique)) {
             $issues[] = $issue_unique;
           }
 
           // NOTE: We can't really fix this, per se, but we may need to remove
           // the key to change the column type. In the best case, the new
           // column type won't be overlong and recreating the key really will
           // fix the issue. In the worst case, we get the right column type and
           // lose the key, which is still better than retaining the key having
           // the wrong column type.
           if ($key->hasIssue($issue_longkey)) {
             $issues[] = $issue_longkey;
           }
 
           if ($issues) {
             $adjustment = array(
               'kind' => 'key',
               'database' => $database_name,
               'table' => $table_name,
               'name' => $key_name,
               'issues' => $issues,
               'exists' => (bool)$actual_key,
               'keep' => $keep_key,
             );
 
             if ($keep_key) {
               $adjustment += array(
                 'columns' => $expect_key->getColumnNames(),
                 'unique' => $expect_key->getUnique(),
                 'indexType' => $expect_key->getIndexType(),
               );
             }
 
             $adjustments[] = $adjustment;
           }
         }
       }
     }
 
     return array($adjustments, $errors);
   }
 
   private function findErrors(PhabricatorConfigStorageSchema $schema) {
     $result = array();
     foreach ($schema->getLocalIssues() as $issue) {
       $status = PhabricatorConfigStorageSchema::getIssueStatus($issue);
       if ($status == PhabricatorConfigStorageSchema::STATUS_FAIL) {
         $result[] = $issue;
       }
     }
     return $result;
   }
 
   private function printErrors(array $errors, $default_return) {
     if (!$errors) {
       return $default_return;
     }
 
     $console = PhutilConsole::getConsole();
 
     $table = id(new PhutilConsoleTable())
       ->addColumn('target', array('title' => pht('Target')))
       ->addColumn('error', array('title' => pht('Error')));
 
     $any_surplus = false;
     $all_surplus = true;
     $any_access = false;
     $all_access = true;
     foreach ($errors as $error) {
       $pieces = array_select_keys(
         $error,
         array('database', 'table', 'name'));
       $pieces = array_filter($pieces);
       $target = implode('.', $pieces);
 
       $name = PhabricatorConfigStorageSchema::getIssueName($error['issue']);
 
       $issue = $error['issue'];
 
       if ($issue === PhabricatorConfigStorageSchema::ISSUE_SURPLUS) {
         $any_surplus = true;
       } else {
         $all_surplus = false;
       }
 
       if ($issue === PhabricatorConfigStorageSchema::ISSUE_ACCESSDENIED) {
         $any_access = true;
       } else {
         $all_access = false;
       }
 
       $table->addRow(
         array(
           'target' => $target,
           'error' => $name,
         ));
     }
 
     $console->writeOut("\n");
     $table->draw();
     $console->writeOut("\n");
 
 
     $message = array();
     if ($all_surplus) {
       $message[] = pht(
         'You have surplus schemata (extra tables or columns which Phabricator '.
         'does not expect). For information on resolving these '.
         'issues, see the "Surplus Schemata" section in the "Managing Storage '.
         'Adjustments" article in the documentation.');
     } else if ($all_access) {
       $message[] = pht(
         'The user you are connecting to MySQL with does not have the correct '.
         'permissions, and can not access some databases or tables that it '.
         'needs to be able to access. GRANT the user additional permissions.');
     } else {
       $message[] = pht(
         'The schemata have errors (detailed above) which the adjustment '.
         'workflow can not fix.');
 
       if ($any_access) {
         $message[] = pht(
           'Some of these errors are caused by access control problems. '.
           'The user you are connecting with does not have permission to see '.
           'all of the database or tables that Phabricator uses. You need to '.
           'GRANT the user more permission, or use a different user.');
       }
 
       if ($any_surplus) {
         $message[] = pht(
           'Some of these errors are caused by surplus schemata (extra '.
           'tables or columns which Phabricator does not expect). These are '.
           'not serious. For information on resolving these issues, see the '.
           '"Surplus Schemata" section in the "Managing Storage Adjustments" '.
           'article in the documentation.');
       }
 
       $message[] = pht(
         'If you are not developing Phabricator itself, report this issue to '.
         'the upstream.');
 
       $message[] = pht(
         'If you are developing Phabricator, these errors usually indicate '.
         'that your schema specifications do not agree with the schemata your '.
         'code actually builds.');
     }
     $message = implode("\n\n", $message);
 
     if ($all_surplus) {
       $console->writeOut(
         "**<bg:yellow> %s </bg>**\n\n%s\n",
         pht('SURPLUS SCHEMATA'),
         phutil_console_wrap($message));
     } else if ($all_access) {
       $console->writeOut(
         "**<bg:yellow> %s </bg>**\n\n%s\n",
         pht('ACCESS DENIED'),
         phutil_console_wrap($message));
     } else {
       $console->writeOut(
         "**<bg:red> %s </bg>**\n\n%s\n",
         pht('SCHEMATA ERRORS'),
         phutil_console_wrap($message));
     }
 
     return 2;
   }
 
   final protected function upgradeSchemata(
+    PhabricatorStorageManagementAPI $api,
     $apply_only = null,
     $no_quickstart = false,
     $init_only = false) {
 
-    $lock = $this->lock();
+    $lock = $this->lock($api);
 
     try {
-      $this->doUpgradeSchemata($apply_only, $no_quickstart, $init_only);
+      $this->doUpgradeSchemata($api, $apply_only, $no_quickstart, $init_only);
     } catch (Exception $ex) {
       $lock->unlock();
       throw $ex;
     }
 
     $lock->unlock();
   }
 
   final private function doUpgradeSchemata(
+    PhabricatorStorageManagementAPI $api,
     $apply_only,
     $no_quickstart,
     $init_only) {
 
-    $api = $this->getAPI();
-
-    $applied = $this->getApi()->getAppliedPatches();
+    $applied = $api->getAppliedPatches();
     if ($applied === null) {
       if ($this->dryRun) {
         echo pht(
           "DRYRUN: Patch metadata storage doesn't exist yet, ".
           "it would be created.\n");
         return 0;
       }
 
       if ($apply_only) {
         throw new PhutilArgumentUsageException(
           pht(
             'Storage has not been initialized yet, you must initialize '.
             'storage before selectively applying patches.'));
         return 1;
       }
 
       // If we're initializing storage for the first time, track it so that
       // we can give the user a nicer experience during the subsequent
       // adjustment phase.
       $this->didInitialize = true;
 
       $legacy = $api->getLegacyPatches($this->patches);
       if ($legacy || $no_quickstart || $init_only) {
 
         // If we have legacy patches, we can't quickstart.
 
         $api->createDatabase('meta_data');
         $api->createTable(
           'meta_data',
           'patch_status',
           array(
             'patch VARCHAR(255) NOT NULL PRIMARY KEY COLLATE utf8_general_ci',
             'applied INT UNSIGNED NOT NULL',
           ));
 
         foreach ($legacy as $patch) {
           $api->markPatchApplied($patch);
         }
       } else {
         echo pht('Loading quickstart template...')."\n";
         $root = dirname(phutil_get_library_root('phabricator'));
         $sql  = $root.'/resources/sql/quickstart.sql';
         $api->applyPatchSQL($sql);
       }
     }
 
     if ($init_only) {
       echo pht('Storage initialized.')."\n";
       return 0;
     }
 
     $applied = $api->getAppliedPatches();
     $applied = array_fuse($applied);
 
     $skip_mark = false;
     if ($apply_only) {
       if (isset($applied[$apply_only])) {
 
         unset($applied[$apply_only]);
         $skip_mark = true;
 
         if (!$this->force && !$this->dryRun) {
           echo phutil_console_wrap(
             pht(
               "Patch '%s' has already been applied. Are you sure you want ".
               "to apply it again? This may put your storage in a state ".
               "that the upgrade scripts can not automatically manage.",
               $apply_only));
           if (!phutil_console_confirm(pht('Apply patch again?'))) {
             echo pht('Cancelled.')."\n";
             return 1;
           }
         }
       }
     }
 
     while (true) {
       $applied_something = false;
       foreach ($this->patches as $key => $patch) {
         if (isset($applied[$key])) {
           unset($this->patches[$key]);
           continue;
         }
 
         if ($apply_only && $apply_only != $key) {
           unset($this->patches[$key]);
           continue;
         }
 
         $can_apply = true;
         foreach ($patch->getAfter() as $after) {
           if (empty($applied[$after])) {
             if ($apply_only) {
               echo pht(
                 "Unable to apply patch '%s' because it depends ".
                 "on patch '%s', which has not been applied.\n",
                 $apply_only,
                 $after);
               return 1;
             }
             $can_apply = false;
             break;
           }
         }
 
         if (!$can_apply) {
           continue;
         }
 
         $applied_something = true;
 
         if ($this->dryRun) {
           echo pht("DRYRUN: Would apply patch '%s'.", $key)."\n";
         } else {
           echo pht("Applying patch '%s'...", $key)."\n";
 
           $t_begin = microtime(true);
           $api->applyPatch($patch);
           $t_end = microtime(true);
 
           if (!$skip_mark) {
             $api->markPatchApplied($key, ($t_end - $t_begin));
           }
         }
 
         unset($this->patches[$key]);
         $applied[$key] = true;
       }
 
       if (!$applied_something) {
         if (count($this->patches)) {
           throw new Exception(
             pht(
-              'Some patches could not be applied: %s',
+              'Some patches could not be applied to "%s": %s',
+              $api->getRef()->getRefKey(),
               implode(', ', array_keys($this->patches))));
         } else if (!$this->dryRun && !$apply_only) {
           echo pht(
-            "Storage is up to date. Use '%s' for details.",
+            'Storage is up to date on "%s". Use "%s" for details.',
+            $api->getRef()->getRefKey(),
             'storage status')."\n";
         }
         break;
       }
     }
   }
 
   final protected function getBareHostAndPort($host) {
     // Split out port information, since the command-line client requires a
     // separate flag for the port.
     $uri = new PhutilURI('mysql://'.$host);
     if ($uri->getPort()) {
       $port = $uri->getPort();
       $bare_hostname = $uri->getDomain();
     } else {
       $port = null;
       $bare_hostname = $host;
     }
 
     return array($bare_hostname, $port);
   }
 
   /**
    * Acquires a @{class:PhabricatorGlobalLock}.
    *
    * @return PhabricatorGlobalLock
    */
-  final protected function lock() {
+  final protected function lock(PhabricatorStorageManagementAPI $api) {
     return PhabricatorGlobalLock::newLock(__CLASS__)
-      ->useSpecificConnection($this->getApi()->getConn(null))
+      ->useSpecificConnection($api->getConn(null))
       ->lock();
   }
 
 }