Add a retracting state, and fully finish E2E retraction.

This means that when retracting an entry, if there were newer commits (merely only PATCHES related commits), then those are "squashed" together into a retract commit with authors preserved and pushed after resetting.

tools/mothership/proto/v1/entry.proto

@@ -109,10 +109,13 @@ message Entry {
     // This entry CAN'T be rescued.
     FAILED = 5;
 
+    // Retracting the entry.
+    RETRACTING = 6;
+
     // Retracted. This entry CAN'T be rescued.
     // Another import may have happened, retraction is usually done
     // if debranding was not complete but successful.
-    RETRACTED = 6;
+    RETRACTED = 7;
   }
   // State of the entry.
   State state = 14 [(google.api.field_behavior) = OUTPUT_ONLY];

tools/mothership/worker_server/BUILD

@@ -39,6 +39,8 @@ go_library(
         "//vendor/github.com/go-git/go-git/v5:go-git",
         "//vendor/github.com/go-git/go-git/v5/config",
         "//vendor/github.com/go-git/go-git/v5/plumbing",
+        "//vendor/github.com/go-git/go-git/v5/plumbing/object",
+        "//vendor/github.com/go-git/go-git/v5/plumbing/transport",
         "//vendor/github.com/go-git/go-git/v5/storage/memory",
         "//vendor/github.com/pkg/errors",
         "//vendor/github.com/sassoftware/go-rpmutils",

tools/mothership/worker_server/process_rpm_test.go

@@ -76,7 +76,7 @@ func TestWorker_ImportRPM_Existing(t *testing.T) {
 
 	remote := testW.forge.GetRemote("basesystem")
 
-	repo, err := getRepo(remote)
+	repo, err := getRepo(remote, nil)
 	require.Nil(t, err)
 
 	commitIter, err := repo.CommitObjects()

tools/mothership/worker_server/retract_entry.go

@@ -20,7 +20,10 @@ import (
 	"github.com/go-git/go-git/v5"
 	"github.com/go-git/go-git/v5/config"
 	"github.com/go-git/go-git/v5/plumbing"
+	"github.com/go-git/go-git/v5/plumbing/object"
+	"github.com/go-git/go-git/v5/plumbing/transport"
 	"github.com/go-git/go-git/v5/storage/memory"
+	"github.com/pkg/errors"
 	base "go.resf.org/peridot/base/go"
 	mshipadminpb "go.resf.org/peridot/tools/mothership/admin/pb"
 	mothership_db "go.resf.org/peridot/tools/mothership/db"
@@ -29,14 +32,13 @@ import (
 	"google.golang.org/grpc/status"
 	"io"
 	"os"
-	"regexp"
+	"strings"
+	"time"
 )
 
-var pkgNameRegexp = regexp.MustCompile(`^([a-zA-Z0-9\-]+)-\d+.*$`)
-
 // getRepo gets a git repository from a remote
 // It clones into an in-memory filesystem
-func getRepo(remote string) (*git.Repository, error) {
+func getRepo(remote string, auth transport.AuthMethod) (*git.Repository, error) {
 	// Just use in memory storage for all repos
 	storer := memory.NewStorage()
 	fs := memfs.New()
@@ -46,7 +48,7 @@ func getRepo(remote string) (*git.Repository, error) {
 	}
 
 	// Add a new remote
-	refspec := config.RefSpec("refs/heads/*:refs/remotes/origin/*")
+	refspec := config.RefSpec("refs/*:refs/*")
 	_, err = repo.CreateRemote(&config.RemoteConfig{
 		Name:  "origin",
 		URLs:  []string{remote},
@@ -59,6 +61,10 @@ func getRepo(remote string) (*git.Repository, error) {
 	// Fetch all the refs from the remote
 	err = repo.Fetch(&git.FetchOptions{
 		RemoteName: "origin",
+		Force:      true,
+		RefSpecs:   []config.RefSpec{refspec},
+		Tags:       git.AllTags,
+		Auth:       auth,
 	})
 	if err != nil {
 		return nil, err
@@ -106,14 +112,14 @@ func clonePathToFS(fromFS billy.Filesystem, toFS billy.Filesystem, rootPath stri
 			}
 		} else {
 			// open the file
-			f, err := fromFS.OpenFile(filePath, os.O_TRUNC|os.O_CREATE|os.O_RDWR, 0644)
+			f, err := fromFS.OpenFile(filePath, os.O_RDONLY, 0644)
 			if err != nil {
 				return err
 			}
 			defer f.Close()
 
 			// create the file in the toFS
-			toFile, err := toFS.Create(filePath)
+			toFile, err := toFS.OpenFile(filePath, os.O_TRUNC|os.O_CREATE|os.O_RDWR, 0644)
 			if err != nil {
 				return err
 			}
@@ -148,22 +154,127 @@ func clonePatchesToTemporaryFS(currentFS billy.Filesystem) (billy.Filesystem, er
 func resetRepoToPoint(repo *git.Repository, commit string) error {
 	wt, err := repo.Worktree()
 	if err != nil {
-		return err
+		return errors.Wrap(err, "failed to get worktree")
+	}
+
+	// Let's find out the commit before the commit we want to revert
+	log, err := repo.Log(&git.LogOptions{
+		From:  plumbing.NewHash(commit),
+		Order: git.LogOrderCommitterTime,
+	})
+	if err != nil {
+		return errors.Wrap(err, "failed to get log")
+	}
+
+	// log.Next() x2 should be the commit we want to revert
+	targetCommit, err := log.Next()
+	if err != nil {
+		return errors.Wrap(err, "failed to get next commit x1")
+	}
+	resetToCommit, err := log.Next()
+	if err != nil {
+		return errors.Wrap(err, "failed to get next commit x2")
+	}
+
+	// Also get all commits that touches the PATCHES directory
+	// until the commit we want to revert
+	firstLog, err := repo.Log(&git.LogOptions{
+		Order: git.LogOrderCommitterTime,
+		PathFilter: func(s string) bool {
+			// Only include PATCHES
+			if strings.HasPrefix(s, "PATCHES") {
+				return true
+			}
+
+			return false
+		},
+		// Limit to until the commit we want to revert
+		Since: &resetToCommit.Author.When,
+	})
+	if err != nil {
+		return errors.Wrap(err, "failed to get log")
+	}
+
+	// Get all authors of the commits, since we're going to copy PATCHES
+	// back into the repo
+	var commits []*object.Commit
+	for {
+		c, err := firstLog.Next()
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return errors.Wrap(err, "failed to get next commit")
+		}
+		// If the commit was created before the resetToCommit, then we don't want to include it
+		if c.Author.When.Before(targetCommit.Author.When) {
+			// Breaking because the commits are sorted by date, so if we encounter a commit that was created before the resetToCommit,
+			// then all the following commits will be created before the resetToCommit
+			break
+		}
+		commits = append(commits, c)
+	}
+
+	// Copy PATCHES into a temporary filesystem
+	patchesFS, err := clonePatchesToTemporaryFS(wt.Filesystem)
+	if err != nil {
+		return errors.Wrap(err, "failed to clone PATCHES")
 	}
 
 	// reset the repo
 	err = wt.Reset(&git.ResetOptions{
-		Commit: plumbing.NewHash(commit),
+		Commit: resetToCommit.Hash,
 		Mode:   git.HardReset,
 	})
 	if err != nil {
-		return err
+		return errors.Wrap(err, "failed to reset repo")
 	}
 
-	// clean the repo
-	return wt.Clean(&git.CleanOptions{
-		Dir: true,
-	})
+	// Copy PATCHES back into the repo
+	err = clonePathToFS(patchesFS, wt.Filesystem, "PATCHES")
+	if err != nil {
+		return errors.Wrap(err, "failed to copy PATCHES")
+	}
+
+	// If there are diffs, then create a commit consisting of the joined messages and authors
+	// of the commits that touches the PATCHES directory
+	if len(commits) > 0 {
+		// Add the files
+		_, err = wt.Add(".")
+		if err != nil {
+			return errors.Wrap(err, "failed to add PATCHES")
+		}
+
+		// Get the commit message
+		commitMsg := "Retract \"" + targetCommit.Message + "\"\n\nFast-forwarded following commits:\n"
+		for _, c := range commits {
+			commitMsg += c.Message + "\n"
+		}
+		commitMsg += "\n"
+
+		// Add the authors as "Co-authored-by"
+		authors := make(map[string]bool)
+		for _, c := range commits {
+			authors[c.Author.Name+" <"+c.Author.Email+">"] = true
+		}
+		for author := range authors {
+			commitMsg += "Co-authored-by: " + author + "\n"
+		}
+
+		// Create the commit
+		_, err = wt.Commit(commitMsg, &git.CommitOptions{
+			Committer: &object.Signature{
+				Name:  targetCommit.Author.Name,
+				Email: targetCommit.Author.Email,
+				When:  time.Now(),
+			},
+		})
+		if err != nil {
+			return errors.Wrap(err, "failed to commit")
+		}
+	}
+
+	return err
 }
 
 func (w *Worker) RetractEntry(name string) (*mshipadminpb.RetractEntryResponse, error) {
@@ -183,25 +294,48 @@ func (w *Worker) RetractEntry(name string) (*mshipadminpb.RetractEntryResponse,
 
 	// Get the repo
 	remote := w.forge.GetRemote(entry.PackageName)
-	repo, err := getRepo(remote)
+	auth, err := w.forge.GetAuthenticator()
+	if err != nil {
+		base.LogErrorf("failed to get forge authenticator: %v", err)
+		return nil, status.Error(codes.Internal, "failed to get forge authenticator")
+	}
+	repo, err := getRepo(remote, auth.AuthMethod)
 	if err != nil {
 		base.LogErrorf("failed to get repo: %v", err)
 		return nil, status.Error(codes.Internal, "failed to get repo")
 	}
 
-	err = repo.DeleteObject(plumbing.NewHash(entry.CommitHash))
+	// Checkout the entry branch
+	wt, err := repo.Worktree()
 	if err != nil {
-		base.LogErrorf("failed to eject commit: %v", err)
-		return nil, status.Error(codes.Internal, "failed to eject commit")
+		base.LogErrorf("failed to get worktree: %v", err)
+		return nil, status.Error(codes.Internal, "failed to get worktree")
 	}
 
-	// If there's a tag, delete it
-	_ = repo.DeleteTag(entry.CommitTag)
+	err = wt.Checkout(&git.CheckoutOptions{
+		Branch: plumbing.NewBranchReferenceName(entry.CommitBranch),
+		Force:  true,
+	})
+	if err != nil {
+		base.LogErrorf("failed to checkout branch: %v", err)
+		return nil, status.Error(codes.Internal, "failed to checkout branch")
+	}
+
+	// Reset the repo to the commit before the commit we want to revert
+	err = resetRepoToPoint(repo, entry.CommitHash)
+	if err != nil {
+		base.LogErrorf("failed to reset repo: %v", err)
+		return nil, status.Error(codes.Internal, "failed to reset repo")
+	}
 
 	// Push the changes
 	err = repo.Push(&git.PushOptions{
 		RemoteName: "origin",
 		Force:      true,
+		Auth:       auth.AuthMethod,
+		RefSpecs: []config.RefSpec{
+			config.RefSpec("refs/heads/" + entry.CommitBranch + ":refs/heads/" + entry.CommitBranch),
+		},
 	})
 	if err != nil {
 		base.LogErrorf("failed to push changes: %v", err)

tools/mothership/worker_server/workflows.go

@@ -201,12 +201,40 @@ func ProcessRPMWorkflow(ctx workflow.Context, args *mothershippb.ProcessRPMArgs)
 // The same source (for the specific entry) can be re-imported by the client, either by calling DuplicateEntry or
 // calling SubmitEntry with the same SRPM URI.
 func RetractEntryWorkflow(ctx workflow.Context, name string) (*mshipadminpb.RetractEntryResponse, error) {
+	// Set entry state to retracting
+	var entry mothershippb.Entry
+	ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+		StartToCloseTimeout: 25 * time.Second,
+	})
+	err := workflow.ExecuteActivity(ctx, w.SetEntryState, name, mothershippb.Entry_RETRACTING, nil).Get(ctx, &entry)
+	if err != nil {
+		return nil, err
+	}
+
+	// Deferring this activity will set the entry state to ARCHIVED if the workflow
+	// is not completed.
+	defer func() {
+		if entry.State == mothershippb.Entry_RETRACTED {
+			return
+		}
+
+		// This is because the entry is still archived, but the commit was not
+		// retracted.
+		ctx, cancel := workflow.NewDisconnectedContext(ctx)
+		defer cancel()
+		ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
+			StartToCloseTimeout: 25 * time.Second,
+		})
+		_ = workflow.ExecuteActivity(ctx, w.SetEntryState, name, mothershippb.Entry_ARCHIVED, nil).Get(ctx, nil)
+	}()
+
+	// Retract commit
 	ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 		StartToCloseTimeout: 60 * time.Second,
 	})
 
 	var res mshipadminpb.RetractEntryResponse
-	err := workflow.ExecuteActivity(ctx, w.RetractEntry, name).Get(ctx, &res)
+	err = workflow.ExecuteActivity(ctx, w.RetractEntry, name).Get(ctx, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -215,7 +243,7 @@ func RetractEntryWorkflow(ctx workflow.Context, name string) (*mshipadminpb.Retr
 	ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 		StartToCloseTimeout: 25 * time.Second,
 	})
-	err = workflow.ExecuteActivity(ctx, w.SetEntryState, name, mothershippb.Entry_RETRACTED, nil).Get(ctx, nil)
+	err = workflow.ExecuteActivity(ctx, w.SetEntryState, name, mothershippb.Entry_RETRACTED, nil).Get(ctx, &entry)
 	if err != nil {
 		return nil, err
 	}