DLV-I11: indexer_bus_state, the bus module's keeper's first view of every trunk module's accepted bus state

Answers {revision, modules:[{module, digest, revision, bus-state}]} from the indexer's own states, at the verdicts state's mark, as A5's keeper reads it. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 12:44:46 +01:00
parent 7fd07d0236
commit b5a88fae9d
5 changed files with 175 additions and 2 deletions
@@ -194,7 +194,7 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) {
 }

 func main() {
-	if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), append(append(catalogTools(), viewTools(os.Getenv, keptSource, time.Now)...), configReadersTool(stdio.NewWatchedStates()))...)); err != nil {
+	if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), append(append(catalogTools(), viewTools(os.Getenv, keptSource, time.Now)...), configReadersTool(stdio.NewWatchedStates()), busStateTool(keptSource))...)); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err)
 		os.Exit(1)
 	}
@@ -81,3 +81,16 @@ func configReadersTool(watched *stdio.WatchedStates) stdio.Tool {
 			return map[string]any{"readers": index.ConfigReaders(module, trunk, kept, pulls)}, nil
 		}}
 }
+
+// busStateTool is indexer_bus_state (design 25 § 12.2; DLV-I11): the bus module's keeper's first view of every trunk
+// module's accepted bus state, read from the indexer's own states only, so it answers as soon as the runtime is up.
+func busStateTool(state func(string) views.Source) stdio.Tool {
+	return stdio.Tool{Name: index.BusStateTool,
+		Description: "Every trunk module whose accepted manifest asks for bus state (design 25 § 12.2): its module, the digest of " +
+			"that manifest, the verdict's revision and its bus-state entries whole, and the revision of the verdicts state the " +
+			"answer is at, so every manifest-accepted above it is newer. The bus module's keeper takes its first view from it.",
+		Input: map[string]any{"type": "object", "properties": map[string]any{}},
+		Run: func(map[string]any) (any, error) {
+			return index.BusStateOf(state("modules"), state(index.VerdictsState), state("manifests"), state(views.MarksState))
+		}}
+}
@@ -0,0 +1,101 @@
+package index
+
+import (
+	"fmt"
+	"slices"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/views"
+)
+
+// BusStateTool is the indexer's tool the bus module's keeper takes its first view from (design 25 § 12.2; DLV-I11):
+// every trunk module whose accepted manifest asks for bus state, then each manifest-accepted after its revision.
+const BusStateTool = "indexer_bus_state"
+
+// BusStateModule is one module of BusStateAnswer: the digest of its trunk's accepted manifest, the verdicts state's
+// revision of that verdict (the one its manifest-accepted carries, ADR 0622 §2), and its bus-state entries whole.
+type BusStateModule struct {
+	Module   string                    `json:"module"`
+	Digest   string                    `json:"digest"`
+	Revision uint64                    `json:"revision"`
+	BusState []catalogue.BusStateEntry `json:"bus-state"`
+}
+
+// BusStateAnswer is indexer_bus_state's answer: Revision is the verdicts state's mark, read before anything else, so
+// every manifest-accepted above it is a change after this answer (ADR 0615 §4); Modules in module order.
+type BusStateAnswer struct {
+	Revision uint64           `json:"revision"`
+	Modules  []BusStateModule `json:"modules"`
+}
+
+// BusStateOf answers indexer_bus_state from the indexer's own states: its trunk modules, the judge's verdicts and the
+// manifests it keeps, and the marks the views family says them up to. A module refused or waiting, or whose manifest
+// asks for no bus state, is left out; a value that cannot be read is an error, never a module silently left out.
+func BusStateOf(modules, verdicts, manifests, marks views.Source) (BusStateAnswer, error) {
+	out := BusStateAnswer{Modules: []BusStateModule{}}
+	if marks != nil {
+		if e, ok, err := marks.Get(VerdictsState); err != nil {
+			return BusStateAnswer{}, fmt.Errorf("the mark of %s cannot be read: %w", VerdictsState, err)
+		} else if ok {
+			m, err := views.DecodeMark(e.Value)
+			if err != nil {
+				return BusStateAnswer{}, fmt.Errorf("the mark of %s: %w", VerdictsState, err)
+			}
+			out.Revision = m.Revision
+		}
+	}
+	names, err := modules.Keys()
+	if err != nil {
+		return BusStateAnswer{}, fmt.Errorf("the modules state cannot be listed: %w", err)
+	}
+	slices.Sort(names)
+	for _, name := range names {
+		e, ok, err := modules.Get(name)
+		if err != nil {
+			return BusStateAnswer{}, fmt.Errorf("modules.%s cannot be read: %w", name, err)
+		}
+		if !ok {
+			continue
+		}
+		mod, err := delivery.DecodeModule(e.Value)
+		if err != nil || mod.Manifest == "" {
+			continue // no manifest held for it, so no verdict
+		}
+		key, err := delivery.VerdictsKey(mod.Manifest)
+		if err != nil {
+			continue
+		}
+		ve, ok, err := verdicts.Get(key)
+		if err != nil {
+			return BusStateAnswer{}, fmt.Errorf("the verdict on %s cannot be read: %w", name, err)
+		}
+		if !ok {
+			continue
+		}
+		v, err := delivery.DecodeManifestVerdict(ve.Value)
+		if err != nil || v.State != delivery.VerdictAccepted {
+			continue
+		}
+		me, ok, err := manifests.Get(mod.Manifest)
+		if err != nil {
+			return BusStateAnswer{}, fmt.Errorf("the manifest of %s cannot be read: %w", name, err)
+		}
+		if !ok {
+			continue
+		}
+		kept, err := delivery.DecodeManifest(me.Value)
+		if err != nil {
+			return BusStateAnswer{}, fmt.Errorf("the manifest of %s: %w", name, err)
+		}
+		m, err := catalogue.Decode([]byte(kept.JSON))
+		if err != nil {
+			return BusStateAnswer{}, fmt.Errorf("the accepted manifest of %s does not read: %w", name, err)
+		}
+		if len(m.BusState) == 0 {
+			continue
+		}
+		out.Modules = append(out.Modules, BusStateModule{Module: name, Digest: mod.Manifest, Revision: ve.Revision, BusState: m.BusState})
+	}
+	return out, nil
+}
@@ -10,6 +10,7 @@ import (

 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/process"
+	"git.novox.be/novox/mesh-sdk/go/views"
 )

 // 1953/D2, D3 (novox/hq ADR 0565 §2): the indexer keeps the catalogue's predefined list from the merged message, a merge older
