feat(scm): detect cross-fork PRs by scanning all project remotes (#2368)
The SCM observer only polled a project's push origin for open PRs, so a PR opened from the fork against an upstream base repo (the standard fork -> upstream flow) was never discovered: GitHub lists a PR under its base repo, not its head, so an origin-only scan can't see it. Sessions that opened such PRs showed no PR on the dashboard. Scan every GitHub remote in the project checkout (origin + upstreams / mirrors) for open PRs, and attribute a PR to a session only when its head branch lives in that session's push origin. This surfaces cross-fork PRs while preserving the no-misattribution guarantee: a stranger's fork PR has a head repo no session owns and is dropped. Tracked cross-fork PRs are keyed for refresh by their own recorded repo (the upstream base) so the GraphQL refetch targets the right repo.
This commit is contained in:
Harshit Singh Bhandari
GitHub
parent
3aa8e044b2
commit
a32d507eb8
|
|
@ -191,12 +191,18 @@ type subject struct {
|
|||
hasPR bool
|
||||
}
|
||||
|
||||
// sessionRepo pairs a live session with its parsed repo and branch for per-repo
|
||||
// branch-prefix discovery of new (including stacked) pull requests.
|
||||
// sessionRepo pairs a live session with a repo to scan and its branch for
|
||||
// per-repo branch-prefix discovery of new (including stacked) pull requests.
|
||||
// A session is scanned against its push origin plus every other remote in the
|
||||
// project checkout, so repo is the repo whose open-PR list is listed while
|
||||
// headRepo is the repo the session's head branch actually lives in (the push
|
||||
// origin). For same-repo PRs repo == headRepo; for a cross-fork PR (fork head,
|
||||
// upstream base) repo is the upstream base and headRepo is the fork origin.
|
||||
type sessionRepo struct {
|
||||
session domain.SessionRecord
|
||||
repo ports.SCMRepo
|
||||
branch string
|
||||
session domain.SessionRecord
|
||||
repo ports.SCMRepo
|
||||
headRepo ports.SCMRepo
|
||||
branch string
|
||||
}
|
||||
|
||||
type repoGuardState struct {
|
||||
|
|
@ -424,6 +430,8 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, [
|
|||
return nil, nil, err
|
||||
}
|
||||
projects := map[domain.ProjectID]domain.ProjectRecord{}
|
||||
originRepos := map[domain.ProjectID]ports.SCMRepo{}
|
||||
scanRepos := map[domain.ProjectID][]ports.SCMRepo{}
|
||||
out := map[string]*subject{}
|
||||
var sessionRepos []sessionRepo
|
||||
for _, sess := range sessions {
|
||||
|
|
@ -453,29 +461,87 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, [
|
|||
}
|
||||
projects[sess.ProjectID] = p
|
||||
proj = p
|
||||
if origin, ok := o.provider.ParseRepository(p.RepoOriginURL); ok {
|
||||
originRepos[sess.ProjectID] = origin
|
||||
scanRepos[sess.ProjectID] = o.resolveScanRepos(p, origin)
|
||||
}
|
||||
}
|
||||
repo, ok := o.provider.ParseRepository(proj.RepoOriginURL)
|
||||
origin, ok := originRepos[sess.ProjectID]
|
||||
if !ok {
|
||||
o.logger.Debug("scm observer: project has no supported SCM origin", "project", proj.ID, "origin", proj.RepoOriginURL)
|
||||
continue
|
||||
}
|
||||
sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, branch: branch})
|
||||
for _, repo := range scanRepos[sess.ProjectID] {
|
||||
sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, headRepo: origin, branch: branch})
|
||||
}
|
||||
prs, err := o.store.ListPRsBySession(ctx, sess.ID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
for _, pr := range openTrackedPRs(prs) {
|
||||
key := prKey(repo, pr.Number)
|
||||
// A tracked PR may live on an upstream repo (cross-fork), so its
|
||||
// refresh subject is keyed by the PR's own recorded repo, not the
|
||||
// push origin, or the GraphQL refetch would target the wrong repo.
|
||||
prRepo := subjectRepoForPR(pr, origin)
|
||||
key := prKey(prRepo, pr.Number)
|
||||
if existing, ok := out[key]; ok {
|
||||
o.logger.Warn("scm observer: duplicate tracked PR ownership skipped", "pr", key, "kept_session", existing.session.ID, "skipped_session", sess.ID)
|
||||
continue
|
||||
}
|
||||
out[key] = &subject{session: sess, repo: repo, branch: branch, known: pr, hasPR: true}
|
||||
out[key] = &subject{session: sess, repo: prRepo, branch: branch, known: pr, hasPR: true}
|
||||
}
|
||||
}
|
||||
return out, sessionRepos, nil
|
||||
}
|
||||
|
||||
// resolveScanRepos returns the deduped set of repos whose open-PR lists should be
|
||||
// scanned to attribute PRs to this project's sessions: the push origin plus every
|
||||
// other GitHub remote configured in the project checkout (upstreams, mirrors).
|
||||
// Attribution still requires a PR's head branch to live in the origin, so scanning
|
||||
// extra remotes only surfaces cross-fork PRs (fork head, upstream base) and can
|
||||
// never misattribute a stranger's PR.
|
||||
//
|
||||
// ponytail: remotes are read once per project per process (memoized by the
|
||||
// caller); a remote added after the daemon started is picked up on restart. Move
|
||||
// to a git-config watch if that latency ever matters.
|
||||
func (o *Observer) resolveScanRepos(proj domain.ProjectRecord, origin ports.SCMRepo) []ports.SCMRepo {
|
||||
repos := []ports.SCMRepo{origin}
|
||||
if strings.TrimSpace(proj.Path) == "" {
|
||||
return repos
|
||||
}
|
||||
seen := map[string]bool{prKey(origin, 0): true}
|
||||
for _, url := range gitRemoteURLs(proj.Path) {
|
||||
repo, ok := o.provider.ParseRepository(url)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
key := prKey(repo, 0)
|
||||
if seen[key] {
|
||||
continue
|
||||
}
|
||||
seen[key] = true
|
||||
repos = append(repos, repo)
|
||||
}
|
||||
return repos
|
||||
}
|
||||
|
||||
// subjectRepoForPR resolves the repo that owns a tracked PR's number for refresh.
|
||||
// A cross-fork PR lives on the base/upstream repo recorded in pr.Repo rather than
|
||||
// the push origin, so the refresh/GraphQL fetch must target pr.Repo. Legacy rows
|
||||
// written before pr.Repo was populated fall back to the origin.
|
||||
func subjectRepoForPR(pr domain.PullRequest, origin ports.SCMRepo) ports.SCMRepo {
|
||||
if strings.TrimSpace(pr.Repo) == "" {
|
||||
return origin
|
||||
}
|
||||
return ports.SCMRepo{
|
||||
Provider: firstNonEmpty(pr.Provider, origin.Provider),
|
||||
Host: firstNonEmpty(pr.Host, origin.Host),
|
||||
Repo: pr.Repo,
|
||||
Owner: ownerOf(pr.Repo),
|
||||
Name: nameOf(pr.Repo),
|
||||
}
|
||||
}
|
||||
|
||||
func openTrackedPRs(prs []domain.PullRequest) []domain.PullRequest {
|
||||
out := make([]domain.PullRequest, 0, len(prs))
|
||||
for _, pr := range prs {
|
||||
|
|
@ -550,20 +616,19 @@ func (o *Observer) discoverNewPRs(ctx context.Context, sessionRepos []sessionRep
|
|||
if pr.Number <= 0 || pr.SourceBranch == "" {
|
||||
continue
|
||||
}
|
||||
// Branch-prefix attribution must only claim PRs whose head branch
|
||||
// lives in the project repo. A fork PR can carry a head branch whose
|
||||
// name matches an AO session branch; its commits live in the fork, so
|
||||
// auto-claiming it would misattribute work. Same-repo PRs always
|
||||
// report the base repo's full name as their head repo, so anything
|
||||
// else (including an empty head repo from a deleted fork) is skipped.
|
||||
if !strings.EqualFold(pr.HeadRepo, repoFullName(repo)) {
|
||||
continue
|
||||
}
|
||||
key := prKey(repo, pr.Number)
|
||||
if _, ok := subjects[key]; ok {
|
||||
continue
|
||||
}
|
||||
sr, ok := matchSession(byRepo[repoKey], pr.SourceBranch)
|
||||
// Branch-prefix attribution must only claim PRs whose head branch
|
||||
// lives in a session's push origin. A same-repo PR has head == origin
|
||||
// == this scanned repo; a cross-fork PR (fork head, upstream base) has
|
||||
// head == origin while this scanned repo is the upstream base. A
|
||||
// stranger's fork PR carries a head repo no session owns and is
|
||||
// dropped (as is an empty head repo from a deleted fork), preserving
|
||||
// the no-misattribution guarantee.
|
||||
eligible := candidatesForHeadRepo(byRepo[repoKey], pr.HeadRepo)
|
||||
sr, ok := matchSession(eligible, pr.SourceBranch)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
|
@ -612,6 +677,24 @@ func (o *Observer) discoverNewPRs(ctx context.Context, sessionRepos []sessionRep
|
|||
// branches are prefixes of the same source branch the longest (most specific)
|
||||
// one wins, so a child session claims its own stacked PRs rather than the
|
||||
// ancestor session.
|
||||
// candidatesForHeadRepo narrows the scanned repo's session candidates to those
|
||||
// whose head branch lives in headRepo (the PR's head repository full name). This
|
||||
// is the fork guard: a PR is only attributable when its head repo equals a
|
||||
// session's push origin, whether the PR was found on the origin itself or on a
|
||||
// scanned upstream base repo.
|
||||
func candidatesForHeadRepo(candidates []sessionRepo, headRepo string) []sessionRepo {
|
||||
if strings.TrimSpace(headRepo) == "" {
|
||||
return nil
|
||||
}
|
||||
var out []sessionRepo
|
||||
for _, sr := range candidates {
|
||||
if strings.EqualFold(repoFullName(sr.headRepo), headRepo) {
|
||||
out = append(out, sr)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func matchSession(candidates []sessionRepo, sourceBranch string) (sessionRepo, bool) {
|
||||
var best sessionRepo
|
||||
bestLen := -1
|
||||
|
|
@ -1232,6 +1315,27 @@ func resolveGitOriginURL(path string) string {
|
|||
return strings.TrimSpace(string(out))
|
||||
}
|
||||
|
||||
// gitRemoteURLs lists the fetch URL of every git remote configured at path. It
|
||||
// returns nil on any error (missing repo, no git, no remotes). The observer uses
|
||||
// it to scan upstream/mirror remotes for cross-fork PRs in addition to origin.
|
||||
func gitRemoteURLs(path string) []string {
|
||||
out, err := aoprocess.Command("git", "-C", path, "remote").Output()
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
var urls []string
|
||||
for _, name := range strings.Fields(string(out)) {
|
||||
u, err := aoprocess.Command("git", "-C", path, "remote", "get-url", name).Output()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if s := strings.TrimSpace(string(u)); s != "" {
|
||||
urls = append(urls, s)
|
||||
}
|
||||
}
|
||||
return urls
|
||||
}
|
||||
|
||||
func scrubLine(s string) string {
|
||||
s = strings.ReplaceAll(s, "\n", " ")
|
||||
s = strings.ReplaceAll(s, "\r", " ")
|
||||
|
|
|
|||
|
|
@ -139,7 +139,18 @@ func (p *fakeProvider) SCMCredentialsAvailable(context.Context) (bool, error) {
|
|||
}
|
||||
|
||||
func (p *fakeProvider) ParseRepository(remote string) (ports.SCMRepo, bool) {
|
||||
return testRepo, remote != ""
|
||||
remote = strings.TrimSpace(remote)
|
||||
if remote == "" {
|
||||
return ports.SCMRepo{}, false
|
||||
}
|
||||
s := strings.TrimSuffix(remote, ".git")
|
||||
s = strings.TrimPrefix(s, "https://github.com/")
|
||||
s = strings.TrimPrefix(s, "git@github.com:")
|
||||
parts := strings.Split(strings.Trim(s, "/"), "/")
|
||||
if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
|
||||
return ports.SCMRepo{}, false
|
||||
}
|
||||
return ports.SCMRepo{Provider: "github", Host: "github.com", Owner: parts[0], Name: parts[1], Repo: parts[0] + "/" + parts[1]}, true
|
||||
}
|
||||
func (p *fakeProvider) RepoPRListGuard(_ context.Context, repo ports.SCMRepo, _ string) (ports.SCMGuardResult, error) {
|
||||
p.mu.Lock()
|
||||
|
|
@ -540,6 +551,111 @@ func TestPoll_IgnoresForkPRWithMatchingBranch(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func mustGit(t *testing.T, args ...string) {
|
||||
t.Helper()
|
||||
if out, err := exec.Command("git", args...).CombinedOutput(); err != nil {
|
||||
t.Fatalf("git %v: %v (%s)", args, err, out)
|
||||
}
|
||||
}
|
||||
|
||||
// A PR opened from the fork push origin against an upstream base repo (the
|
||||
// standard fork -> upstream contribution flow) must be discovered and attributed
|
||||
// by scanning every remote in the project checkout, not just origin. Its head
|
||||
// lives in origin (o/r) while its base lives in upstream (up/r); the persisted
|
||||
// row records the upstream base repo so refresh refetches against it.
|
||||
func TestPoll_DiscoversCrossForkPRFromUpstreamRemote(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
mustGit(t, "init", dir)
|
||||
mustGit(t, "-C", dir, "remote", "add", "origin", "https://github.com/o/r.git")
|
||||
mustGit(t, "-C", dir, "remote", "add", "upstream", "https://github.com/up/r.git")
|
||||
|
||||
upstream := ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "up", Name: "r", Repo: "up/r"}
|
||||
store := &fakeStore{
|
||||
sessions: []domain.SessionRecord{{ID: "p-1", ProjectID: "p", Metadata: domain.SessionMetadata{Branch: "ao/p-1/root"}}},
|
||||
projects: map[string]domain.ProjectRecord{"p": {ID: "p", Path: dir, RepoOriginURL: "https://github.com/o/r.git"}},
|
||||
prs: map[domain.SessionID][]domain.PullRequest{},
|
||||
checks: map[string][]domain.PullRequestCheck{},
|
||||
}
|
||||
crossObs := ports.SCMObservation{
|
||||
Fetched: true, Provider: "github", Host: "github.com", Repo: "up/r",
|
||||
PR: ports.SCMPRObservation{URL: "https://github.com/up/r/pull/1", Number: 1, State: "open", SourceBranch: "ao/p-1/feat", HeadRepo: "o/r", TargetBranch: "main", HeadSHA: "sha1", Title: "PR"},
|
||||
CI: ports.SCMCIObservation{Summary: string(domain.CIPassing), HeadSHA: "sha1"},
|
||||
Review: ports.SCMReviewObservation{Decision: string(domain.ReviewNone)},
|
||||
Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable), Mergeable: true},
|
||||
}
|
||||
provider := &fakeProvider{
|
||||
repoGuards: map[string]ports.SCMGuardResult{prKey(testRepo, 0): {ETag: "origin"}, prKey(upstream, 0): {ETag: "up"}},
|
||||
openPRs: map[string][]ports.SCMPRObservation{
|
||||
prKey(upstream, 0): {{URL: "https://github.com/up/r/pull/1", Number: 1, SourceBranch: "ao/p-1/feat", HeadRepo: "o/r", TargetBranch: "main", HeadSHA: "sha1"}},
|
||||
},
|
||||
observations: map[string]ports.SCMObservation{prKey(upstream, 1): crossObs},
|
||||
}
|
||||
lc := &fakeLifecycle{}
|
||||
obs := newTestObserver(store, provider, lc, time.Unix(1, 0).UTC())
|
||||
if err := obs.Poll(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(store.writes) == 0 {
|
||||
t.Fatal("expected cross-fork PR write")
|
||||
}
|
||||
got := store.writes[0].pr
|
||||
if got.SessionID != "p-1" {
|
||||
t.Fatalf("session id = %q, want p-1", got.SessionID)
|
||||
}
|
||||
if got.Repo != "up/r" {
|
||||
t.Fatalf("pr repo = %q, want up/r (upstream base)", got.Repo)
|
||||
}
|
||||
if got.SourceBranch != "ao/p-1/feat" {
|
||||
t.Fatalf("source branch = %q, want ao/p-1/feat", got.SourceBranch)
|
||||
}
|
||||
fetched := false
|
||||
for _, batch := range provider.fetchBatches {
|
||||
for _, ref := range batch {
|
||||
if ref.Repo.Repo == "up/r" && ref.Number == 1 {
|
||||
fetched = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if !fetched {
|
||||
t.Fatalf("cross-fork PR must be refreshed against upstream, batches=%#v", provider.fetchBatches)
|
||||
}
|
||||
}
|
||||
|
||||
// A PR on a scanned upstream remote whose head lives in some third-party fork
|
||||
// (not this project's origin) must never be attributed, even though its branch
|
||||
// name matches a session. Scanning extra remotes stays safe.
|
||||
func TestPoll_IgnoresUpstreamPRFromForeignHead(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
mustGit(t, "init", dir)
|
||||
mustGit(t, "-C", dir, "remote", "add", "origin", "https://github.com/o/r.git")
|
||||
mustGit(t, "-C", dir, "remote", "add", "upstream", "https://github.com/up/r.git")
|
||||
|
||||
upstream := ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "up", Name: "r", Repo: "up/r"}
|
||||
store := &fakeStore{
|
||||
sessions: []domain.SessionRecord{{ID: "p-1", ProjectID: "p", Metadata: domain.SessionMetadata{Branch: "ao/p-1/root"}}},
|
||||
projects: map[string]domain.ProjectRecord{"p": {ID: "p", Path: dir, RepoOriginURL: "https://github.com/o/r.git"}},
|
||||
prs: map[domain.SessionID][]domain.PullRequest{},
|
||||
checks: map[string][]domain.PullRequestCheck{},
|
||||
}
|
||||
provider := &fakeProvider{
|
||||
repoGuards: map[string]ports.SCMGuardResult{prKey(testRepo, 0): {ETag: "origin"}, prKey(upstream, 0): {ETag: "up"}},
|
||||
openPRs: map[string][]ports.SCMPRObservation{
|
||||
prKey(upstream, 0): {{URL: "https://github.com/up/r/pull/9", Number: 9, SourceBranch: "ao/p-1/feat", HeadRepo: "stranger/r", TargetBranch: "main", HeadSHA: "sha9"}},
|
||||
},
|
||||
observations: map[string]ports.SCMObservation{},
|
||||
}
|
||||
obs := newTestObserver(store, provider, &fakeLifecycle{}, time.Unix(1, 0).UTC())
|
||||
if err := obs.Poll(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(provider.fetchBatches) != 0 {
|
||||
t.Fatalf("foreign-head upstream PR must not be fetched, got %#v", provider.fetchBatches)
|
||||
}
|
||||
if len(store.writes) != 0 {
|
||||
t.Fatalf("foreign-head upstream PR must not be persisted, got %d writes", len(store.writes))
|
||||
}
|
||||
}
|
||||
|
||||
// A newly discovered open PR is persisted as a baseline row during discovery,
|
||||
// before the refresh/lifecycle pass. This is what lets a same-poll terminal
|
||||
// observation for a sibling PR see the open PR in the store and avoid completing
|
||||
|
|
|
|||
Loading…
Reference in New Issue