Merge commit '7b9b7e6' into feat/47-G-fix11-b1e

go/catalogue/streams.go

@@ -64,6 +64,9 @@ type StreamDeclaration struct {
 	// Duplicates is a work-queue stream's duplicate window in hours, at most its maximum age; refused
 	// without retention work-queue (novox/hq ADR 0436 §6).
 	Duplicates int `json:"duplicates,omitempty"`
+	// LastPerSubject says the stream keeps only the last message of each subject, as a seat event's `keep` does (design 47 G §703,
+	// §396, A-181): a part that follows it reads it whole at each start (§556).
+	LastPerSubject bool `json:"last-per-subject,omitempty"`
 	// KeptLayout is the layout its holder says it writes: its form and its catch-up tool (design 47).
 	KeptLayout
 }
@@ -86,7 +89,7 @@ func (s *StreamDeclaration) UnmarshalJSON(raw []byte) error {
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&p); err != nil {
 		return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant, retention, max-deliveries, "+
-			"duplicates, form, catch-up}: %w", typedUnknown(err))
+			"duplicates, last-per-subject, form, catch-up}: %w", typedUnknown(err))
 	}
 	if err := p.KeptLayout.refuse(fmt.Sprintf("stream %q", p.Name)); err != nil {
 		return err

go/catalogue/streams_test.go

@@ -182,3 +182,14 @@ func subAStarEmitterHearsEachDeclaringModulesUnrestrictedStream(t *testing.T) {
 		t.Fatalf("a wildcard reached %+v", parts)
 	}
 }
+
+// A stream declares that it keeps the last message of each subject (design 47 G §703, A-181), and says no unless it does.
+func TestAStreamDeclaresItIsKeptLastPerSubject(t *testing.T) {
+	m, err := decodeForTest([]byte(`{"module":"feed","version":"1","streams":[{"name":"ends","events":["end.**"],"last-per-subject":true},{"name":"all","events":["a.**"]}]}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !m.Streams[0].LastPerSubject || m.Streams[1].LastPerSubject {
+		t.Fatalf("last-per-subject read as %+v", m.Streams)
+	}
+}

go/process/event_hub_integration_test.go

@@ -142,8 +142,59 @@ func subAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
 	}
 }
 
+// A-181: a module hears another module's own stream kept last per subject (a manifest's `streams` entry with `last-per-subject`) as
+// it hears a seat's: the last message of each subject read whole at its start, though its durable began after them, then the durable.
+func subAModuleHearsAnotherModulesStreamKeptLastPerSubject(t *testing.T) {
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	stream, subject := "BUILDER_BUILDS", process.ModuleStreamSubject("builder", "build.ended.>")
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range []struct{ event, v string }{{"build.ended.web.fp1", "old"}, {"build.ended.web.fp1", "new"}, {"build.ended.web.fp2", "two"}} {
+		if _, err := js.Publish(ctx, process.ModuleStreamSubject("builder", m.event), []byte(m.v)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	cred := process.Account{Node: "ace", Module: "assigner"}
+	durable := cred.SeatEventConsumerConfig([]string{subject}, true)
+	durable.DeliverPolicy = jetstream.DeliverNewPolicy
+	if _, err := js.CreateOrUpdateConsumer(ctx, stream, durable); err != nil {
+		t.Fatal(err)
+	}
+	got := make(chan string, 4)
+	var reads []string
+	cc, err := cred.HearStream(ctx, js, stream, func(m jetstream.Msg) { got <- string(m.Data()); _ = m.Ack() }, t.Logf, process.OnRead(func(s string) { reads = append(reads, s) }))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cc.Stop()
+	heard := map[string]bool{}
+	for len(heard) < 2 {
+		select {
+		case h := <-got:
+			heard[h] = true
+		case <-ctx.Done():
+			t.Fatalf("heard %v", heard)
+		}
+	}
+	if !heard["new"] || !heard["two"] || heard["old"] || len(reads) != 2 {
+		t.Errorf("heard %v with reads %v, want the last of each subject, read whole", heard, reads)
+	}
+}
+
 // TestTheHubBindsALateConsumerAndSaysItIsLive runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
 func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) {
 	t.Run("TheHubBindsALateConsumerAndSaysItIsLive", subTheHubBindsALateConsumerAndSaysItIsLive)
 	t.Run("AModuleHearsASeatEventKeptOnItsOwnStream", subAModuleHearsASeatEventKeptOnItsOwnStream)
+	t.Run("AModuleHearsAnotherModulesStreamKeptLastPerSubject", subAModuleHearsAnotherModulesStreamKeptLastPerSubject)
 }

go/process/seatstream.go

@@ -97,6 +97,13 @@ func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, str
 		return nil, fmt.Errorf("%q is not the bus's name of a seat event's stream: a contract's Stream<E> is %s<seat>_<event>, "+
 			"with no dot (seatdef.KeptStream)", stream, seatdef.KeptStreamPrefix)
 	}
+	return a.HearStream(ctx, js, stream, handle, logf, opts...)
+}
+
+// HearStream is HearSeatEvents for the stream of a module's own events (a manifest's `streams` entry, named as
+// catalogue.StreamBusName names it), which the module is granted through its `consumes`: the same one rule, a stream the bus's own
+// info says is kept last per subject (`last-per-subject`) read whole first, then the durable followed (design 47 G §556, §703, A-181).
+func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any), opts ...HearOption) (jetstream.ConsumeContext, error) {
 	if handle == nil {
 		return nil, fmt.Errorf("hearing %s needs a handler", stream)
 	}