Give the forge's merges a durable consumer of their own: no max-deliver, no reset, a growing delay while held (issue 973)

internal/broker/controller_buckets_test.go

@@ -320,3 +320,51 @@ func TestTheMergeWindowsKeysExpireAndAWatcherHearsIt(t *testing.T) {
 		}
 	}
 }
+
+// **The merges have a consumer of their own, which a reset of the events consumer leaves alone** (issue 973, novox/hq
+// ADR 0407 decision 3): healer H4 re-makes the events consumer from now, dropping what it holds pending; a merge
+// waiting in the merges consumer is not that consumer's, and nothing would catch it up.
+func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testing.T) {
+	js, err := Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer js.Close()
+	if err := AssertMeshStreams(js); err != nil {
+		t.Fatal(err)
+	}
+	if err := AssertMeshConsumers(js); err != nil {
+		t.Fatal(err)
+	}
+	for _, c := range MeshConsumers() {
+		if c.Stream == EventsStream && c.Name == MergesConsumerName {
+			if c.Resettable != "" || c.MaxDeliver != 0 || len(c.Filters) != 1 || c.Filters[0] != ControllerFollows[3] {
+				t.Errorf("the merges consumer is %+v; want one filter, no max-deliver, not resettable", c)
+			}
+		}
+		if c.Stream == EventsStream && c.Name == ControllerName {
+			for _, f := range c.Filters {
+				if f == ControllerFollows[3] {
+					t.Errorf("the events consumer still follows the merges, which the merges consumer takes")
+				}
+			}
+		}
+	}
+	if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil {
+		t.Fatal(err)
+	}
+	merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
+	if err != nil || merges.NumPending != 1 {
+		t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err)
+	}
+	if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
+		t.Fatal(err)
+	}
+	merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
+	if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 {
+		t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err)
+	}
+}

internal/broker/derived.go

