From ff00752b5734522e0e3d107f114ca5c73776e1af Mon Sep 17 00:00:00 2001 From: dawn Date: Sat, 18 Jul 2026 11:39:50 +0300 Subject: [PATCH] appview/{pulls,db,pages}: use spindle describe wf xrpc for wf change detection Signed-off-by: dawn --- appview/db/db.go | 7 ++ appview/db/pulls.go | 24 +++++-- appview/models/pull.go | 1 + appview/pages/pages.go | 4 +- appview/pulls/create.go | 12 ++-- appview/pulls/pulls.go | 57 +++++++++------- appview/pulls/resubmit.go | 11 +-- appview/pulls/single.go | 5 +- appview/pulls/trigger_ci.go | 31 ++------- appview/pulls/workflow_change.go | 114 +++++++++++++++++++++++++++++++ 10 files changed, 194 insertions(+), 72 deletions(-) create mode 100644 appview/pulls/workflow_change.go diff --git a/appview/db/db.go b/appview/db/db.go index 507104b5..d2ac87ae 100644 --- a/appview/db/db.go +++ b/appview/db/db.go @@ -2441,6 +2441,13 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { return err }) + orm.RunMigration(conn, logger, "add-pull-submissions-merge-base", func(tx *sql.Tx) error { + _, err := tx.Exec(` + alter table pull_submissions add column merge_base text; + `) + return err + }) + return &DB{ db, logger, diff --git a/appview/db/pulls.go b/appview/db/pulls.go index bd2b28cd..22921368 100644 --- a/appview/db/pulls.go +++ b/appview/db/pulls.go @@ -196,17 +196,19 @@ func createNewPull(tx *sql.Tx, pull *models.Pull) error { patch, combined, source_rev, + merge_base, patch_blob_ref, patch_blob_mime, patch_blob_size ) - values (?, ?, ?, ?, ?, ?, ?, ?) + values (?, ?, ?, ?, ?, ?, ?, ?, ?) `, pull.AtUri(), i, s.Patch, s.Combined, s.SourceRev, + s.MergeBase, s.Blob.Ref.String(), s.Blob.MimeType, s.Blob.Size, @@ -257,17 +259,19 @@ func updatePull(tx *sql.Tx, pull *models.Pull, existingPull *models.Pull) error patch, combined, source_rev, + merge_base, patch_blob_ref, patch_blob_mime, patch_blob_size ) - values (?, ?, ?, ?, ?, ?, ?, ?) + values (?, ?, ?, ?, ?, ?, ?, ?, ?) `, pull.AtUri(), i, s.Patch, s.Combined, s.SourceRev, + s.MergeBase, s.Blob.Ref.String(), s.Blob.MimeType, s.Blob.Size, @@ -278,7 +282,7 @@ func updatePull(tx *sql.Tx, pull *models.Pull, existingPull *models.Pull) error } if err := putReferences(tx, pull.AtUri(), pull.References); err != nil { - return fmt.Errorf("put reference_links: %w", err) + return err } return nil } @@ -507,6 +511,7 @@ func GetPullSubmissions(e Execer, filters ...orm.Filter) (map[syntax.ATURI][]*mo combined, created, source_rev, + merge_base, patch_blob_ref, patch_blob_mime, patch_blob_size @@ -528,7 +533,7 @@ func GetPullSubmissions(e Execer, filters ...orm.Filter) (map[syntax.ATURI][]*mo for rows.Next() { var submission models.PullSubmission var submissionCreatedStr string - var submissionSourceRev, submissionCombined sql.Null[string] + var submissionSourceRev, submissionCombined, submissionMergeBase sql.Null[string] var patchBlobRef, patchBlobMime sql.Null[string] var patchBlobSize sql.Null[int64] err := rows.Scan( @@ -539,6 +544,7 @@ func GetPullSubmissions(e Execer, filters ...orm.Filter) (map[syntax.ATURI][]*mo &submissionCombined, &submissionCreatedStr, &submissionSourceRev, + &submissionMergeBase, &patchBlobRef, &patchBlobMime, &patchBlobSize, @@ -555,6 +561,10 @@ func GetPullSubmissions(e Execer, filters ...orm.Filter) (map[syntax.ATURI][]*mo submission.SourceRev = submissionSourceRev.V } + if submissionMergeBase.Valid { + submission.MergeBase = submissionMergeBase.V + } + if submissionCombined.Valid { submission.Combined = submissionCombined.V } @@ -729,6 +739,7 @@ func ResubmitPull( newPatch string, combinedPatch string, newSourceRev string, + mergeBase string, blob *lexutil.LexBlob, ) error { _, err := e.Exec(` @@ -738,12 +749,13 @@ func ResubmitPull( patch, combined, source_rev, + merge_base, patch_blob_ref, patch_blob_mime, patch_blob_size ) - values (?, ?, ?, ?, ?, ?, ?, ?) - `, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, blob.Ref.String(), blob.MimeType, blob.Size) + values (?, ?, ?, ?, ?, ?, ?, ?, ?) + `, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, mergeBase, blob.Ref.String(), blob.MimeType, blob.Size) return err } diff --git a/appview/models/pull.go b/appview/models/pull.go index 3988b0d0..bbc580c3 100644 --- a/appview/models/pull.go +++ b/appview/models/pull.go @@ -311,6 +311,7 @@ type PullSubmission struct { Combined string Comments []Comment SourceRev string // include the rev that was used to create this submission: only for branch/fork PRs + MergeBase string // merge-base of source and target at submission time // meta Created time.Time diff --git a/appview/pages/pages.go b/appview/pages/pages.go index 82c0a3ce..f1dfc59e 100644 --- a/appview/pages/pages.go +++ b/appview/pages/pages.go @@ -1522,8 +1522,8 @@ type PullActionsParams struct { Stack models.Stack // Workflow warning state for fork-based pulls without a pipeline on the - // latest commit. WorkflowsChanged and ChangedWorkflowFiles are computed - // from the latest round's patch. + // latest commit, derived from the spindle's workflow-definition + // fingerprints at the pull head and its merge-base with the target branch. WorkflowsChanged bool ChangedWorkflowFiles []string HasPipeline bool diff --git a/appview/pulls/create.go b/appview/pulls/create.go index 2767f663..469553d6 100644 --- a/appview/pulls/create.go +++ b/appview/pulls/create.go @@ -70,10 +70,11 @@ func (s *Pulls) handleBranchBasedPull( sourceRev := comparison.Rev2 patch := comparison.FormatPatchRaw combined := comparison.CombinedPatchRaw + mergeBase := comparison.MergeBase if err := validatePatch(&patch); err != nil { s.logger.Error("failed to validate patch", "err", err) - s.pages.Notice(w, "pull", "Invalid patch format. Please provide a valid diff.") + s.pages.Notice(w, "pull", "Failed to create pull request. Try again later.") return } @@ -81,7 +82,7 @@ func (s *Pulls) handleBranchBasedPull( Branch: sourceBranch, } - s.createPullRequest(w, r, repo, userDid, title, body, targetBranch, patch, combined, sourceRev, pullSource, isStacked, stackTitles, stackBodies) + s.createPullRequest(w, r, repo, userDid, title, body, targetBranch, patch, combined, sourceRev, mergeBase, pullSource, isStacked, stackTitles, stackBodies) } func (s *Pulls) handlePatchBasedPull(w http.ResponseWriter, r *http.Request, repo *models.Repo, userDid syntax.DID, title, body, targetBranch, patch string, isStacked bool, stackTitles, stackBodies map[string]string) { @@ -91,7 +92,7 @@ func (s *Pulls) handlePatchBasedPull(w http.ResponseWriter, r *http.Request, rep return } - s.createPullRequest(w, r, repo, userDid, title, body, targetBranch, patch, "", "", nil, isStacked, stackTitles, stackBodies) + s.createPullRequest(w, r, repo, userDid, title, body, targetBranch, patch, "", "", "", nil, isStacked, stackTitles, stackBodies) } func (s *Pulls) handleForkBasedPull(w http.ResponseWriter, r *http.Request, repo *models.Repo, userDid syntax.DID, forkRepoDid string, title, body, targetBranch, sourceBranch string, isStacked bool, stackTitles, stackBodies map[string]string) { @@ -177,6 +178,7 @@ func (s *Pulls) handleForkBasedPull(w http.ResponseWriter, r *http.Request, repo sourceRev := comparison.Rev2 patch := comparison.FormatPatchRaw combined := comparison.CombinedPatchRaw + mergeBase := comparison.MergeBase if err := validatePatch(&patch); err != nil { s.logger.Error("failed to validate patch", "err", err) @@ -190,7 +192,7 @@ func (s *Pulls) handleForkBasedPull(w http.ResponseWriter, r *http.Request, repo RepoDid: &forkDid, } - s.createPullRequest(w, r, repo, userDid, title, body, targetBranch, patch, combined, sourceRev, pullSource, isStacked, stackTitles, stackBodies) + s.createPullRequest(w, r, repo, userDid, title, body, targetBranch, patch, combined, sourceRev, mergeBase, pullSource, isStacked, stackTitles, stackBodies) } func (s *Pulls) createPullRequest( @@ -202,6 +204,7 @@ func (s *Pulls) createPullRequest( patch string, combined string, sourceRev string, + mergeBase string, pullSource *models.PullSource, isStacked bool, stackTitles, stackBodies map[string]string, @@ -288,6 +291,7 @@ func (s *Pulls) createPullRequest( Patch: patch, Combined: combined, SourceRev: sourceRev, + MergeBase: mergeBase, Blob: *blob.Blob, Created: now, }, diff --git a/appview/pulls/pulls.go b/appview/pulls/pulls.go index 1adcb13e..fdc084ff 100644 --- a/appview/pulls/pulls.go +++ b/appview/pulls/pulls.go @@ -9,6 +9,7 @@ import ( "strings" "time" + "tangled.org/core/api/tangled" "tangled.org/core/appview/config" "tangled.org/core/appview/db" pulls_indexer "tangled.org/core/appview/indexer/pulls" @@ -36,19 +37,21 @@ const ( ) type Pulls struct { - oauth *oauth.OAuth - repoResolver *reporesolver.RepoResolver - pages *pages.Pages - idResolver *idresolver.Resolver - mentionsResolver *mentions.Resolver - db *db.DB - config *config.Config - notifier notify.Notifier - acl *knotacl.Service - logger *slog.Logger - indexer *pulls_indexer.Indexer - ogreClient *ogre.Client - diffCache *expirable.LRU[string, types.DiffRenderer] + oauth *oauth.OAuth + repoResolver *reporesolver.RepoResolver + pages *pages.Pages + idResolver *idresolver.Resolver + mentionsResolver *mentions.Resolver + db *db.DB + config *config.Config + notifier notify.Notifier + acl *knotacl.Service + logger *slog.Logger + indexer *pulls_indexer.Indexer + ogreClient *ogre.Client + diffCache *expirable.LRU[string, types.DiffRenderer] + workflowDescCache *expirable.LRU[string, *tangled.CiDescribeWorkflowDefinition_Output] + mergeBaseCache *expirable.LRU[string, string] } func New( @@ -65,19 +68,21 @@ func New( logger *slog.Logger, ) *Pulls { return &Pulls{ - oauth: oauth, - repoResolver: repoResolver, - pages: pages, - idResolver: resolver, - mentionsResolver: mentionsResolver, - db: db, - config: config, - notifier: notifier, - acl: acl, - logger: logger, - indexer: indexer, - ogreClient: ogre.NewClient(config.Ogre.Host), - diffCache: expirable.NewLRU[string, types.DiffRenderer](diffCacheSize, nil, diffCacheTTL), + oauth: oauth, + repoResolver: repoResolver, + pages: pages, + idResolver: resolver, + mentionsResolver: mentionsResolver, + db: db, + config: config, + notifier: notifier, + acl: acl, + logger: logger, + indexer: indexer, + ogreClient: ogre.NewClient(config.Ogre.Host), + diffCache: expirable.NewLRU[string, types.DiffRenderer](diffCacheSize, nil, diffCacheTTL), + workflowDescCache: expirable.NewLRU[string, *tangled.CiDescribeWorkflowDefinition_Output](workflowCacheSize, nil, workflowCacheTTL), + mergeBaseCache: expirable.NewLRU[string, string](workflowCacheSize, nil, workflowCacheTTL), } } diff --git a/appview/pulls/resubmit.go b/appview/pulls/resubmit.go index 5fb6fbd7..aa0251d2 100644 --- a/appview/pulls/resubmit.go +++ b/appview/pulls/resubmit.go @@ -91,7 +91,7 @@ func (s *Pulls) resubmitPatch(w http.ResponseWriter, r *http.Request) { patch := r.FormValue("patch") - s.resubmitPullHelper(w, r, f, syntax.DID(user.Did), pull, patch, "", "") + s.resubmitPullHelper(w, r, f, syntax.DID(user.Did), pull, patch, "", "", "") } func (s *Pulls) resubmitBranch(w http.ResponseWriter, r *http.Request) { @@ -154,7 +154,7 @@ func (s *Pulls) resubmitBranch(w http.ResponseWriter, r *http.Request) { patch := comparison.FormatPatchRaw combined := comparison.CombinedPatchRaw - s.resubmitPullHelper(w, r, f, syntax.DID(user.Did), pull, patch, combined, sourceRev) + s.resubmitPullHelper(w, r, f, syntax.DID(user.Did), pull, patch, combined, sourceRev, comparison.MergeBase) } func (s *Pulls) resubmitFork(w http.ResponseWriter, r *http.Request) { @@ -252,7 +252,7 @@ func (s *Pulls) resubmitFork(w http.ResponseWriter, r *http.Request) { patch := comparison.FormatPatchRaw combined := comparison.CombinedPatchRaw - s.resubmitPullHelper(w, r, f, syntax.DID(user.Did), pull, patch, combined, sourceRev) + s.resubmitPullHelper(w, r, f, syntax.DID(user.Did), pull, patch, combined, sourceRev, comparison.MergeBase) } func (s *Pulls) resubmitPullHelper( @@ -264,6 +264,7 @@ func (s *Pulls) resubmitPullHelper( patch string, combined string, sourceRev string, + mergeBase string, ) { l := s.logger.With("handler", "resubmitPullHelper", "user", userDid, "pull_id", pull.PullId, "target_branch", pull.TargetBranch) @@ -338,7 +339,7 @@ func (s *Pulls) resubmitPullHelper( return } - err = db.ResubmitPull(s.db, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, blob.Blob) + err = db.ResubmitPull(s.db, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, mergeBase, blob.Blob) if err != nil { l.Error("failed to resubmit pull request in database", "err", err, "round_number", newRoundNumber) s.pages.Notice(w, "resubmit-error", "Failed to create pull request. Try again later.") @@ -550,7 +551,7 @@ func (s *Pulls) resubmitStackedPullHelper( } // create new round - err = db.ResubmitPull(tx, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, blob.Blob) + err = db.ResubmitPull(tx, pullAt, newRoundNumber, newPatch, combinedPatch, newSourceRev, np.LatestSubmission().MergeBase, blob.Blob) if err != nil { l.Error("failed to update pull in database", "err", err, "pull_id", op.PullId, "round_number", newRoundNumber) s.pages.Notice(w, "pull-resubmit-error", "Failed to resubmit pull request. Try again later.") diff --git a/appview/pulls/single.go b/appview/pulls/single.go index 6dfcda5b..2b4d4f46 100644 --- a/appview/pulls/single.go +++ b/appview/pulls/single.go @@ -76,11 +76,10 @@ func (s *Pulls) PullActions(w http.ResponseWriter, r *http.Request) { } if pull.IsForkBased() && !hasPipeline { - changedWorkflows, err = changedWorkflowFiles(pull.LatestSubmission().CombinedPatch()) + workflowsChanged, changedWorkflows, err = s.workflowChangeFromSpindle(r, f, pull) if err != nil { - l.Error("failed to inspect latest round's patch for workflow changes", "err", err) + l.Error("failed to describe workflow definitions", "err", err) } - workflowsChanged = len(changedWorkflows) > 0 } } diff --git a/appview/pulls/trigger_ci.go b/appview/pulls/trigger_ci.go index 61dccf79..63e08580 100644 --- a/appview/pulls/trigger_ci.go +++ b/appview/pulls/trigger_ci.go @@ -8,31 +8,8 @@ import ( "tangled.org/core/api/tangled" "tangled.org/core/appview/db" "tangled.org/core/appview/models" - "tangled.org/core/patchutil" - "tangled.org/core/workflow" ) -func changedWorkflowFiles(patch string) ([]string, error) { - files, err := patchutil.AsDiff(patch) - if err != nil { - return nil, err - } - - var changed []string - for _, f := range files { - if f == nil { - continue - } - for _, name := range []string{f.NewName, f.OldName} { - if name != "" && strings.HasPrefix(name, workflow.WorkflowDir+"/") { - changed = append(changed, name) - break - } - } - } - return changed, nil -} - // TriggerCi manually triggers a CI pipeline for a fork-based pull request. // authorized against and recorded under the target repo, but checked out // from the fork at the latest round's commit. @@ -78,12 +55,14 @@ func (s *Pulls) TriggerCi(w http.ResponseWriter, r *http.Request) { return } - changedFiles, err := changedWorkflowFiles(latest.CombinedPatch()) + workflowsChanged, changedFiles, err := s.workflowChangeFromSpindle(r, f, pull) if err != nil { - fail("failed to inspect the latest round's patch", err) + // without the fingerprint comparison there is no way to tell fork + // workflows apart from reviewed ones, so refuse to run them + fail("failed to verify the workflow definitions of this round", err) return } - if len(changedFiles) > 0 && r.URL.Query().Get("confirm") != "1" { + if workflowsChanged && r.URL.Query().Get("confirm") != "1" { fail(fmt.Sprintf("workflow files changed in this round (%s); review before running", strings.Join(changedFiles, ", ")), nil) return } diff --git a/appview/pulls/workflow_change.go b/appview/pulls/workflow_change.go new file mode 100644 index 00000000..6287a79e --- /dev/null +++ b/appview/pulls/workflow_change.go @@ -0,0 +1,114 @@ +package pulls + +import ( + "encoding/json" + "fmt" + "net/http" + "slices" + "time" + + indigoxrpc "github.com/bluesky-social/indigo/xrpc" + + "tangled.org/core/api/tangled" + "tangled.org/core/appview/db" + "tangled.org/core/appview/models" + "tangled.org/core/hostutil" + "tangled.org/core/types" +) + +// spindle upgrades may change how a commit is resolved and hashed (even if very rare...) +const ( + workflowCacheSize = 4096 + workflowCacheTTL = time.Hour +) + +func (s *Pulls) describeWorkflowAt(r *http.Request, f *models.Repo, sha, sourceRepo string) (*tangled.CiDescribeWorkflowDefinition_Output, error) { + key := fmt.Sprintf("%s|%s|%s|%s", f.Spindle, f.RepoDid, sha, sourceRepo) + if cached, ok := s.workflowDescCache.Get(key); ok { + return cached, nil + } + + spindleUrl, err := hostutil.EnsureHttpScheme(f.Spindle) + if err != nil { + return nil, err + } + client := &indigoxrpc.Client{Host: spindleUrl} + out, err := tangled.CiDescribeWorkflowDefinition(r.Context(), client, f.RepoDid, sha, sourceRepo) + if err != nil { + return nil, err + } + + s.workflowDescCache.Add(key, out) + return out, nil +} + +// returns the merge-base of the pull's source and target branches. +// +// older rounds resolve it live on the fork's knot, which tracks the +// target branch as refs/hidden//. +func (s *Pulls) mergeBase(r *http.Request, pull *models.Pull) (string, error) { + if mb := pull.LatestSubmission().MergeBase; mb != "" { + return mb, nil + } + + sourceBranch := pull.PullSource.Branch + hiddenRef := fmt.Sprintf("hidden/%s/%s", sourceBranch, pull.TargetBranch) + + key := fmt.Sprintf("%s|%s|%s", pull.PullSource.RepoDid, sourceBranch, pull.TargetBranch) + if cached, ok := s.mergeBaseCache.Get(key); ok { + return cached, nil + } + + forkRepo, err := db.GetRepoByDid(s.db, pull.PullSource.RepoDid.String()) + if err != nil { + return "", fmt.Errorf("resolving fork repo: %w", err) + } + compareBytes, err := tangled.RepoCompare(r.Context(), s.knotClient(forkRepo.Knot), forkRepo.RepoIdentifier(), hiddenRef, sourceBranch) + if err != nil { + return "", fmt.Errorf("comparing %s...%s: %w", hiddenRef, sourceBranch, err) + } + var comparison types.RepoFormatPatchResponse + if err := json.Unmarshal(compareBytes, &comparison); err != nil { + return "", fmt.Errorf("decoding comparison: %w", err) + } + if comparison.MergeBase == "" { + return "", fmt.Errorf("no merge base between %s and %s", hiddenRef, sourceBranch) + } + + s.mergeBaseCache.Add(key, comparison.MergeBase) + return comparison.MergeBase, nil +} + +// compares the spindle's workflow-definition fingerprints at the +// pull head and its merge-base with the target branch. +func (s *Pulls) workflowChangeFromSpindle(r *http.Request, f *models.Repo, pull *models.Pull) (bool, []string, error) { + sourceRepo := pull.PullSource.RepoDid.String() + + head, err := s.describeWorkflowAt(r, f, pull.LatestSha(), sourceRepo) + if err != nil { + return false, nil, err + } + // a definition not derived from the repo cannot change between commits + if !head.Derived { + return false, nil, nil + } + + baseSha, err := s.mergeBase(r, pull) + if err != nil { + return false, nil, err + } + base, err := s.describeWorkflowAt(r, f, baseSha, sourceRepo) + if err != nil { + return false, nil, err + } + if !base.Derived { + return false, nil, nil + } + if head.Hash != nil && base.Hash != nil && *head.Hash == *base.Hash { + return false, nil, nil + } + + names := slices.Concat(head.Workflows, base.Workflows) + slices.Sort(names) + return true, slices.Compact(names), nil +} -- 2.51.2