Make the merges consumer before the events consumer narrows, from the old one's ack floor, so no merge is lost to the change (issue 973)

internal/broker/controller_buckets_test.go

@@ -2,6 +2,7 @@ package broker
 
 import (
 	"context"
+	"github.com/nats-io/nats.go"
 	"slices"
 	"strings"
 	"testing"
@@ -368,3 +369,33 @@ func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testin
 		t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err)
 	}
 }
+
+// **Delivering this change loses no merge** (review of issue 973): a merge published under the old filter of the
+// `controller` consumer and never acknowledged by it is delivered by the merges consumer once the assertion has
+// made it, because it starts where `controller` has acknowledged up to, not from now.
+func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(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)
+	}
+	old := &nats.ConsumerConfig{Durable: ControllerName, AckPolicy: nats.AckExplicitPolicy, AckWait: 30 * time.Second,
+		MaxDeliver: 5, MaxAckPending: 1, DeliverSubject: "_DELIVER.controller.EVENTS", DeliverPolicy: nats.DeliverNewPolicy,
+		FilterSubjects: append(append([]string(nil), ControllerFollows...), AgainFilter(ControllerName))}
+	if _, err := js.Context().AddConsumer(EventsStream, old); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.Context().Publish(ControllerFollows[3], []byte(`{"commit":"c1"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if err := AssertMeshConsumers(js); 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 the old consumer never acknowledged", merges, err)
+	}
+}

internal/broker/derived.go

@@ -36,6 +36,10 @@ 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
+	// TakesOverFrom names a consumer of the same stream whose filter this one takes over (issue 973): made while
+	// that consumer still filters this one's subject, it starts after what that consumer has acknowledged, so a
+	// message the old one held unacknowledged is not lost to the change. Otherwise it starts as FromNow says.
+	TakesOverFrom string
 	// DeliverSubject, when set, is where a push consumer delivers instead of DeliverSubjectFor's.
 	DeliverSubject string
 	// AckWaitSeconds before an unacknowledged delivery is redelivered.

internal/broker/jetstream.go

@@ -343,6 +343,9 @@ func (j *JetStream) EnsureConsumer(c Consumer) error {
 		if c.FromNow {
 			want.DeliverPolicy = nats.DeliverNewPolicy
 		}
+		if from := j.ackFloorOfTheConsumerItTakesOver(c); from > 0 {
+			want.DeliverPolicy, want.OptStartSeq = nats.DeliverByStartSequencePolicy, from
+		}
 		if _, err := j.js.AddConsumer(c.Stream, want); err != nil {
 			return fmt.Errorf("creating consumer %s on %s: %w", c.Name, c.Stream, err)
 		}
@@ -352,6 +355,26 @@ func (j *JetStream) EnsureConsumer(c Consumer) error {
 	}
 }
 
+// ackFloorOfTheConsumerItTakesOver is the stream sequence a consumer made now starts at so as to lose nothing the
+// consumer it takes over from (Consumer.TakesOverFrom) has not acknowledged: that consumer's ack floor plus one, when
+// it exists and still filters this one's subject; zero otherwise.
+func (j *JetStream) ackFloorOfTheConsumerItTakesOver(c Consumer) uint64 {
+	if c.TakesOverFrom == "" || len(c.Filters) != 1 {
+		return 0
+	}
+	have, err := j.js.ConsumerInfo(c.Stream, c.TakesOverFrom)
+	if err != nil {
+		return 0
+	}
+	filters := append([]string{have.Config.FilterSubject}, have.Config.FilterSubjects...)
+	for _, f := range filters {
+		if f == c.Filters[0] {
+			return have.AckFloor.Stream + 1
+		}
+	}
+	return 0
+}
+
 // ConsumerState is what a reset says about a consumer, before and after.
 type ConsumerState struct {
 	DeliverPolicy string

internal/broker/streams.go

@@ -472,6 +472,23 @@ func MeshConsumers() []Consumer {
 			Why: "the controller is the single consumer of what nodes say; explicit ack and no " +
 				"max-deliver, because the store window's bound is the controller's own",
 		},
+		// 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. **Asserted before
+		// the events consumer**, which stops filtering merges: the order of this list is the order of assertion.
+		{
+			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,
+			TakesOverFrom:  ControllerName,
+			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",
+		},
 		{
 			Name:           ControllerName,
 			Stream:         "EVENTS",
@@ -496,21 +513,6 @@ func MeshConsumers() []Consumer {
 				"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.
 		{