From df3b732ecae9c3af8283bebae362abd648b6db9f Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:09:56 +0100 Subject: [PATCH 1/3] MV1: move contracts and sdk under modules/, importers follow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The delivery repository's contracts and SDK live under modules/ as mesh-delivery-contracts and mesh-delivery-sdk, their Go module paths renamed to match the folders (the TypeScript names already did), and every importer in mesh-delivery, mesh-catalog, mesh-host, mesh-tools and mesh-sdk follows in the same group: a design as it should be, one path per module, checked by the group's check building every importer. Builds: design 47 G §427, §88 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/packages_list_test.go | 18 +++++++++--------- go/seatlist/bootstrap_contracts_test.go | 6 +++--- .../seats/mesh-catalog/seat.json.captured | 0 .../mesh-delivery-checker/seat.json.captured | 0 4 files changed, 12 insertions(+), 12 deletions(-) rename go/testdata/beside/mesh-delivery/{contracts => modules/mesh-delivery-contracts}/seats/mesh-catalog/seat.json.captured (100%) rename go/testdata/beside/mesh-delivery/{contracts => modules/mesh-delivery-contracts}/seats/mesh-delivery-checker/seat.json.captured (100%) diff --git a/go/packages_list_test.go b/go/packages_list_test.go index f246289f..7abf51dd 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -359,15 +359,15 @@ const sdkModule = "git.novox.be/novox/mesh-sdk/go" // that is a folder module of a repository (mesh-networking's modules/mesh-connectivity-contracts) is refused too. func TestContractsAndDomainSDKsAreRecognisedInEverySegment(t *testing.T) { for path, want := range map[string]bool{ - "git.novox.be/novox/mesh-contracts/go/seats/meshconfig": true, - "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog": true, - "git.novox.be/novox/mesh-delivery/sdk": true, - "git.novox.be/novox/mesh-networking/modules/mesh-connectivity-contracts/go/seats/x": true, - "git.novox.be/novox/mesh-networking/modules/mesh-connectivity-sdk": true, - "git.novox.be/novox/mesh-sdk/go/catalogue": false, - "git.novox.be/novox/mesh-sdk/go/seatlist": false, - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer": false, - "github.com/nats-io/nats.go": false, + "git.novox.be/novox/mesh-contracts/go/seats/meshconfig": true, + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog": true, + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk": true, + "git.novox.be/novox/mesh-networking/modules/mesh-connectivity-contracts/go/seats/x": true, + "git.novox.be/novox/mesh-networking/modules/mesh-connectivity-sdk": true, + "git.novox.be/novox/mesh-sdk/go/catalogue": false, + "git.novox.be/novox/mesh-sdk/go/seatlist": false, + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer": false, + "github.com/nats-io/nats.go": false, } { if got := contractsOrDomainSDK(path); got != want { t.Errorf("contractsOrDomainSDK(%q) = %v, want %v", path, got, want) diff --git a/go/seatlist/bootstrap_contracts_test.go b/go/seatlist/bootstrap_contracts_test.go index affbdc6e..b14290a1 100644 --- a/go/seatlist/bootstrap_contracts_test.go +++ b/go/seatlist/bootstrap_contracts_test.go @@ -12,7 +12,7 @@ import ( // ADR 0635 §3, ADR 0608 §2: a bootstrapped definition is a copy of its seat's file in its contracts module, byte for // byte, and a seat with a copy has a row here. The contracts are read beside this repository, as a merge check places -// them (mesh-contracts, and mesh-delivery for its `contracts` folder) or, elsewhere, from the copy captured in +// them (mesh-contracts, and mesh-delivery for its `modules/mesh-delivery-contracts` folder) or, elsewhere, from the copy captured in // testdata/beside. func TestEveryBootstrappedDefinitionEqualsItsContractsFile(t *testing.T) { rows := []struct{ seat, repository, file string }{ @@ -24,8 +24,8 @@ func TestEveryBootstrappedDefinitionEqualsItsContractsFile(t *testing.T) { {"mesh-hand-acts", "mesh-contracts", "health-and-repair/seats/mesh-hand-acts/seat.json"}, {"mesh-watchdogs", "mesh-contracts", "health-and-repair/seats/mesh-watchdogs/seat.json"}, {"operator-channel", "mesh-contracts", "operator-and-conversation/seats/operator-channel/seat.json"}, - {"mesh-catalog", "mesh-delivery", "contracts/seats/mesh-catalog/seat.json"}, - {"mesh-delivery-checker", "mesh-delivery", "contracts/seats/mesh-delivery-checker/seat.json"}, + {"mesh-catalog", "mesh-delivery", "modules/mesh-delivery-contracts/seats/mesh-catalog/seat.json"}, + {"mesh-delivery-checker", "mesh-delivery", "modules/mesh-delivery-contracts/seats/mesh-delivery-checker/seat.json"}, } var have []string for _, d := range Bootstrapped() { diff --git a/go/testdata/beside/mesh-delivery/contracts/seats/mesh-catalog/seat.json.captured b/go/testdata/beside/mesh-delivery/modules/mesh-delivery-contracts/seats/mesh-catalog/seat.json.captured similarity index 100% rename from go/testdata/beside/mesh-delivery/contracts/seats/mesh-catalog/seat.json.captured rename to go/testdata/beside/mesh-delivery/modules/mesh-delivery-contracts/seats/mesh-catalog/seat.json.captured diff --git a/go/testdata/beside/mesh-delivery/contracts/seats/mesh-delivery-checker/seat.json.captured b/go/testdata/beside/mesh-delivery/modules/mesh-delivery-contracts/seats/mesh-delivery-checker/seat.json.captured similarity index 100% rename from go/testdata/beside/mesh-delivery/contracts/seats/mesh-delivery-checker/seat.json.captured rename to go/testdata/beside/mesh-delivery/modules/mesh-delivery-contracts/seats/mesh-delivery-checker/seat.json.captured From 631654eec62f7912934481eb5eb8d9e1ac660fe4 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:40:57 +0100 Subject: [PATCH 2/3] A-176: a helper writes a durable consumer's facts into the module's state consumers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit process.ConsumerLedger writes stream, name, acknowledged sequence, pending, last event applied with its at and the delivery of the oldest unacknowledged message into one key per consumer, after each message delivered and applied, so a tool reads them. Builds: design 47 G §579, A-166, A-176 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/process/consumers.go | 122 +++++++++++++++++++++++++++++++++++ go/process/consumers_test.go | 52 +++++++++++++++ 2 files changed, 174 insertions(+) create mode 100644 go/process/consumers.go create mode 100644 go/process/consumers_test.go diff --git a/go/process/consumers.go b/go/process/consumers.go new file mode 100644 index 00000000..80ffb536 --- /dev/null +++ b/go/process/consumers.go @@ -0,0 +1,122 @@ +package process + +import ( + "context" + "encoding/json" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// ConsumersState is the name of the state a module keeps its durable consumers' facts in, one key per consumer by its name, +// history 1 (design 47 G A-166, A-176): what a module's tool reads of a consumer, since a tool runs in a runtime of its own and +// cannot read the consumer's information. +const ConsumersState = "consumers" + +// ConsumerFacts is one durable consumer as ConsumerLedger keeps it. +type ConsumerFacts struct { + // Stream and Name are the consumer's stream and its name. + Stream string `json:"stream"` + Name string `json:"name"` + // Acknowledged is the stream sequence of the last message applied. + Acknowledged uint64 `json:"acknowledged"` + // Pending is the messages the consumer has not been given yet, as the last message said. + Pending uint64 `json:"pending"` + // LastEvent is the subject of the last event applied and LastAt its own `at`, or when the bus stored it where it has none. + LastEvent string `json:"last-event,omitempty"` + LastAt time.Time `json:"last-at,omitempty"` + // OldestUnacknowledged is when the oldest message delivered and not yet applied was delivered; zero when none is. + OldestUnacknowledged time.Time `json:"oldest-unacknowledged,omitempty"` +} + +// ConsumerSink stores one key of the module's state consumers. +type ConsumerSink func(ctx context.Context, key string, value []byte) error + +// ConsumerLedger writes a durable consumer's facts into the module's state consumers after each message delivered and each message +// applied, so a tool reads them and nothing is read on a timer (design 47 G A-166, A-176). One ledger serves one consumer. +type ConsumerLedger struct { + stream, name string + sink ConsumerSink + now func() time.Time + logf func(string, ...any) + + mu sync.Mutex + facts ConsumerFacts + unacks map[uint64]time.Time // delivered and not applied, by stream sequence +} + +// NewConsumerLedger is the ledger of the consumer name on stream, writing through sink; a nil now is the clock and a nil logf says +// nothing. +func NewConsumerLedger(stream, name string, sink ConsumerSink, now func() time.Time, logf func(string, ...any)) *ConsumerLedger { + if now == nil { + now = time.Now + } + if logf == nil { + logf = func(string, ...any) {} + } + return &ConsumerLedger{stream: stream, name: name, sink: sink, now: now, logf: logf, unacks: map[uint64]time.Time{}, + facts: ConsumerFacts{Stream: stream, Name: name}} +} + +// Delivered notes that m was given to the consumer, and writes the facts. +func (l *ConsumerLedger) Delivered(m jetstream.Msg) { + md, err := m.Metadata() + if err != nil { + l.logf("consumers: %s's message has no metadata, its delivery is not kept: %v", l.name, err) + return + } + l.mu.Lock() + l.unacks[md.Sequence.Stream] = l.now().UTC() + l.facts.Pending = md.NumPending + l.mu.Unlock() + l.write() +} + +// Applied notes that m was applied and acknowledged, at the event's own `at` (zero: when the bus stored it), and writes the facts. +func (l *ConsumerLedger) Applied(m jetstream.Msg, at time.Time) { + md, err := m.Metadata() + if err != nil { + l.logf("consumers: %s's message has no metadata, its application is not kept: %v", l.name, err) + return + } + if at.IsZero() { + at = md.Timestamp + } + l.mu.Lock() + delete(l.unacks, md.Sequence.Stream) + if md.Sequence.Stream > l.facts.Acknowledged { + l.facts.Acknowledged = md.Sequence.Stream + } + l.facts.Pending = md.NumPending + l.facts.LastEvent, l.facts.LastAt = m.Subject(), at.UTC() + l.mu.Unlock() + l.write() +} + +// Facts are the facts now. +func (l *ConsumerLedger) Facts() ConsumerFacts { + l.mu.Lock() + defer l.mu.Unlock() + f := l.facts + f.OldestUnacknowledged = time.Time{} + for _, t := range l.unacks { + if f.OldestUnacknowledged.IsZero() || t.Before(f.OldestUnacknowledged) { + f.OldestUnacknowledged = t + } + } + return f +} + +func (l *ConsumerLedger) write() { + if l.sink == nil { + return + } + data, err := json.Marshal(l.Facts()) + if err == nil { + err = l.sink(context.Background(), l.name, data) + } + if err != nil { + l.logf("consumers: the facts of %s are not written to the state %s: %v", l.name, ConsumersState, err) + } +} diff --git a/go/process/consumers_test.go b/go/process/consumers_test.go new file mode 100644 index 00000000..5f5fb9ce --- /dev/null +++ b/go/process/consumers_test.go @@ -0,0 +1,52 @@ +package process + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +type ledgerMsg struct { + jetstream.Msg + subject string + seq uint64 + pending uint64 + stored time.Time +} + +func (m ledgerMsg) Subject() string { return m.subject } +func (m ledgerMsg) Metadata() (*jetstream.MsgMetadata, error) { + return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq, Consumer: m.seq}, NumPending: m.pending, Timestamp: m.stored}, nil +} + +func TestConsumerLedgerKeepsTheFactsAfterEachMessage(t *testing.T) { + kept := map[string][]byte{} + now := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + l := NewConsumerLedger("EVENTS", "n1_m", func(_ context.Context, key string, value []byte) error { kept[key] = value; return nil }, func() time.Time { return now }, nil) + read := func() ConsumerFacts { + var f ConsumerFacts + if err := json.Unmarshal(kept["n1_m"], &f); err != nil { + t.Fatal(err) + } + return f + } + a, b := ledgerMsg{subject: "mesh.seat.s.event.a", seq: 7, pending: 1}, ledgerMsg{subject: "mesh.seat.s.event.b", seq: 8, pending: 0} + l.Delivered(a) + now = now.Add(time.Minute) + l.Delivered(b) + if f := read(); f.Pending != 0 || !f.OldestUnacknowledged.Equal(now.Add(-time.Minute)) { + t.Errorf("after two deliveries: %+v, want pending 0 and the first one's delivery as the oldest unacknowledged", f) + } + at := now.Add(-time.Hour) + l.Applied(a, at) + if f := read(); f.Acknowledged != 7 || f.LastEvent != "mesh.seat.s.event.a" || !f.LastAt.Equal(at) || !f.OldestUnacknowledged.Equal(now) { + t.Errorf("after the first applied: %+v, want sequence 7, its event and at, and the second one's delivery as the oldest", f) + } + l.Applied(b, time.Time{}) + if f := read(); f.Acknowledged != 8 || !f.OldestUnacknowledged.IsZero() { + t.Errorf("after both applied: %+v, want sequence 8 and none unacknowledged", f) + } +} From 7b9b7e65ecf64a50fa960c5fe199944c1f37dd18 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:53:02 +0100 Subject: [PATCH 3/3] A-181 (sdk): a manifest stream says last-per-subject, and a part hears another module's stream so, read whole at its start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stream entry takes `last-per-subject`, and process.Account.HearStream reads a module's own stream kept so whole through an ordered consumer before following its durable, as HearSeatEvents does for a seat's event. Builds: design 47 G §703, §556, A-181 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/catalogue/streams.go | 5 ++- go/catalogue/streams_test.go | 11 +++++ go/process/event_hub_integration_test.go | 51 ++++++++++++++++++++++++ go/process/seatstream.go | 7 ++++ 4 files changed, 73 insertions(+), 1 deletion(-) diff --git a/go/catalogue/streams.go b/go/catalogue/streams.go index 609fac94..74ec969c 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 66ee46a2..f51218b4 100644 --- a/go/catalogue/streams_test.go +++ b/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) + } +} diff --git a/go/process/event_hub_integration_test.go b/go/process/event_hub_integration_test.go index 749928a9..73e84576 100644 --- a/go/process/event_hub_integration_test.go +++ b/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) } diff --git a/go/process/seatstream.go b/go/process/seatstream.go index 7da0ebc8..9dffa21f 100644 --- a/go/process/seatstream.go +++ b/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 is %s_, "+ "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) }