From 4e21bae65a173d2fc3b3c0eed407a10193c81e78 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 13 Nov 2026 11:36:07 +0100 Subject: [PATCH] S1 sdk-stream-last-per-subject: a stream may declare last-per-subject MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §703, §50 (A-343, A-344) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/catalogue/streams.go | 5 ++++- go/catalogue/streams_test.go | 15 +++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/go/catalogue/streams.go b/go/catalogue/streams.go index b9f9565c..556406ee 100644 --- a/go/catalogue/streams.go +++ b/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 diff --git a/go/catalogue/streams_test.go b/go/catalogue/streams_test.go index 6a5cd52b..d2e4fcd1 100644 --- a/go/catalogue/streams_test.go +++ b/go/catalogue/streams_test.go @@ -152,3 +152,18 @@ func TestNarrowingTwoPatterns(t *testing.T) { t.Fatalf("uncovered: %v", got) } } + +// A stream declares that it keeps the last message of each subject (design 47 G §703), says no unless it does, and a stream +// without it reads as before. +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 || m.UnknownField() != "" { + t.Fatalf("last-per-subject read as %+v (set aside %q)", m.Streams, m.UnknownField()) + } + if !slices.Contains(ManifestReads(), "word:streams.last-per-subject") { + t.Fatal("ManifestReads does not state word:streams.last-per-subject") + } +}