@@ -37,6 +38,50 @@ func TestTheControlRecords(t *testing.T) {
 			t.Errorf("rows = %+v\nwant %+v", got, want)
 		}
 	})
+	// DLV-I11 (design 25 § 12.2): indexer_bus_state answers every trunk module whose accepted manifest asks for bus state,
+	// with the verdict's revision, at the verdicts state's mark; a refused one, one asking none, and a module with no
+	// manifest held are left out.
+	t.Run("indexer_bus_state", func(t *testing.T) {
+		src := memSource{}
+		dAsk, dNone, dRefused := strings.Repeat("a", 64), strings.Repeat("b", 64), strings.Repeat("c", 64)
+		put := func(s memSource, key string, rev uint64, v any) {
+			raw, err := json.Marshal(v)
+			if err != nil {
+				t.Fatal(err)
+			}
+			s[key] = &views.Entry{Value: raw, Revision: rev}
+		}
+		modules, verdicts, manifests, marks := memSource{}, memSource{}, memSource{}, src
+		putVerdict := func(digest string, rev uint64, v delivery.ManifestVerdict) {
+			raw, err := delivery.EncodeManifestVerdict(v)
+			if err != nil {
+				t.Fatal(err)
+			}
+			verdicts[digest] = &views.Entry{Value: raw, Revision: rev}
+		}
+		for name, digest := range map[string]string{"cache": dAsk, "plain": dNone, "bad": dRefused, "bare": ""} {
+			raw, err := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: digest,
+				Repository: "novox/mesh-catalog", Commit: "c"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			modules[name] = &views.Entry{Value: raw, Revision: 1}
+		}
+		asks := `{"module":"cache","version":"1","requires":[{"name":"c","storage":"key-value","template":"state"}]}`
+		for digest, raw := range map[string]string{dAsk: asks, dNone: `{"module":"plain","version":"1"}`, dRefused: asks} {
+			put(manifests, digest, 1, delivery.Manifest{V: delivery.Version, Module: "m", JSON: raw})
+		}
+		putVerdict(dAsk, 7, delivery.ManifestVerdict{Module: "cache", State: delivery.VerdictAccepted, Judge: "j"})
+		putVerdict(dNone, 8, delivery.ManifestVerdict{Module: "plain", State: delivery.VerdictAccepted, Judge: "j"})
+		putVerdict(dRefused, 9, delivery.ManifestVerdict{Module: "bad", State: delivery.VerdictRefused, Judge: "j", Problems: []string{"p"}})
+		put(marks, VerdictsState, 1, views.Mark{V: views.Version, State: VerdictsState, Revision: 9})
+		got, err := BusStateOf(modules, verdicts, manifests, marks)
+		if err != nil || got.Revision != 9 || len(got.Modules) != 1 || got.Modules[0].Module != "cache" ||
+			got.Modules[0].Digest != dAsk || got.Modules[0].Revision != 7 || len(got.Modules[0].BusState) != 1 ||
+			got.Modules[0].BusState[0].Name != "c" {
+			t.Errorf("indexer_bus_state = %+v (%v)", got, err)
+		}
+	})
 	ctx := context.Background()
 	at := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC)
 	t.Run("the predefined list, never replaced by an older merge", func(t *testing.T) {
@@ -108,3 +153,16 @@ func TestTheControlRecords(t *testing.T) {
 		}
 	})
 }
+
+// memSource is a state as views.Source, for the indexer's tools that answer from its own states.
+type memSource map[string]*views.Entry
+
+func (m memSource) Keys() ([]string, error) {
+	var keys []string
+	for k := range m {
+		keys = append(keys, k)
+	}
+	return keys, nil
+}
+
+func (m memSource) Get(key string) (*views.Entry, bool, error) { e, ok := m[key]; return e, ok, nil }
@@ -35,7 +35,8 @@
    "indexer_read_findings",
    "indexer_first_views",
    "indexer_first_view_again",
-    "indexer_config_readers"
+    "indexer_config_readers",
+    "indexer_bus_state"
  ],
  "consumes": [
    "mesh-build-go.toolchain.reported",