@@ -36,6 +36,8 @@ type Consumer struct {
 	// them over, and a loop that acts on one message at a time is the property the store window
 	// depends on. A queue group implies this, because a group has nothing to pull from.
 	Push bool
+	// DeliverSubject, when set, is where a push consumer delivers instead of DeliverSubjectFor's.
+	DeliverSubject string
 	// AckWaitSeconds before an unacknowledged delivery is redelivered.
 	AckWaitSeconds int
 	// MaxDeliver is how often a message is handed over before the consumer gives it up; zero for no

internal/broker/nats.go

@@ -368,6 +368,11 @@ var DegradableGrants = []DegradableGrant{
 		Subject: "mesh.seat." + VerbsTheControllerAsksTheOperatorChannel[0].Seat + ".tool." + VerbsTheControllerAsksTheOperatorChannel[0].Verb,
 		Reason:  "refused: the controller raises the secret-given condition as before (ADR 0380 §9)",
 	},
+	{
+		Subject: "$JS.ACK.EVENTS." + MergesConsumerName + ".>",
+		Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at " +
+			"its first push, which is when it acknowledges a merge (ADR 0407 decision 3, issue 973)",
+	},
 	{
 		Subject: "$KV." + MergeWindowBucket + ".>",
 		Reason: "refused: the controller closes a batch's window by a timer of its own for that batch, as the " +
@@ -640,7 +645,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		// delivery from EVENTS — a consumer that cannot ack has every message redelivered for
 		// ever, refused by the list it already has.
 		sub = append(sub, ControllerFollows...)
-		pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>")
+		pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>")
 
 		// **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That
 		// permits one reply to the reply subject of a message the user received — and a message a

internal/broker/streams.go

@@ -233,6 +233,18 @@ const ProvAppliedStream = "PROV_APPLIED"
 // all (ProvisionerAppliedSubject).
 const ProvAppliedSubjects = "prov.applied.>"
 
+// eventsTheControllerFollows are the events the controller's events consumer takes: all it follows but the forge's
+// merges, which the merges consumer takes (MergesConsumerName).
+func eventsTheControllerFollows() []string {
+	var out []string
+	for _, subject := range ControllerFollows {
+		if subject != ControllerFollows[3] {
+			out = append(out, subject)
+		}
+	}
+	return out
+}
+
 // DeliverSubjectFor is where a push consumer's messages land.
 //
 // **Per consumer, which means per stream as well as per name** (novox/hq 04-ISSUES/146). A push
@@ -248,6 +260,9 @@ const ProvAppliedSubjects = "prov.applied.>"
 // durable's name to its stream, and this subject was the one place that scoping was dropped. It
 // stays inside what a controller may already subscribe (`_DELIVER.controller.>`).
 func DeliverSubjectFor(c Consumer) string {
+	if c.DeliverSubject != "" {
+		return c.DeliverSubject
+	}
 	return "_DELIVER." + c.Name + "." + c.Stream
 }
 
@@ -317,6 +332,13 @@ func Overlaps() []string {
 // ack subject is derived from (nats.go: `$JS.ACK.<stream>.controller.>`).
 const ControllerName = "controller"
 
+// MergesConsumerName is the controller's durable consumer of the forge's merged event on EVENTS, apart from the
+// events consumer (novox/hq ADR 0407 decision 3, issue 973). A merge is never dropped, so this consumer has no
+// max-deliver, which would send a merge held five times to DEAD_LETTERS to wait for a person, and no healer
+// resets it, which would drop the merges it holds. A consumer of its own because both of those are properties of
+// a consumer, not of a subject.
+const MergesConsumerName = "controller-merges"
+
 // ControllerSeat is the role the control plane holds, and ControllerStates are the facts it states
 // under it (novox/hq ADR 0134).
 //
@@ -453,7 +475,7 @@ func MeshConsumers() []Consumer {
 		{
 			Name:           ControllerName,
 			Stream:         "EVENTS",
-			Filters:        append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName)),
+			Filters:        append(eventsTheControllerFollows(), AgainFilter(ControllerName)),
 			Push:           true,
 			AckWaitSeconds: 30,
 			MaxDeliver:     5,
@@ -463,16 +485,32 @@ func MeshConsumers() []Consumer {
 			MaxAckPending: 1,
 			FromNow:       true,
 			// **The one consumer the mesh resets by itself** (healer H4, issue 248): it fell a week
-			// behind once and held every merge after it. What a reset drops is caught up elsewhere —
-			// a merge by the catch-up pass that reads the forge (issue 266), a build's outcome from the
-			// build records a plan settles from (issue 214), a provider's failing word by the provider
-			// saying it again every quarter of an hour (ADR 0224).
-			Resettable: "what it drops is caught up: merges by the catch-up pass (issue 266), build outcomes " +
-				"from the build records (issue 214), a provider's failing word said again (ADR 0224)",
+			// behind once and held every merge after it. What a reset drops is caught up elsewhere — a
+			// build's outcome from the build records a plan settles from (issue 214), a provider's failing
+			// word by the provider saying it again every quarter of an hour (ADR 0224). **Not the merges**:
+			// they are the merges consumer's below, which a reset leaves alone (ADR 0407 decision 3).
+			Resettable: "what it drops is caught up: build outcomes from the build records (issue 214), a " +
+				"provider's failing word said again (ADR 0224); the forge's merges are not its, they have a " +
+				"consumer of their own that no reset touches",
 			Why: "the events the mesh's own controller reacts to, one at a time; after " +
 				"max-deliver it gives the event up, and the controller keeps it in DEAD_LETTERS until " +
 				"a person delivers it again or drops it",
 		},
+		// The forge's merges (novox/hq ADR 0407 decision 3, issue 973): never dropped, so no max-deliver and no
+		// reset; a merge the controller cannot record yet is held with a growing delay until it can.
+		{
+			Name:   MergesConsumerName,
+			Stream: "EVENTS",
+			// Within what the controller may subscribe already (`_DELIVER.controller.>`).
+			DeliverSubject: "_DELIVER." + ControllerName + ".merges",
+			Filters:        []string{ControllerFollows[3]},
+			Push:           true,
+			AckWaitSeconds: 30,
+			MaxAckPending:  1,
+			FromNow:        true,
+			Why: "the forge's merges, one at a time; no max-deliver and never reset, because a merge is never " +
+				"dropped and no person releases one",
+		},
 		// What the server said a consumer gave up on (novox/hq issue 330): copied into DEAD_LETTERS and
 		// acknowledged. No max-deliver: a notice the controller could not copy is offered again, and said.
 		{

internal/broker/testdata/composed.conf

@@ -24,7 +24,7 @@ accounts {
     jetstream: enabled
     users = [
       { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
-        publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
+        publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] }
         subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }

internal/link/advisories.go

@@ -145,7 +145,7 @@ func MeshNamed(stream, name string) bool {
 		return true
 	case strings.HasPrefix(stream, "SEAT_"):
 		return strings.HasSuffix(name, "_worker")
-	case name == broker.ControllerName:
+	case name == broker.ControllerName || name == broker.MergesConsumerName:
 		return true
 	case stream == broker.EventsStream:
 		return strings.Contains(name, "_")
@@ -156,7 +156,7 @@ func MeshNamed(stream, name string) bool {
 // ConsumerInWords is a durable consumer as the mesh says it: whose, and for what.
 func ConsumerInWords(stream, name string) string {
 	switch {
-	case name == broker.ControllerName:
+	case name == broker.ControllerName || name == broker.MergesConsumerName:
 		return "the controller's consumer on " + stream
 	case stream == "NODES":
 		return "how " + name + " hears what it should be (its declaration consumer)"

internal/link/receive_nats.go

@@ -148,6 +148,24 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
 		defer func() { _ = followed.Unsubscribe() }()
 	}
 
+	// And the forge's merges, off a consumer of their own: no max-deliver and no reset, so a merge held for the
+	// plans waits for them however long (novox/hq ADR 0407 decision 3). Bound to the same channel: one loop.
+	if n.follows[KindSourceMoved] {
+		if events == nil {
+			events = make(chan *nats.Msg, Prefetch)
+		}
+		merges, err := standingBy(ctx, log.Default(), "EVENTS", func() (*nats.Subscription, error) {
+			return js.ChanSubscribe(MergedSubject, events, nats.Bind("EVENTS", broker.MergesConsumerName))
+		})
+		if err != nil {
+			return fmt.Errorf("subscribing to the forge's merges: %w", err)
+		}
+		if merges == nil {
+			return nil
+		}
+		defer func() { _ = merges.Unsubscribe() }()
+	}
+
 	// A connection that dropped is said, not discovered. A controller whose bus connection is gone
 	// is a mesh where nothing can be told anything.
 	gone := make(chan error, 1)

internal/link/receive_nats_test.go

@@ -590,3 +590,77 @@ func TestNatsAReportThatIsNotNewsIsNotStated(t *testing.T) {
 type sameAgain struct{}
 
 func (sameAgain) Heard(context.Context, Report) (bool, error) { return false, nil }
+
+// manyTimesBusy is an upgrader whose plans are busy for its first merges, as another controller's hold is.
+type manyTimesBusy struct {
+	toldAbout
+	busy  int
+	calls int
+}
+
+func (u *manyTimesBusy) SourceMoved(context.Context, SourceMoved) error {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	u.calls++
+	if u.calls <= u.busy {
+		return errors.Join(ErrTryAgain, errors.New("another controller is working the plans"))
+	}
+	return nil
+}
+
+func (u *manyTimesBusy) called() int {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	return u.calls
+}
+
+// **A merge held more than five times is still handed over and recorded** (issue 973, novox/hq ADR 0407 decision
+// 3): the merges have a durable consumer of their own with no max-deliver, so they never go to DEAD_LETTERS to wait
+// for a person; the delay grows, and is capped.
+func TestNatsAMergeHeldMoreThanFiveTimesIsStillHandedOverAndRecorded(t *testing.T) {
+	first, most := MergeHeldFirst, MergeHeldMost
+	MergeHeldFirst, MergeHeldMost = 5*time.Millisecond, 40*time.Millisecond
+	t.Cleanup(func() { MergeHeldFirst, MergeHeldMost = first, most })
+	js := aBus(t)
+	up := &manyTimesBusy{busy: 8}
+	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
+	if err := s.Follows(up); err != nil {
+		t.Fatal(err)
+	}
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go func() { _ = s.Serve(ctx) }()
+	eventually(t, "the merges consumer being made", func() bool {
+		_, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
+		return err == nil
+	})
+
+	body, _ := json.Marshal(SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"})
+	if _, err := js.Context().Publish(MergedSubject, body); err != nil {
+		t.Fatal(err)
+	}
+	eventually(t, "the ninth hand-over recording the merge", func() bool { return up.called() == 9 })
+	eventually(t, "the merge being acknowledged", func() bool {
+		info, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
+		return err == nil && info.NumAckPending == 0 && info.NumPending == 0
+	})
+	if info, err := js.Context().StreamInfo(broker.DeadLettersStream); err == nil && info.State.Msgs != 0 {
+		t.Fatalf("%d message(s) in dead letters: a merge was given up on", info.State.Msgs)
+	}
+}
+
+// The delay a held merge waits grows from the first to the most and stays there.
+func TestAHeldMergesDelayGrowsAndIsCapped(t *testing.T) {
+	var got []time.Duration
+	for _, held := range []time.Duration{0, 4 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
+		200 * time.Second, 20 * time.Minute} {
+		got = append(got, MergeRedeliverAfter(held))
+	}
+	want := []time.Duration{5 * time.Second, 5 * time.Second, 10 * time.Second, 25 * time.Second, 70 * time.Second,
+		200 * time.Second, 5 * time.Minute}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Errorf("held %d: waits %v; want %v", i, got[i], want[i])
+		}
+	}
+}

internal/link/serve.go

@@ -636,8 +636,19 @@ func short(commit string) string {
 	return commit
 }
 
-// mergeHeldFor is how long a merge the controller could not record yet waits before the bus hands it over again.
-const mergeHeldFor = 5 * time.Second
+// MergeHeldFirst is how long a merge the controller could not record yet waits before the bus hands it over
+// again, and MergeHeldMost the longest it ever waits: the delay grows with how long the merge has been held, from
+// the first to the most (novox/hq ADR 0407 decision 3). Variables so a test need not wait.
+var (
+	MergeHeldFirst = 5 * time.Second
+	MergeHeldMost  = 5 * time.Minute
+)
+
+// MergeRedeliverAfter is the delay for a merge held for the time given: that time, so the delay about doubles
+// with each hold, between MergeHeldFirst and MergeHeldMost.
+func MergeRedeliverAfter(heldFor time.Duration) time.Duration {
+	return min(max(heldFor, MergeHeldFirst), MergeHeldMost)
+}
 
 // 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
@@ -663,13 +674,14 @@ func (s *Server) sourceMoved(ctx context.Context, m Control) {
 		if ctx.Err() != nil {
 			return
 		}
-		if held := m.Hold(mergeHeldFor); held != nil {
+		delay := MergeRedeliverAfter(m.HeldFor())
+		if held := m.Hold(delay); held != nil {
 			s.log.Printf("%s/%s moved to %.8s and could not be held (%v): %v; left for the bus to hand over again",
 				moved.Owner, moved.Repo, moved.Commit, held, err)
 			return
 		}
 		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)
+			moved.Commit, err, delay)
 		return
 	}
 	if err != nil {