diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 1cea4226..65dcd948 100644 --- a/internal/broker/controller_buckets_test.go +++ b/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) + } +} diff --git a/internal/broker/derived.go b/internal/broker/derived.go index 899ead78..c724b439 100644 --- a/internal/broker/derived.go +++ b/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. diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 495c5290..f709ea5f 100644 --- a/internal/broker/jetstream.go +++ b/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 diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 17e1b080..f3384f70 100644 --- a/internal/broker/streams.go +++ b/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. {