Read the bus through a module's declared states in frombus, with the runtime's adapter in fromruntime, so mesh-provisioner and disk-load compose from one reader instead of two copies (issue 1955)

2026-10-29 23:31:34 +01:00
parent bb993dd606
commit 08fc70783a
3 changed files with 329 additions and 0 deletions
@@ -14,6 +14,7 @@ import (
 	"os"
 	"reflect"
 	"slices"
+	"sort"
 	"strings"
 	"testing"

@@ -724,6 +725,17 @@ func TestInput(t *testing.T) {
 			}
 		}
 	}
+	// And the same through StatesReader, over the fake's states as a module's runtime gives them (one reader for
+	// mesh-provisioner and disk-load, the lead 2026-10-29).
+	viaStates, _, err := DataItems(ctx, StatesReader(statesOf(t, ctx, noSecrets), func(_ context.Context, url string) ([]byte, error) {
+		for _, body := range noSecrets.archives {
+			return body, nil
+		}
+		return nil, errors.New("no archive")
+	}), Mesh{}, Scope{Node: "n1"})
+	if err != nil || !reflect.DeepEqual(viaStates, items) {
+		t.Errorf("through the states the data items are %v, %v; through the reader %v", viaStates, err, items)
+	}

 	// And what a provider on another node serves from its settings (mailu's smtp domain, issue 1126: shanks refused
 	// '${setting:domain} ... nothing sets "domain"'): read for that provider's served keys alone, through Served.
@@ -769,3 +781,72 @@ func TestInput(t *testing.T) {
 		}
 	}
 }
