Indexer: its views family is its ordinary events (views.<state> among its emits, no stream of its own), the assigner's placement followed on the shared events stream, its findings through the views' writer; the seat mesh-catalog's verbs from the contracts, described by the tool runner; a seat list in its tests (D-25 owner half, D-29 tools; issue 1966)

The indexer's consumers of its durable on the shared stream still compete; they join the module's one hub.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-05 15:08:25 +01:00
parent 2e8b4dcdcd
commit 06d4e19631
12 changed files with 107 additions and 104 deletions
@@ -5,8 +5,8 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
@@ -17,7 +17,7 @@ import (
// holdsTool reads the indexer's mesh-wide holds (novox/hq ADR 0632 §2), asks one again now, or moves one by a person's
// word: the asks go through the takeins state, as indexer_take_in's do, and the indexer answers there.
func holdsTool(read stdio.KeptState, ask asker, now func() time.Time, acts handacts.Recorder) stdio.Tool {
func holdsTool(read stdio.KeptState, ask asker, now func() time.Time, acts meshhandactsclient.Recorder) stdio.Tool {
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
return stdio.Tool{Name: index.HoldsTool,
Description: "The modules whose new manifest the indexer holds from the whole mesh (ADR 0632 §2): a manifest every node-engine " +
@@ -52,7 +52,7 @@ func holdsTool(read stdio.KeptState, ask asker, now func() time.Time, acts handa
}
}
}
if _, err := acts.Record(handacts.Request{What: what,
if _, err := acts.Record(meshhandactsclient.Request{What: what,
Why: str(args, "why"), Cause: index.HoldsTool, Verb: "hold-release"}); err != nil {
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so the hold was not moved: %w", err)
}
@@ -15,8 +15,8 @@ import (
"sync"
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/handacts"
"mesh-delivery-indexer/internal/index"
)
@@ -25,7 +25,7 @@ import (
type asker func(key string, value []byte) (index.TakeInAsk, error)
// Tools are the indexer's tools. read is the indexer's state of verdicts, which indexer_verdicts reads; counts the old sections' count.
func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func() (index.OldSectionsCount, error), acts handacts.Recorder) []stdio.Tool {
func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func() (index.OldSectionsCount, error), acts meshhandactsclient.Recorder) []stdio.Tool {
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
return []stdio.Tool{
{Name: index.OldSectionsTool,
@@ -118,7 +118,7 @@ func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func()
if err != nil {
return nil, err
}
if _, err := acts.Record(handacts.Request{What: "said manifest-accepted of " + str(args, "module") + " again by hand",
if _, err := acts.Record(meshhandactsclient.Request{What: "said manifest-accepted of " + str(args, "module") + " again by hand",
Why: str(args, "why"), Cause: index.SayAcceptedTool, Verb: "say-accepted"}); err != nil {
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was said again: %w", err)
}
@@ -149,7 +149,7 @@ func Tools(ask asker, now func() time.Time, read stdio.KeptState, counts func()
if str(args, "why") == "" {
return nil, fmt.Errorf("say why it is read again (why): it is a hand act")
}
if _, err := acts.Record(handacts.Request{What: "read the seat definitions of " + str(args, "module") + " again by hand",
if _, err := acts.Record(meshhandactsclient.Request{What: "read the seat definitions of " + str(args, "module") + " again by hand",
Why: str(args, "why"), Cause: index.SeatDefinitionsAgainTool, Verb: "seat-definitions-again"}); err != nil {
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was read again: %w", err)
}
@@ -273,7 +273,7 @@ var errReadOnly = errors.New("the tools bundle only reads this state")
func main() {
asks := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second)
acts := handacts.Recorder{Ask: stdio.Ask}
acts := meshhandactsclient.Recorder{Ask: stdio.Ask}
tools := Tools(asks, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState)), acts)
tools = append(tools, catalogTools()...)
tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...)
@@ -5,9 +5,9 @@ import (
"fmt"
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/views"
@@ -29,7 +29,7 @@ func viewTools(getenv func(string) string, state func(string) views.Source, now
out = append(out, views.ReadTool(delivery.IndexerModule, t.Tool, t.State, t.Words, state(t.State), marks, now))
}
// Its own first views, of the assigner's placement (DLV-F3r): read, and asked again by hand.
return append(out, views.StatusTools(delivery.IndexerModule, "indexer_first_views", "indexer_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, now)...)
return append(out, views.StatusTools(delivery.IndexerModule, "indexer_first_views", "indexer_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, now)...)
}
// keptSource is one of the indexer's states as its runtime reads it.
@@ -82,11 +82,6 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
}
}
// The take-in events are the indexer's ordinary events (ADR 0633 §5), on the path manifest-accepted takes.
for name, subject := range map[string]string{"INDEXER_EVENTS": "mesh.mod.mesh-delivery-indexer.event.>"} {
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: []string{subject}}); err != nil {
t.Fatal(err)
}
}
tc, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1})
if err != nil {
t.Fatal(err)
@@ -217,7 +212,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
return ack.Sequence
}
m1Key := takein.PendingKey(merged(m1))
events, err := js.Stream(ctx, "INDEXER_EVENTS")
events, err := js.Stream(ctx, "EVENTS")
if err != nil {
t.Fatal(err)
}
@@ -432,7 +427,7 @@ func viewsEvents(t *testing.T, ctx context.Context, js jetstream.JetStream, stat
var out []views.Event
for seq := info.State.FirstSeq; seq > 0 && seq <= info.State.LastSeq; seq++ {
m, err := st.GetMsg(ctx, seq)
if err != nil || m.Subject != "mesh.stream.mesh-delivery-indexer.event."+views.EventName(state) {
if err != nil || m.Subject != "mesh.mod.mesh-delivery-indexer.event."+views.EventName(state) {
continue
}
e, err := views.DecodeEvent(m.Data)
@@ -458,7 +453,7 @@ func waitViews(t *testing.T, ctx context.Context, js jetstream.JetStream, state,
return views.Event{}
}
const views0Stream = "INDEXER_VIEWS"
const views0Stream = "EVENTS"
// acceptedHeard counts what a waiter on manifest-accepted heard.
var acceptedHeard atomic.Int64
@@ -533,11 +528,8 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
Subjects: []string{buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog) + ".>"}, AllowDirect: true}); err != nil {
t.Fatal(err)
}
// The indexer's views family and its marks (ADR 0615 §5), as the grants part makes them from the old sections, and a
// waiter on manifest-accepted, which the views family never wakes.
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: views0Stream, Subjects: []string{"mesh.stream.mesh-delivery-indexer.event.views.>"}}); err != nil {
t.Fatal(err)
}
// The indexer's marks (ADR 0615 §5), and a waiter on manifest-accepted, which the views family never wakes (its events
// are ordinary events on the shared stream, ADR 0635 §5).
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.MarksBucket, History: 1}); err != nil {
t.Fatal(err)
}
@@ -547,6 +539,10 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
t.Fatal(err)
}
t.Cleanup(func() { _ = waiter.Unsubscribe() })
// The shared events stream: the indexer's ordinary events and the assigner's views events (ADR 0635 §5).
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
t.Fatal(err)
}
put(ctx, buildLog{ctx: ctx, js: js})
for _, b := range before {
b(ctx, js)
@@ -659,9 +655,6 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.VerdictsBucket, History: delivery.VerdictsHistory}); err != nil {
t.Fatal(err)
}
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "INDEXER_EVENTS", Subjects: []string{"mesh.mod.mesh-delivery-indexer.event.>"}}); err != nil {
t.Fatal(err)
}
})
forge := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301,
Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0",
@@ -729,7 +722,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
t.Errorf("the refusal names no problem or the wait no field: %+v", got)
}
// manifest-refused, said once for each refusal (a wait is no refusal).
events, err := js.Stream(ctx, "INDEXER_EVENTS")
events, err := js.Stream(ctx, "EVENTS")
if err != nil {
t.Fatal(err)
}
@@ -1112,13 +1105,11 @@ func serveAssignerPlacement(t *testing.T, ctx context.Context, js jetstream.JetS
if err != nil {
t.Fatal(err)
}
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.views.>"}}); err != nil {
if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy,
FilterSubjects: []string{"mesh.mod." + delivery.AssignerModule + ".event.views.>"}}); err != nil {
t.Fatal(err)
}
if _, err := js.CreateOrUpdateConsumer(ctx, "ASSIGNER_VIEWS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
t.Fatal(err)
}
em := views.Emitter{Owner: delivery.AssignerModule, Subject: func(e string) string { return process.ModuleStreamSubject(delivery.AssignerModule, e) },
em := views.Emitter{Owner: delivery.AssignerModule, Subject: process.Account{Node: "n1", Module: delivery.AssignerModule}.EventSubject,
Publish: views.JetStreamPublish(js), Marks: marks}
if err := em.Start(ctx, delivery.PlacementState, placement, nil); err != nil {
t.Fatal(err)
@@ -0,0 +1,15 @@
package main
import (
"os"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest"
)
// TestMain gives catalogue the seat list a host reads (ADR 0636 §1): none is compiled into mesh-sdk.
func TestMain(m *testing.M) {
catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions()))
os.Exit(m.Run())
}
@@ -10,11 +10,11 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings"
"git.novox.be/novox/mesh-delivery/sdk/graph"
"git.novox.be/novox/mesh-delivery/sdk/heard"
"git.novox.be/novox/mesh-sdk/go/views"
@@ -39,11 +39,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, cred process.Accoun
if kv == nil {
return
}
k, v, err := views.TooLargeFinding(delivery.IndexerModule, state, key, size, time.Now())
if err == nil {
_, err = kv.Put(ctx, k, v)
}
if err != nil {
if err := sdkfindings.Writer(ctx, findings.KV).Put(views.TooLargeFinding(delivery.IndexerModule, state, key, size, time.Now())); err != nil {
logf("views: the finding of %s.%s too large is not written: %v", state, key, err)
}
}
@@ -63,20 +59,10 @@ func startViews(ctx context.Context, js jetstream.JetStream, cred process.Accoun
// view-unsaid, in its own findings once bound (review round 2).
func viewsEmitter(ctx context.Context, js jetstream.JetStream, cred process.Account, mk jetstream.KeyValue, findings *process.AwaitedBucket,
tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter {
w := notHeardWriter(ctx, findings, logf)
return views.Emitter{Owner: delivery.IndexerModule, Subject: cred.StreamSubject, Publish: views.JetStreamPublish(js),
Marks: mk, TooLarge: tooLarge, Findings: funcFindings{put: w.Put, del: w.Delete}, Logf: logf}
return views.Emitter{Owner: delivery.IndexerModule, Subject: cred.EventSubject, Publish: views.JetStreamPublish(js),
Marks: mk, TooLarge: tooLarge, Findings: sdkfindings.Writer(ctx, findings.KV), Logf: logf}
}
// funcFindings is a findings state as two functions: views.Emitter's Findings.
type funcFindings struct {
put func(string, []byte) error
del func(string) error
}
func (f funcFindings) Put(key string, value []byte) error { return f.put(key, value) }
func (f funcFindings) Delete(key string) error { return f.del(key) }
func sortedStates(states map[string]readyOf) []string {
out := make([]string, 0, len(states))
for s := range states {
@@ -126,9 +112,6 @@ var firstViewsBucket = delivery.IndexerModule + "_" + views.StatusState
// placementTool is the assigner's first-view tool of its placement, as the indexer's invokes names it (ADR 0621 §2).
const placementTool = delivery.AssignerModule + "." + delivery.PlacementState
// assignerViewsStream is the assigner's views family, as the assigner's manifest's `streams` names it (ADR 0615 §5).
const assignerViewsStream = "assigner-views"
// followPlacement reads the seat mesh-placement's placement through the assigner's first view and then its views
// family, never its bucket (ADR 0615 §1, DLV-F1r), into gate: each key as it changes, settled once answered, with
// changed told whenever what the gate passes on changed. It waits for the first answer at most views.Bound, so a
@@ -137,15 +120,13 @@ const assignerViewsStream = "assigner-views"
func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.Account, call func(string, any) (json.RawMessage, error),
gate *buildrun.ToolchainGate, changed func(context.Context), findings, status *process.AwaitedBucket, logf func(string, ...any)) *views.Follower {
finding := func(s views.Status, raised bool) {
kv := findings.KV()
if kv == nil {
return
}
key, value, err := views.UnansweredFinding(delivery.IndexerModule, s, time.Now())
if err == nil && raised {
_, err = kv.Put(ctx, key, value)
} else if err == nil {
err = kv.Delete(ctx, key)
w := sdkfindings.Writer(ctx, findings.KV)
f := views.UnansweredFinding(delivery.IndexerModule, s, time.Now())
var err error
if raised {
err = w.Put(f)
} else {
err = w.Delete(f)
}
if err != nil {
logf("views: the finding of the first view %s is not written: %v", s.Tool, err)
@@ -185,7 +166,7 @@ func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.A
}()
// The assigner's views family, on the indexer's durable consumer, once the first view is asked; not heard, it is the
// finding family-not-heard until it binds (ADR 0615 §5).
stream := catalogue.StreamBusName(assignerViewsStream)
stream := process.EventsStream
_ = notHeardWriter(ctx, findings, logf).Bind(stream, "the placement is the first view's alone, asked again by hand, until the indexer starts again", func() error {
cons, err := cred.EventConsumer(ctx, js, stream)
if err != nil {
+2
View File
@@ -3,6 +3,8 @@ module mesh-delivery-indexer
go 1.26.0
require (
git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
@@ -12,6 +12,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/views"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/sdk/graph"
)
@@ -153,39 +154,27 @@ func (c CatalogRead) graphAndModules() (graph.Graph, map[string]delivery.Module,
return g, mods, nil
}
// CatalogTools are the seat mesh-catalog's four verbs as tools named <seat>.<verb>, with the descriptions and inputs the
// seat's contract gives them (novox/hq ADR 0604 §1, §2), answered from the indexer's own states. They are served once the
// indexer's manifest claims the seat, in the release after L3's (the lead's decision): until then main serves them not.
// CatalogTools are the seat mesh-catalog's verbs as tools named <seat>.<verb>, answered from the indexer's own states
// (novox/hq ADR 0604 §1, §2, ADR 0636 §4). The tool runner describes each with the seat's definition from the seat list
// (ADR 0635 §1), so the tools carry no description or input.
func CatalogTools(c CatalogRead) []stdio.Tool {
seat, _ := catalogue.SeatNamed(CatalogSeat)
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s }
runs := map[string]func(map[string]any) (any, error){
"modules": func(map[string]any) (any, error) { return c.CatalogModulesAnswer() },
"module": func(a map[string]any) (any, error) { return c.CatalogModuleAnswer(str(a, "module")) },
"provides": func(a map[string]any) (any, error) { return c.CatalogProvidesAnswer(str(a, "provision")) },
"dependents": func(a map[string]any) (any, error) { return c.CatalogDependentsAnswer(str(a, "module")) },
}
var out []stdio.Tool
for _, v := range seat.Serves {
if run, ok := runs[v.Name]; ok {
out = append(out, stdio.Tool{Name: CatalogSeat + "." + v.Name, Description: v.Description, Input: v.Input, Run: run})
}
meshcatalog.VerbModules: func(map[string]any) (any, error) { return c.CatalogModulesAnswer() },
meshcatalog.VerbModule: func(a map[string]any) (any, error) { return c.CatalogModuleAnswer(str(a, "module")) },
meshcatalog.VerbProvides: func(a map[string]any) (any, error) { return c.CatalogProvidesAnswer(str(a, "provision")) },
meshcatalog.VerbDependents: func(a map[string]any) (any, error) { return c.CatalogDependentsAnswer(str(a, "module")) },
}
if c.SeatDefinitions != nil {
out = append(out, stdio.Tool{Name: CatalogSeat + "." + SeatDefinitionsVerb, Description: SeatDefinitionsDescription, Input: SeatDefinitionsInput,
Run: func(a map[string]any) (any, error) {
return SeatDefinitionsOf(c.SeatDefinitions, c.Marks, str(a, "seat"))
}})
runs[meshcatalog.VerbSeatDefinitions] = func(a map[string]any) (any, error) {
return SeatDefinitionsOf(c.SeatDefinitions, c.Marks, str(a, "seat"))
}
}
var out []stdio.Tool
for _, v := range meshcatalog.Verbs() {
if run, ok := runs[v]; ok {
out = append(out, stdio.Tool{Name: meshcatalog.Seat + "." + v, Run: run})
}
}
return out
}
// SeatDefinitionsDescription and SeatDefinitionsInput are the verb seat-definitions of the seat mesh-catalog (ADR 0636 §4).
const SeatDefinitionsDescription = "The seat list: the definitions of the seats the trunks hold. With no seat, every trunk seat with the digest of " +
"its definition and the revision of the state it is read from; with a seat, that seat's trunk definition, resolved. Only the " +
"merged definitions, never a pull request's (indexer_seat_definition answers those). Every reader of the seat list reads it " +
"here, and follows seat-definition-changed and seat-definition-removed."
// SeatDefinitionsInput is the verb's input.
var SeatDefinitionsInput = map[string]any{"type": "object", "properties": map[string]any{
"seat": map[string]any{"type": "string", "description": "one seat's name; absent for the list of every trunk seat"}}}
@@ -0,0 +1,15 @@
package index
import (
"os"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest"
)
// TestMain gives catalogue the seat list a host reads (ADR 0636 §1): none is compiled into mesh-sdk.
func TestMain(m *testing.M) {
catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions()))
os.Exit(m.Run())
}
@@ -8,10 +8,6 @@ import (
"git.novox.be/novox/mesh-sdk/go/views"
)
// ViewsStream is the indexer's views family on its old streams section (ADR 0615 §5): one event per changed key of
// each state in ViewedStates. Its name is the manifest's `streams` entry.
const ViewsStream = "indexer-views"
// MarksBucket is the indexer's state of what its views family said each state up to.
const MarksBucket = delivery.IndexerModule + "_" + views.MarksState
+10 -5
View File
@@ -110,7 +110,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
// 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state. ADR 0578 §1: it says
// a refused manifest as manifest-refused, and (issue 1966 D3, 1966/AC10) a newly accepted one asking for bus state as
// manifest-accepted.
if !reflect.DeepEqual(m.Emits, []string{"manifest-refused", "manifest-accepted", takein.TakeInPendingEvent, takein.TakeInTakenEvent}) {
if own := slices.DeleteFunc(slices.Clone(m.Emits), func(e string) bool { return strings.HasPrefix(e, views.EventPrefix) }); !reflect.DeepEqual(own, []string{"manifest-refused", "manifest-accepted", takein.TakeInPendingEvent, takein.TakeInTakenEvent}) {
t.Errorf("emits = %v, want manifest-refused, manifest-accepted and the two take-in events", m.Emits)
}
// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
@@ -213,11 +213,16 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
}
}
}
// ADR 0615 §5, ADR 0619 §2: its views family on its old streams section, 7 days and 256 MiB, dropping its oldest.
// ADR 0635 §5: its views family is its ordinary events, `views.<state>` among its emits, and it has no stream of its own.
// ADR 0636 §4: seat-definition-changed and -removed are the seat mesh-catalog's events, said on the seat's subject and kept
// on the shared events stream: no stream of the indexer's, and not among its own emits.
if want := []stream{{Name: index.ViewsStream, Events: []string{views.EventPattern}, MaxAgeDays: views.StreamDays, MaxMiB: views.StreamMiB}}; !reflect.DeepEqual(m.Streams, want) {
t.Errorf("streams = %+v, want %+v", m.Streams, want)
// on the shared events stream, and not among its own emits.
if len(m.Streams) != 0 {
t.Errorf("streams = %+v, want none", m.Streams)
}
for _, state := range []string{"candidates", "config-candidates", "findings", "graph", "heard", "manifests", "modules", "packages", "predefined", "seat-definitions", "taken", "verdicts"} {
if !slices.Contains(m.Emits, views.EventPrefix+state) {
t.Errorf("emits = %v: want %s%s", m.Emits, views.EventPrefix, state)
}
}
// ADR 0604 §2, ADR 0636 §4: it holds the seat mesh-catalog, serving its four verbs and seat-definitions from its tools
// bundle (which stays without a request subject).
+13 -4
View File
@@ -88,14 +88,23 @@
{ "name": "hold-bounds", "history": 1, "key-ttl": true },
{ "name": "pending", "history": 1 }
],
"streams": [
{ "name": "indexer-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 }
],
"emits": [
"manifest-refused",
"manifest-accepted",
"take-in.pending",
"take-in.taken"
"take-in.taken",
"views.candidates",
"views.config-candidates",
"views.findings",
"views.graph",
"views.heard",
"views.manifests",
"views.modules",
"views.packages",
"views.predefined",
"views.seat-definitions",
"views.taken",
"views.verdicts"
],
"reads": [
"mesh-build-go.toolchains",