Hold a merge the controller cannot record now, and hand it over again, instead of taking it (issue 921)

cmd/mesh-controller/related_test.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
+	"fmt"
 	"strings"
 	"testing"
 	"time"
@@ -220,3 +222,14 @@ func TestARetryWaitsOnlyForAnOpenWalkItIsRelatedTo(t *testing.T) {
 		t.Error("an own-path retry went beside an open walk")
 	}
 }
+
+// Plans another controller is working are "not now", not a failure: the merge is held and handed over again
+// (issue 921).
+func TestPlansHeldByAnotherControllerAreNotNow(t *testing.T) {
+	if err := notNow(fmt.Errorf("x: %w", inventory.ErrPlansBusy)); !errors.Is(err, link.ErrTryAgain) {
+		t.Errorf("busy plans read as %v; want ErrTryAgain", err)
+	}
+	if err := notNow(errors.New("no such module")); errors.Is(err, link.ErrTryAgain) {
+		t.Errorf("another failure read as ErrTryAgain: %v", err)
+	}
+}

cmd/mesh-controller/upgrades.go

@@ -294,10 +294,10 @@ func (f following) Announceable(ctx context.Context) ([]link.Announcement, error
 	return out, nil
 }
 
-// notNow marks a store that could not be read right now, so the announcement is held rather than
+// notNow marks a store that could not be read right now, or plans another controller is working (issue 921), so
-// lost; anything else is returned as it was.
+// the announcement is held rather than lost; anything else is returned as it was.
 func notNow(err error) error {
-	if inventory.Unreachable(err) {
+	if inventory.Unreachable(err) || errors.Is(err, inventory.ErrPlansBusy) {
 		return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
 	}
 	return err

internal/link/serve.go

@@ -636,8 +636,14 @@ func short(commit string) string {
 	return commit
 }
 
-// sourceMoved acts on the forge's announcement of a merge. Taken whatever happens: a build that
+// mergeHeldFor is how long a merge the controller could not record yet waits before the bus hands it over again.
-// fails is reported by the build itself, and re-delivering the merge would only re-fail it.
+const mergeHeldFor = 5 * time.Second
+
+// sourceMoved acts on the forge's announcement of a merge. **A merge is never dropped** (issue 921, novox/hq ADR
+// 0407 decision 3): one the controller could not record for now (the plans held by another controller, the store
+// away: ErrTryAgain) is held and handed over again until it is recorded; the controller giving up on it would
+// lose a change for good. Anything else is an answer and is taken: a build that fails is reported by the build
+// itself, and re-delivering the merge would only re-fail it.
 func (s *Server) sourceMoved(ctx context.Context, m Control) {
 	var moved SourceMoved
 	if err := json.Unmarshal(m.Body(), &moved); err != nil {
@@ -650,7 +656,17 @@ func (s *Server) sourceMoved(ctx context.Context, m Control) {
 		_ = m.Took()
 		return
 	}
-	if err := s.upgrader.SourceMoved(ctx, moved); err != nil {
+	err := s.upgrader.SourceMoved(ctx, moved)
+	if err != nil && errors.Is(err, ErrTryAgain) && ctx.Err() == nil {
+		if held := m.Hold(mergeHeldFor); held == nil {
+			s.log.Printf("%s/%s moved to %.8s and is held: %v; handed over again in %s", moved.Owner, moved.Repo,
+				moved.Commit, err, mergeHeldFor)
+			return
+		} else {
+			s.log.Printf("LOST %s/%s moved to %.8s: it could not be held (%v): %v", moved.Owner, moved.Repo,
+				moved.Commit, held, err)
+		}
+	} else if err != nil {
 		s.log.Printf("%s/%s moved to %.8s and the mesh could not act on it: %v", moved.Owner, moved.Repo, moved.Commit, err)
 	}
 	_ = m.Took()

internal/link/store_window_test.go

@@ -106,3 +106,33 @@ func TestAMessageHandledDuringShutdownIsLeftForTheBus(t *testing.T) {
 		t.Fatalf("a build result handled during shutdown was settled, and so lost: %+v", *to)
 	}
 }
+
+// A merge the controller could not record for now (the plans held by another controller, the store away) is held
+// and handed over again, never taken: the bus keeps it until it is recorded (issue 921). Any other failure is an
+// answer and is taken, as before.
+func TestAMergeThatCouldNotBeRecordedNowIsHeldNotDropped(t *testing.T) {
+	merged := SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"}
+	for _, c := range []struct {
+		what string
+		err  error
+		want func(*settled) bool
+	}{
+		{"the plans held by another controller", errors.Join(ErrTryAgain, errors.New("another controller is working the plans")),
+			func(s *settled) bool { return s.unsettled() }},
+		{"the store away", errors.Join(ErrTryAgain, restarting), func(s *settled) bool { return s.unsettled() }},
+		{"cannot act", errors.New("no such module"), func(s *settled) bool { return s.acked && !s.nacked }},
+		{"kept", nil, func(s *settled) bool { return s.acked && !s.nacked }},
+	} {
+		s, in := serving()
+		s.upgrader = upgradesWith{err: c.err}
+		to := &settled{}
+		held := in.sends(t, to, KindSourceMoved, merged).(*fakeControl)
+		s.act(context.Background(), held)
+		if !c.want(to) {
+			t.Errorf("%s: a merge was settled as %+v", c.what, *to)
+		}
+		if c.err != nil && errors.Is(c.err, ErrTryAgain) && len(in.held) != 1 {
+			t.Errorf("%s: the merge was not held to be handed over again", c.what)
+		}
+	}
+}