go/catalogue/busstate.go

@@ -50,6 +50,9 @@ const (
 	OptionAgeDays = "max-age-days"
 	// OptionCapMiB is the cap of a stream, a bucket or an object store, in MiB.
 	OptionCapMiB = "max-mib"
+	// OptionObjectsCapMiB is the cap of a state's objects store `<state>-objects`, in MiB, where the state needs
+	// another than the one the bus module plans (design 47 G §513, §547).
+	OptionObjectsCapMiB = "objects-max-mib"
 	// OptionGrant names the trusted config value granting another module a part of it (ADR 0363 §4, ADR 0592 §9).
 	OptionGrant = "grant"
 	// OptionDeliveries is how many times a work queue's message is delivered before it is a dead letter.

go/catalogue/objects_test.go

@@ -53,3 +53,10 @@ func decodeStrictObjects(raw string) (Manifest, error) {
 type errString string
 
 func (e errString) Error() string { return string(e) }
+
+// The option naming a state's objects cap is exported with the bounds the judge checks it by (design 47 G §513, §547).
+func TestTheObjectsCapOptionIsNamedWithItsBounds(t *testing.T) {
+	if OptionObjectsCapMiB != "objects-max-mib" || ObjectsLeastMiB != 1 || ObjectsMostMiB != 8192 {
+		t.Fatalf("option %q, bounds %d to %d", OptionObjectsCapMiB, ObjectsLeastMiB, ObjectsMostMiB)
+	}
+}

go/catalogue/process_bus.go

@@ -19,6 +19,9 @@ type ProcessBus struct {
 	// Emits are the events this process emits, named locally, as the module's own Emits names them
 	// (EmitsAll): a process emits only what its module may emit as a whole.
 	Emits []string `json:"emits,omitempty"`
+	// Consumes are the events this process hears, each as its module's `consumes` names it: only an event its module
+	// consumes (design 47 G A-340, after design 42 zfs-owns-zfs V5 and G1).
+	Consumes []string `json:"consumes,omitempty"`
 	// Queues are the module's own work-queue streams (ADR 0436 §6) this process consumes.
 	Queues []string `json:"queues,omitempty"`
 	// State is the module's own state, by local name, this process reads and writes.
@@ -97,7 +100,7 @@ func (b *ProcessBus) UnmarshalJSON(raw []byte) error {
 	dec := json.NewDecoder(bytes.NewReader(raw))
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&p); err != nil {
-		return fmt.Errorf("a process's bus is {emits, queues, state, asks, invokes, keeps}: %w", typedUnknown(err))
+		return fmt.Errorf("a process's bus is {emits, consumes, queues, state, asks, invokes, keeps}: %w", typedUnknown(err))
 	}
 	*b = ProcessBus(p)
 	return nil

go/catalogue/process_bus_test.go

@@ -1,5 +1,10 @@
 package catalogue
 
+import (
+	"encoding/json"
+	"testing"
+)
+
 func strSliceEqual(a, b []string) bool {
 	if len(a) != len(b) {
 		return false
@@ -11,3 +16,20 @@ func strSliceEqual(a, b []string) bool {
 	}
 	return true
 }
+
+// Design 47 G A-340: a process's bus names the events it hears under `consumes`; a bus of today reads as today and an
+// unknown key is still refused.
+func TestAProcessBusMayNameWhatItConsumes(t *testing.T) {
+	var heard ProcessBus
+	if err := json.Unmarshal([]byte(`{"emits":["a"],"consumes":["mesh-conditions.findings-asked"]}`), &heard); err != nil ||
+		!strSliceEqual(heard.Consumes, []string{"mesh-conditions.findings-asked"}) || !strSliceEqual(heard.Emits, []string{"a"}) {
+		t.Errorf("a bus's consumes read as %v (%v)", heard.Consumes, err)
+	}
+	var today ProcessBus
+	if err := json.Unmarshal([]byte(`{"emits":["a"],"queues":["q"]}`), &today); err != nil || len(today.Consumes) != 0 {
+		t.Errorf("a bus of today read as %+v (%v)", today, err)
+	}
+	if err := json.Unmarshal([]byte(`{"hears":["x"]}`), &ProcessBus{}); err == nil {
+		t.Error("an unknown key was accepted")
+	}
+}

go/catalogue/seat_from_definition.go

@@ -26,10 +26,16 @@ func SeatFromDefinition(d seatdef.Definition) Seat {
 			Optional: v.Optional, Replaces: v.Replaces})
 	}
 	for _, st := range d.States {
+		s.States = append(s.States, st.Name)
 		if st.View {
 			s.Views = append(s.Views, st.Name)
 		}
 	}
+	for _, a := range d.Accepts {
+		if a.Stream != "" {
+			s.AcceptedOn = append(s.AcceptedOn, a.Stream+"."+a.Name)
+		}
+	}
 	for _, r := range d.Receives {
 		rv := Receivable{Kind: r.Kind, Comment: r.Comment, Dirs: r.Dirs, Offered: r.Offered}
 		if r.Shape != nil {

go/catalogue/seat_from_definition_test.go

@@ -0,0 +1,43 @@
+package catalogue
+
+import (
+	"slices"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/seatdef"
+)
+
+// A seat's compiled form lists the states its definition declares and the kinds it accepts on a stream (design 47 G A-102,
+// A-128); a seat without either lists none, as before.
+func TestSeatFromDefinitionListsItsStatesAndStreamKinds(t *testing.T) {
+	s := SeatFromDefinition(seatdef.Definition{Seat: "mesh-provisioner", Scope: "mesh",
+		States:  []seatdef.State{{Name: "plans", View: true}, {Name: "provisions"}},
+		Accepts: []seatdef.Message{{Name: "step", Stream: "check"}, {Name: "step"}}})
+	if !slices.Equal(s.States, []string{"plans", "provisions"}) || !slices.Equal(s.Views, []string{"plans"}) ||
+		!slices.Equal(s.AcceptedOn, []string{"check.step"}) {
+		t.Errorf("states = %v, views = %v, accepted on = %v", s.States, s.Views, s.AcceptedOn)
+	}
+	bare := SeatFromDefinition(seatdef.Definition{Seat: "x", Scope: "mesh"})
+	if bare.States != nil || bare.AcceptedOn != nil {
+		t.Errorf("a bare seat lists %v, %v", bare.States, bare.AcceptedOn)
+	}
+}
+
+// A queue invoke `seat:<seat>.accept.<queue>` is split by one helper (design 47 G A-231); a verb invoke is not one.
+func TestQueueInvoke(t *testing.T) {
+	for _, row := range []struct {
+		invoke, seat, queue string
+		ok                  bool
+	}{
+		{"seat:node-build-go.accept.check", "node-build-go", "check", true},
+		{"seat:x.verb", "", "", false},
+		{"seat:x.accept.a.b", "", "", false},
+		{"seat:x.accept.", "", "", false},
+		{"node-tools.x", "", "", false},
+	} {
+		seat, queue, ok := QueueInvoke(row.invoke)
+		if seat != row.seat || queue != row.queue || ok != row.ok {
+			t.Errorf("QueueInvoke(%q) = %q, %q, %v", row.invoke, seat, queue, ok)
+		}
+	}
+}

go/catalogue/seat_invokes.go

@@ -18,3 +18,18 @@ func SeatInvoke(invoke string) (seat, verb string, ok bool) {
 	seat, verb, ok = strings.Cut(rest, ".")
 	return seat, verb, ok && seat != "" && verb != ""
 }
+
+// QueueInvoke splits an invoke `seat:<seat>.accept.<queue>`, by which an asker names a build seat's queue it asks (design 47 G
+// §544, §601, A-231), into its seat and queue; ok is false for anything else. The one reader of that form: the manifest
+// validator, the indexer's served-verb edge and the grants part read it here.
+func QueueInvoke(invoke string) (seat, queue string, ok bool) {
+	seat, verb, ok := SeatInvoke(invoke)
+	if !ok {
+		return "", "", false
+	}
+	queue, isQueue := strings.CutPrefix(verb, "accept.")
+	if !isQueue || queue == "" || strings.Contains(queue, ".") {
+		return "", "", false
+	}
+	return seat, queue, true
+}

go/catalogue/seats.go

@@ -45,9 +45,15 @@ type Seat struct {
 	// Serves carries each verb in full — name, description, schema — because a role's tools are the
 	// mesh's to define and an agent's to call (novox/hq ADR 0132, design 33 §2).
 	Serves []Verb
+	// States are the names of the state the seat's holder keeps, which a module reads by `seat:<seat>.<state>` (design 47 G
+	// A-116, A-128). Compiled from the definition's states.
+	States []string
 	// Views are the names of the states the seat offers as views, which a module follows by `consumes`
 	// `seat:<seat>.views.<state>` (design 47 G §616, §699). Compiled from the definition's states marked `view`.
 	Views []string
+	// AcceptedOn are the accepted kinds that ride a stream, each as `<stream>.<kind>`, which a part names in `invokes` as
+	// `seat:<seat>.<stream>.<kind>` (design 47 G A-102, §415). Compiled from the definition's accepts.
+	AcceptedOn []string
 	// Receives is what other modules may contribute to the seat's holder, by kind (novox/hq ADR
 	// 0212): each kind is text in the tool's own grammar, placed by the holder with
 	// ${contribution:<seat>:<kind>}. Compiled, never stored: like the protocol, it is the mesh's

go/catalogue/state.go

@@ -78,6 +78,25 @@ func (s StateDeclaration) MarshalJSON() ([]byte, error) {
 	return json.Marshal(plain(s))
 }
 
+// SeatReadPrefix opens a read of the state a seat's holder keeps: `seat:<seat>.<state>` (design 47 G A-116, A-128).
+const SeatReadPrefix = "seat:"
+
+// SeatRead splits a read of a seat holder's state, `seat:<seat>.<state>`, into the seat and the local name; ok is false for a
+// read that is not of this kind (a module's `<module>.<state>`, which ReadState splits), and err says why a `seat:` read is
+// malformed. The holder is not in it: it is resolved where the read is made, so a new holder is read with no change of the
+// manifest.
+func SeatRead(read string) (seat, local string, ok bool, err error) {
+	rest, isSeat := strings.CutPrefix(read, SeatReadPrefix)
+	if !isSeat {
+		return "", "", false, nil
+	}
+	at := strings.LastIndex(rest, ".")
+	if at <= 0 || at == len(rest)-1 || !stateName.MatchString(rest[:at]) || !stateName.MatchString(rest[at+1:]) {
+		return "", "", true, fmt.Errorf("%q does not name a seat and its state: a read of a seat holder's state is seat:<seat>.<name>", read)
+	}
+	return rest[:at], rest[at+1:], true, nil
+}
+
 // SeatViewPrefix opens a consumed view of a seat, `seat:<seat>.views.<state>` (design 47 G §616).
 const SeatViewPrefix = "seat:"
 

go/catalogue/state_test.go

@@ -51,3 +51,26 @@ func TestAStateMayHaveKeysThatExpireOneByOne(t *testing.T) {
 		t.Fatalf("read-only given twice: %v", err)
 	}
 }
+
+// A read of a seat holder's state is its own kind, `seat:<seat>.<state>`, kept apart from a module's read (design 47 G A-128).
+func TestSeatRead(t *testing.T) {
+	for _, row := range []struct {
+		read, seat, local string
+		ok, bad           bool
+	}{
+		{"seat:mesh-provisioner.plans", "mesh-provisioner", "plans", true, false},
+		{"mesh-provisioner.plans", "", "", false, false},
+		{"seat:mesh-provisioner", "", "", true, true},
+		{"seat:.plans", "", "", true, true},
+		{"seat:Mesh.plans", "", "", true, true},
+	} {
+		seat, local, ok, err := SeatRead(row.read)
+		if seat != row.seat || local != row.local || ok != row.ok || (err != nil) != row.bad {
+			t.Errorf("SeatRead(%q) = %q, %q, %v, %v", row.read, seat, local, ok, err)
+		}
+	}
+	// A module's read reads as today.
+	if module, local, err := ReadState("mesh-provisioner.plans"); module != "mesh-provisioner" || local != "plans" || err != nil {
+		t.Errorf("a module's read = %q, %q, %v", module, local, err)
+	}
+}

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

@@ -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")
+	}
+}