@recaptime-dev's working patches + fork for Phorge, a community fork of Phabricator. (Upstream dev and stable branches are at upstream/main and upstream/stable respectively.)
hq.recaptime.dev/wiki/Phorge
phorge
phabricator
Version clustered, observed repositories in a reasonable way (by largest discovered HEAD)
Summary:
Ref T4292. For hosted, clustered repositories we have a good way to increment the internal version of the repository: every time a user pushes something, we increment the version by 1.
We don't have a great way to do this for observed/remote repositories because when we `git fetch` we might get nothing, or we might get some changes, and we can't easily tell //what// changes we got.
For example, if we see that another node is at "version 97", and we do a fetch and see some changes, we don't know if we're in sync with them (i.e., also at "version 97") or ahead of them (at "version 98").
This implements a simple way to version an observed repository:
- Take the head of every branch/tag.
- Look them up.
- Pick the biggest internal ID number.
This will work //except// when branches are deleted, which could cause the version to go backward if the "biggest commit" is the one that was deleted. This should be OK, since it's rare and the effects are minor and the repository will "self-heal" on the next actual push.
Test Plan:
- Created an observed repository.
- Ran `bin/repository update` and observed a sensible version number appear in the version table.
- Pushed to the remote, did another update, saw a sensible update.
- Did an update with no push, saw no effect on version number.
- Toggled repository to hosted, saw the version reset.
- Simulated read traffic to out-of-sync node, saw it do a remote fetch.
Reviewers: chad
Reviewed By: chad
Maniphest Tasks: T4292
Differential Revision: https://secure.phabricator.com/D15986
+316
-70
+4
-1
src/applications/diffusion/conduit/DiffusionBranchQueryConduitAPIMethod.php
+4
-1
src/applications/diffusion/conduit/DiffusionBranchQueryConduitAPIMethod.php
+4
-1
src/applications/diffusion/conduit/DiffusionTagsQueryConduitAPIMethod.php
+4
-1
src/applications/diffusion/conduit/DiffusionTagsQueryConduitAPIMethod.php
+13
src/applications/diffusion/editor/DiffusionURIEditor.php
+13
src/applications/diffusion/editor/DiffusionURIEditor.php
···
463
463
break;
464
464
}
465
465
466
+
$was_hosted = $repository->isHosted();
467
+
466
468
if ($observe_uri) {
467
469
$repository
468
470
->setHosted(false)
···
476
478
}
477
479
478
480
$repository->save();
481
+
482
+
$is_hosted = $repository->isHosted();
483
+
484
+
// If we've swapped the repository from hosted to observed or vice versa,
485
+
// reset all the cluster version clocks.
486
+
if ($was_hosted != $is_hosted) {
487
+
$cluster_engine = id(new DiffusionRepositoryClusterEngine())
488
+
->setViewer($this->getActor())
489
+
->setRepository($repository)
490
+
->synchronizeWorkingCopyAfterHostingChange();
491
+
}
479
492
480
493
return $xactions;
481
494
}
+22
-6
src/applications/diffusion/management/DiffusionRepositoryStorageManagementPanel.php
+22
-6
src/applications/diffusion/management/DiffusionRepositoryStorageManagementPanel.php
···
137
137
$version = idx($versions, $device->getPHID());
138
138
if ($version) {
139
139
$version_number = $version->getRepositoryVersion();
140
-
$version_number = phutil_tag(
141
-
'a',
142
-
array(
143
-
'href' => "/diffusion/pushlog/view/{$version_number}/",
144
-
),
145
-
$version_number);
140
+
141
+
$href = null;
142
+
if ($repository->isHosted()) {
143
+
$href = "/diffusion/pushlog/view/{$version_number}/";
144
+
} else {
145
+
$commit = id(new DiffusionCommitQuery())
146
+
->setViewer($viewer)
147
+
->withIDs(array($version_number))
148
+
->executeOne();
149
+
if ($commit) {
150
+
$href = $commit->getURI();
151
+
}
152
+
}
153
+
154
+
if ($href) {
155
+
$version_number = phutil_tag(
156
+
'a',
157
+
array(
158
+
'href' => $href,
159
+
),
160
+
$version_number);
161
+
}
146
162
} else {
147
163
$version_number = '-';
148
164
}
+163
-24
src/applications/diffusion/protocol/DiffusionRepositoryClusterEngine.php
+163
-24
src/applications/diffusion/protocol/DiffusionRepositoryClusterEngine.php
···
85
85
/**
86
86
* @task sync
87
87
*/
88
+
public function synchronizeWorkingCopyAfterHostingChange() {
89
+
if (!$this->shouldEnableSynchronization()) {
90
+
return;
91
+
}
92
+
93
+
$repository = $this->getRepository();
94
+
$repository_phid = $repository->getPHID();
95
+
96
+
$versions = PhabricatorRepositoryWorkingCopyVersion::loadVersions(
97
+
$repository_phid);
98
+
$versions = mpull($versions, null, 'getDevicePHID');
99
+
100
+
// After converting a hosted repository to observed, or vice versa, we
101
+
// need to reset version numbers because the clocks for observed and hosted
102
+
// repositories run on different units.
103
+
104
+
// We identify all the cluster leaders and reset their version to 0.
105
+
// We identify all the cluster followers and demote them.
106
+
107
+
// This allows the cluter to start over again at version 0 but keep the
108
+
// same leaders.
109
+
110
+
if ($versions) {
111
+
$max_version = (int)max(mpull($versions, 'getRepositoryVersion'));
112
+
foreach ($versions as $version) {
113
+
$device_phid = $version->getDevicePHID();
114
+
115
+
if ($version->getRepositoryVersion() == $max_version) {
116
+
PhabricatorRepositoryWorkingCopyVersion::updateVersion(
117
+
$repository_phid,
118
+
$device_phid,
119
+
0);
120
+
} else {
121
+
PhabricatorRepositoryWorkingCopyVersion::demoteDevice(
122
+
$repository_phid,
123
+
$device_phid);
124
+
}
125
+
}
126
+
}
127
+
128
+
return $this;
129
+
}
130
+
131
+
132
+
/**
133
+
* @task sync
134
+
*/
88
135
public function synchronizeWorkingCopyBeforeRead() {
89
136
if (!$this->shouldEnableSynchronization()) {
90
137
return;
···
149
196
150
197
$max_version = (int)max(mpull($versions, 'getRepositoryVersion'));
151
198
if ($max_version > $this_version) {
152
-
$fetchable = array();
153
-
foreach ($versions as $version) {
154
-
if ($version->getRepositoryVersion() == $max_version) {
155
-
$fetchable[] = $version->getDevicePHID();
199
+
if ($repository->isHosted()) {
200
+
$fetchable = array();
201
+
foreach ($versions as $version) {
202
+
if ($version->getRepositoryVersion() == $max_version) {
203
+
$fetchable[] = $version->getDevicePHID();
204
+
}
156
205
}
157
-
}
158
206
159
-
$this->synchronizeWorkingCopyFromDevices($fetchable);
207
+
$this->synchronizeWorkingCopyFromDevices($fetchable);
208
+
} else {
209
+
$this->synchornizeWorkingCopyFromRemote();
210
+
}
160
211
161
212
PhabricatorRepositoryWorkingCopyVersion::updateVersion(
162
213
$repository_phid,
···
329
380
}
330
381
331
382
383
+
public function synchronizeWorkingCopyAfterDiscovery($new_version) {
384
+
if (!$this->shouldEnableSynchronization()) {
385
+
return;
386
+
}
387
+
388
+
$repository = $this->getRepository();
389
+
$repository_phid = $repository->getPHID();
390
+
if ($repository->isHosted()) {
391
+
return;
392
+
}
393
+
394
+
$viewer = $this->getViewer();
395
+
396
+
$device = AlmanacKeys::getLiveDevice();
397
+
$device_phid = $device->getPHID();
398
+
399
+
// NOTE: We are not holding a lock here because this method is only called
400
+
// from PhabricatorRepositoryDiscoveryEngine, which already holds a device
401
+
// lock. Even if we do race here and record an older version, the
402
+
// consequences are mild: we only do extra work to correct it later.
403
+
404
+
$versions = PhabricatorRepositoryWorkingCopyVersion::loadVersions(
405
+
$repository_phid);
406
+
$versions = mpull($versions, null, 'getDevicePHID');
407
+
408
+
$this_version = idx($versions, $device_phid);
409
+
if ($this_version) {
410
+
$this_version = (int)$this_version->getRepositoryVersion();
411
+
} else {
412
+
$this_version = -1;
413
+
}
414
+
415
+
if ($new_version > $this_version) {
416
+
PhabricatorRepositoryWorkingCopyVersion::updateVersion(
417
+
$repository_phid,
418
+
$device_phid,
419
+
$new_version);
420
+
}
421
+
}
422
+
423
+
332
424
/**
333
425
* @task sync
334
426
*/
···
471
563
return false;
472
564
}
473
565
474
-
// TODO: It may eventually make sense to try to version and synchronize
475
-
// observed repositories (so that daemons don't do reads against out-of
476
-
// date hosts), but don't bother for now.
477
-
if (!$repository->isHosted()) {
566
+
$device = AlmanacKeys::getLiveDevice();
567
+
if (!$device) {
478
568
return false;
479
569
}
480
570
571
+
return true;
572
+
}
573
+
574
+
575
+
/**
576
+
* @task internal
577
+
*/
578
+
private function synchornizeWorkingCopyFromRemote() {
579
+
$repository = $this->getRepository();
481
580
$device = AlmanacKeys::getLiveDevice();
482
-
if (!$device) {
483
-
return false;
581
+
582
+
$local_path = $repository->getLocalPath();
583
+
$fetch_uri = $repository->getRemoteURIEnvelope();
584
+
585
+
if ($repository->isGit()) {
586
+
$this->requireWorkingCopy();
587
+
588
+
$argv = array(
589
+
'fetch --prune -- %P %s',
590
+
$fetch_uri,
591
+
'+refs/*:refs/*',
592
+
);
593
+
} else {
594
+
throw new Exception(pht('Remote sync only supported for git!'));
484
595
}
485
596
486
-
return true;
597
+
$future = DiffusionCommandEngine::newCommandEngine($repository)
598
+
->setArgv($argv)
599
+
->setSudoAsDaemon(true)
600
+
->setCredentialPHID($repository->getCredentialPHID())
601
+
->setProtocol($repository->getRemoteProtocol())
602
+
->newFuture();
603
+
604
+
$future->setCWD($local_path);
605
+
606
+
try {
607
+
$future->resolvex();
608
+
} catch (Exception $ex) {
609
+
$this->logLine(
610
+
pht(
611
+
'Synchronization of "%s" from remote failed: %s',
612
+
$device->getName(),
613
+
$ex->getMessage()));
614
+
throw $ex;
615
+
}
487
616
}
488
617
489
618
···
560
689
$local_path = $repository->getLocalPath();
561
690
562
691
if ($repository->isGit()) {
563
-
if (!Filesystem::pathExists($local_path)) {
564
-
throw new Exception(
565
-
pht(
566
-
'Repository "%s" does not have a working copy on this device '.
567
-
'yet, so it can not be synchronized. Wait for the daemons to '.
568
-
'construct one or run `bin/repository update %s` on this host '.
569
-
'("%s") to build it explicitly.',
570
-
$repository->getDisplayName(),
571
-
$repository->getMonogram(),
572
-
$device->getName()));
573
-
}
692
+
$this->requireWorkingCopy();
574
693
575
694
$argv = array(
576
695
'fetch --prune -- %s %s',
···
622
741
}
623
742
return $this;
624
743
}
744
+
745
+
private function requireWorkingCopy() {
746
+
$repository = $this->getRepository();
747
+
$local_path = $repository->getLocalPath();
748
+
749
+
if (!Filesystem::pathExists($local_path)) {
750
+
$device = AlmanacKeys::getLiveDevice();
751
+
752
+
throw new Exception(
753
+
pht(
754
+
'Repository "%s" does not have a working copy on this device '.
755
+
'yet, so it can not be synchronized. Wait for the daemons to '.
756
+
'construct one or run `bin/repository update %s` on this host '.
757
+
'("%s") to build it explicitly.',
758
+
$repository->getDisplayName(),
759
+
$repository->getMonogram(),
760
+
$device->getName()));
761
+
}
762
+
}
763
+
625
764
}
+19
-16
src/applications/diffusion/query/lowlevel/DiffusionLowLevelGitRefQuery.php
+19
-16
src/applications/diffusion/query/lowlevel/DiffusionLowLevelGitRefQuery.php
···
6
6
*/
7
7
final class DiffusionLowLevelGitRefQuery extends DiffusionLowLevelQuery {
8
8
9
-
private $isTag;
10
-
private $isOriginBranch;
9
+
private $refTypes;
11
10
12
-
public function withIsTag($is_tag) {
13
-
$this->isTag = $is_tag;
11
+
public function withRefTypes(array $ref_types) {
12
+
$this->refTypes = $ref_types;
14
13
return $this;
15
14
}
16
15
17
-
public function withIsOriginBranch($is_origin_branch) {
18
-
$this->isOriginBranch = $is_origin_branch;
19
-
return $this;
20
-
}
16
+
protected function executeQuery() {
17
+
$ref_types = $this->refTypes;
18
+
if ($ref_types) {
19
+
$type_branch = PhabricatorRepositoryRefCursor::TYPE_BRANCH;
20
+
$type_tag = PhabricatorRepositoryRefCursor::TYPE_TAG;
21
+
22
+
$ref_types = array_fuse($ref_types);
23
+
24
+
$with_branches = isset($ref_types[$type_branch]);
25
+
$with_tags = isset($ref_types[$type_tag]);
26
+
} else {
27
+
$with_branches = true;
28
+
$with_tags = true;
29
+
}
21
30
22
-
protected function executeQuery() {
23
31
$repository = $this->getRepository();
24
32
25
33
$prefixes = array();
26
34
27
-
$any = ($this->isTag || $this->isOriginBranch);
28
-
if (!$any) {
29
-
throw new Exception(pht('Specify types of refs to query.'));
30
-
}
31
-
32
-
if ($this->isOriginBranch) {
35
+
if ($with_branches) {
33
36
if ($repository->isWorkingCopyBare()) {
34
37
$prefix = 'refs/heads/';
35
38
} else {
···
39
42
$prefixes[] = $prefix;
40
43
}
41
44
42
-
if ($this->isTag) {
45
+
if ($with_tags) {
43
46
$prefixes[] = 'refs/tags/';
44
47
}
45
48
+5
-2
src/applications/diffusion/query/lowlevel/DiffusionLowLevelResolveRefsQuery.php
+5
-2
src/applications/diffusion/query/lowlevel/DiffusionLowLevelResolveRefsQuery.php
···
66
66
// First, resolve branches and tags.
67
67
$ref_map = id(new DiffusionLowLevelGitRefQuery())
68
68
->setRepository($repository)
69
-
->withIsTag(true)
70
-
->withIsOriginBranch(true)
69
+
->withRefTypes(
70
+
array(
71
+
PhabricatorRepositoryRefCursor::TYPE_BRANCH,
72
+
PhabricatorRepositoryRefCursor::TYPE_TAG,
73
+
))
71
74
->execute();
72
75
$ref_map = mgroup($ref_map, 'getShortName');
73
76
+61
-1
src/applications/repository/engine/PhabricatorRepositoryDiscoveryEngine.php
+61
-1
src/applications/repository/engine/PhabricatorRepositoryDiscoveryEngine.php
···
63
63
64
64
private function discoverCommitsWithLock() {
65
65
$repository = $this->getRepository();
66
+
$viewer = $this->getViewer();
66
67
67
68
$vcs = $repository->getVersionControlSystem();
68
69
switch ($vcs) {
···
104
105
$this->commitCache[$ref->getIdentifier()] = true;
105
106
}
106
107
108
+
$version = $this->getObservedVersion($repository);
109
+
if ($version !== null) {
110
+
id(new DiffusionRepositoryClusterEngine())
111
+
->setViewer($viewer)
112
+
->setRepository($repository)
113
+
->synchronizeWorkingCopyAfterDiscovery($version);
114
+
}
115
+
107
116
return $refs;
108
117
}
109
118
···
121
130
$this->verifyGitOrigin($repository);
122
131
}
123
132
133
+
// TODO: This should also import tags, but some of the logic is still
134
+
// branch-specific today.
135
+
124
136
$branches = id(new DiffusionLowLevelGitRefQuery())
125
137
->setRepository($repository)
126
-
->withIsOriginBranch(true)
138
+
->withRefTypes(
139
+
array(
140
+
PhabricatorRepositoryRefCursor::TYPE_BRANCH,
141
+
))
127
142
->execute();
128
143
129
144
if (!$branches) {
···
640
655
}
641
656
642
657
return true;
658
+
}
659
+
660
+
661
+
private function getObservedVersion(PhabricatorRepository $repository) {
662
+
if ($repository->isHosted()) {
663
+
return null;
664
+
}
665
+
666
+
if ($repository->isGit()) {
667
+
return $this->getGitObservedVersion($repository);
668
+
}
669
+
670
+
return null;
671
+
}
672
+
673
+
private function getGitObservedVersion(PhabricatorRepository $repository) {
674
+
$refs = id(new DiffusionLowLevelGitRefQuery())
675
+
->setRepository($repository)
676
+
->execute();
677
+
if (!$refs) {
678
+
return null;
679
+
}
680
+
681
+
// In Git, the observed version is the most recently discovered commit
682
+
// at any repository HEAD. It's possible for this to regress temporarily
683
+
// if a branch is pushed and then deleted. This is acceptable because it
684
+
// doesn't do anything meaningfully bad and will fix itself on the next
685
+
// push.
686
+
687
+
$ref_identifiers = mpull($refs, 'getCommitIdentifier');
688
+
$ref_identifiers = array_fuse($ref_identifiers);
689
+
690
+
$version = queryfx_one(
691
+
$repository->establishConnection('w'),
692
+
'SELECT MAX(id) version FROM %T WHERE repositoryID = %d
693
+
AND commitIdentifier IN (%Ls)',
694
+
id(new PhabricatorRepositoryCommit())->getTableName(),
695
+
$repository->getID(),
696
+
$ref_identifiers);
697
+
698
+
if (!$version) {
699
+
return null;
700
+
}
701
+
702
+
return (int)$version['version'];
643
703
}
644
704
645
705
}
+17
-17
src/applications/repository/engine/PhabricatorRepositoryPullEngine.php
+17
-17
src/applications/repository/engine/PhabricatorRepositoryPullEngine.php
···
108
108
} else {
109
109
$this->executeSubversionCreate();
110
110
}
111
-
} else {
112
-
if (!$repository->isHosted()) {
113
-
$this->logPull(
114
-
pht(
115
-
'Updating the working copy for repository "%s".',
116
-
$repository->getDisplayName()));
117
-
if ($is_git) {
118
-
$this->verifyGitOrigin($repository);
119
-
$this->executeGitUpdate();
120
-
} else if ($is_hg) {
121
-
$this->executeMercurialUpdate();
122
-
}
111
+
}
112
+
113
+
id(new DiffusionRepositoryClusterEngine())
114
+
->setViewer($viewer)
115
+
->setRepository($repository)
116
+
->synchronizeWorkingCopyBeforeRead();
117
+
118
+
if (!$repository->isHosted()) {
119
+
$this->logPull(
120
+
pht(
121
+
'Updating the working copy for repository "%s".',
122
+
$repository->getDisplayName()));
123
+
if ($is_git) {
124
+
$this->verifyGitOrigin($repository);
125
+
$this->executeGitUpdate();
126
+
} else if ($is_hg) {
127
+
$this->executeMercurialUpdate();
123
128
}
124
129
}
125
130
126
131
if ($repository->isHosted()) {
127
-
id(new DiffusionRepositoryClusterEngine())
128
-
->setViewer($viewer)
129
-
->setRepository($repository)
130
-
->synchronizeWorkingCopyBeforeRead();
131
-
132
132
if ($is_git) {
133
133
$this->installGitHook();
134
134
} else if ($is_svn) {
+8
-2
src/applications/repository/engine/PhabricatorRepositoryRefEngine.php
+8
-2
src/applications/repository/engine/PhabricatorRepositoryRefEngine.php
···
452
452
private function loadGitBranchPositions(PhabricatorRepository $repository) {
453
453
return id(new DiffusionLowLevelGitRefQuery())
454
454
->setRepository($repository)
455
-
->withIsOriginBranch(true)
455
+
->withRefTypes(
456
+
array(
457
+
PhabricatorRepositoryRefCursor::TYPE_BRANCH,
458
+
))
456
459
->execute();
457
460
}
458
461
···
463
466
private function loadGitTagPositions(PhabricatorRepository $repository) {
464
467
return id(new DiffusionLowLevelGitRefQuery())
465
468
->setRepository($repository)
466
-
->withIsTag(true)
469
+
->withRefTypes(
470
+
array(
471
+
PhabricatorRepositoryRefCursor::TYPE_TAG,
472
+
))
467
473
->execute();
468
474
}
469
475