+
+// mapStates are read states by name, key to value, as a module's runtime gives them.
+type mapStates map[string]map[string][]byte
+
+func (m mapStates) Get(state, key string) ([]byte, bool, error) {
+	v, ok := m[state][key]
+	return v, ok, nil
+}
+
+func (m mapStates) Keys(state string) ([]string, error) {
+	var out []string
+	for k := range m[state] {
+		out = append(out, k)
+	}
+	sort.Strings(out)
+	return out, nil
+}
+
+// statesOf is b laid out as the states StatesReader reads.
+func statesOf(t *testing.T, ctx context.Context, b *bus) mapStates {
+	t.Helper()
+	out := mapStates{}
+	put := func(state, key string, raw []byte, err error) {
+		if err != nil {
+			t.Fatal(err)
+		}
+		if out[state] == nil {
+			out[state] = map[string][]byte{}
+		}
+		out[state][key] = raw
+	}
+	placements, _ := b.Placements(ctx)
+	for node, p := range placements {
+		k, _ := delivery.PlacementKey(node)
+		raw, err := delivery.EncodePlacement(p)
+		put(PlacementRead, k, raw, err)
+	}
+	records, _ := b.Records(ctx)
+	for node, r := range records {
+		k, _ := delivery.HaveNodeKey(node)
+		rec := r
+		raw, err := delivery.EncodeHaveNode(delivery.HaveNode{Record: &rec})
+		put(HaveRead, k, raw, err)
+	}
+	modules, _ := b.Modules(ctx)
+	for name, m := range modules {
+		raw, err := delivery.EncodeModule(m)
+		put(ModulesRead, name, raw, err)
+		var mf struct{ Module string }
+		_ = json.Unmarshal(b.manifests[m.Manifest], &mf)
+		k, _ := delivery.ManifestsKey(m.Manifest)
+		raw, err = delivery.EncodeManifest(delivery.Manifest{Module: mf.Module, JSON: string(b.manifests[m.Manifest])})
+		put(ManifestsRead, k, raw, err)
+	}
+	for key, build := range b.builds {
+		module, fp, _ := strings.Cut(key, " ")
+		k, _ := delivery.BuilderBuildKey(module, fp)
+		raw, err := delivery.EncodeBuilderBuild(*build)
+		put(BuildsRead, k, raw, err)
+	}
+	for nm, values := range b.values {
+		for k, v := range values {
+			value, _ := json.Marshal(v)
+			raw, err := json.Marshal(map[string]any{"value": json.RawMessage(value), "source": catalogue.SourceModule})
+			put(ResolvedRead, nm+"."+k, raw, err)
+		}
+	}
+	return out
+}
@@ -0,0 +1,199 @@
+package frombus
+
+// **The bus read through a module's declared `reads`** (the lead, 2026-10-29: one reader in place of the copies in
+// mesh-provisioner and disk-load): a Reader over the states a module reads, by a runtime's Get and Keys (a bundle's
+// stdio.State, a process's own), for a holder that composes plans or data items and reads no value: placement, each
+// node's record in its own have key, the indexer's modules and manifests, the builder's builds and mesh-config's resolved
+// values. No own secret and no credential is read (compose.Planning). Pure as the rest of this package: a bundle's
+// runtime states and an HTTP fetch are compose/fromruntime's.
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/compose"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// The states a module names in its manifest's `reads` to compose through StatesReader.
+const (
+	PlacementRead = delivery.PlacementRead
+	ModulesRead   = delivery.IndexerModule + ".modules"
+	ManifestsRead = delivery.IndexerModule + ".manifests"
+	BuildsRead    = delivery.BuilderModule + ".builds"
+	HaveRead      = "node-engine.have"
+	ResolvedRead  = "mesh-config.resolved"
+)
+
+// StatesReads are those reads, in a manifest's order.
+var StatesReads = []string{PlacementRead, ModulesRead, ManifestsRead, BuildsRead, HaveRead, ResolvedRead}
+
+// States is a module's read states as its runtime gives them: a key's value, false when it has none.
+type States interface {
+	Get(state, key string) (value []byte, found bool, err error)
+	Keys(state string) ([]string, error)
+}
+
+// Fetch reads an archive at a URL (fromruntime.HTTPFetch over HTTP).
+type Fetch func(ctx context.Context, url string) ([]byte, error)
+
+// StatesReader is the Reader over s; fetch reads a templates archive, nil for a holder that reads none.
+func StatesReader(s States, fetch Fetch) Reader { return statesReader{s: s, fetch: fetch} }
+
+type statesReader struct {
+	s     States
+	fetch Fetch
+}
+
+func (r statesReader) each(read string, visit func(key string, value []byte)) error {
+	keys, err := r.s.Keys(read)
+	if err != nil {
+		return fmt.Errorf("%s: %w", read, err)
+	}
+	for _, k := range keys {
+		v, found, err := r.s.Get(read, k)
+		if err != nil {
+			return fmt.Errorf("%s %s: %w", read, k, err)
+		}
+		if found {
+			visit(k, v)
+		}
+	}
+	return nil
+}
+
+func (r statesReader) Placements(context.Context) (map[string]delivery.Placement, error) {
+	out := map[string]delivery.Placement{}
+	err := r.each(PlacementRead, func(k string, v []byte) {
+		if node, err := delivery.ParsePlacementKey(k); err == nil {
+			if p, err := delivery.DecodePlacement(v); err == nil && p.Node == node {
+				out[node] = p
+			}
+		}
+	})
+	return out, err
+}
+
+func (r statesReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
+	out := map[string]delivery.NodeRecord{}
+	err := r.each(HaveRead, func(k string, v []byte) {
+		if node, _, isNode, err := delivery.ParseHaveKey(k); err == nil && isNode {
+			if h, err := delivery.DecodeHaveNode(v); err == nil && h.Record != nil {
+				out[node] = *h.Record
+			}
+		}
+	})
+	return out, err
+}
+
+func (r statesReader) Modules(context.Context) (map[string]delivery.Module, error) {
+	out := map[string]delivery.Module{}
+	err := r.each(ModulesRead, func(k string, v []byte) {
+		if module, err := delivery.ParseModulesKey(k); err == nil {
+			if m, err := delivery.DecodeModule(v); err == nil {
+				out[module] = m
+			}
+		}
+	})
+	return out, err
+}
+
+func (r statesReader) Manifest(_ context.Context, digest string) ([]byte, error) {
+	key, err := delivery.ManifestsKey(digest)
+	if err != nil {
+		return nil, err
+	}
+	v, found, err := r.s.Get(ManifestsRead, key)
+	if err != nil {
+		return nil, err
+	}
+	if !found {
+		return nil, fmt.Errorf("the indexer holds no manifest %s", digest)
+	}
+	m, err := delivery.DecodeManifest(v)
+	if err != nil {
+		return nil, err
+	}
+	return []byte(m.JSON), nil
+}
+
+func (r statesReader) Values(_ context.Context, node, module string) (map[string]any, error) {
+	var out map[string]any
+	prefix := node + "." + module + "."
+	err := r.each(ResolvedRead, func(k string, v []byte) {
+		name, ok := strings.CutPrefix(k, prefix)
+		if !ok || strings.Contains(name, ".") {
+			return
+		}
+		if value, laid := catalogue.LaidValue(v); laid {
+			if out == nil {
+				out = map[string]any{}
+			}
+			out[name] = value
+		}
+	})
+	return out, err
+}
+
+func (statesReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
+	return nil, nil
+}
+
+func (statesReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
+	return nil, nil, nil
+}
+
+func (r statesReader) Build(_ context.Context, module, fp string) (*delivery.BuilderBuild, error) {
+	key, err := delivery.BuilderBuildKey(module, fp)
+	if err != nil {
+		return nil, err
+	}
+	v, found, err := r.s.Get(BuildsRead, key)
+	if err != nil || !found {
+		return nil, err
+	}
+	b, err := delivery.DecodeBuilderBuild(v)
+	if err != nil {
+		return nil, err
+	}
+	return &b, nil
+}
+
+// archiveWithin bounds one read of a templates archive.
+const archiveWithin = 30 * time.Second
+
+func (r statesReader) Archive(ctx context.Context, url, digest string) ([]byte, error) {
+	if r.fetch == nil {
+		return nil, fmt.Errorf("this holder reads no archive (%s)", url)
+	}
+	fetching, cancel := context.WithTimeout(ctx, archiveWithin)
+	defer cancel()
+	body, err := r.fetch(fetching, url)
+	if err != nil {
+		return nil, err
+	}
+	sum := sha256.Sum256(body)
+	if got := hex.EncodeToString(sum[:]); got != strings.TrimPrefix(digest, "sha256:") {
+		return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got)
+	}
+	return body, nil
+}
+
+func (r statesReader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) {
+	all, err := r.Values(ctx, node, module)
+	var out map[string]any
+	for _, k := range keys {
+		if v, ok := all[k]; ok {
+			if out == nil {
+				out = map[string]any{}
+			}
+			out[k] = v
+		}
+	}
+	return out, err
+}
@@ -0,0 +1,49 @@
+// Package fromruntime is compose/frombus's StatesReader over a bundle's runtime (the lead, 2026-10-29): the states a
+// module's manifest names in `reads` (frombus.StatesReads), read through stdio.State, and a templates archive fetched
+// over HTTP. Apart from frombus so that frombus stays pure for the mesh-compose command.
+package fromruntime
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
+)
+
+// States are a bundle's read states through its runtime.
+type States struct{}
+
+// Get is a key's value, false when it has none.
+func (States) Get(state, key string) ([]byte, bool, error) {
+	e, err := stdio.State(state).Get(key)
+	if err != nil || e == nil {
+		return nil, false, err
+	}
+	return e.Value, true, nil
+}
+
+// Keys are a state's keys.
+func (States) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
+
+// Reader is frombus's Reader over this bundle's read states, fetching archives over HTTP.
+func Reader() frombus.Reader { return frombus.StatesReader(States{}, HTTPFetch) }
+
+// HTTPFetch reads an archive over HTTP, at most 256 MiB.
+func HTTPFetch(ctx context.Context, url string) ([]byte, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("%s answered %s", url, resp.Status)
+	}
+	return io.ReadAll(io.LimitReader(resp.Body, 256<<20))
+}