A node's plans took 2.6-13.6 s of CPU on the provisioner: one composition resolved every machine of the mesh some two hundred times (who is on the private network, and the rest of the mesh, worked out again on every look), three compositions did so again each, and the holder composed twice. A composer now works each out once over its unchanging Input; Composing gives Resolve, Render and Generators one composer; PlansHeldTwice composes the wanted and the newest builds over one read of the bus, says where the time went by phase, and the states reader keeps a build's take-in by its content. WatchedStates takes a process's own writes into the copy at once, so a holder reads its own states from it. Measured on a captured novox mesh: 11.7 s to 0.49 s per composition of an unchanged node, plans and provisions identical on all four nodes. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
205 lines
6.2 KiB
Go
205 lines
6.2 KiB
Go
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.
|
|
// It keeps what each build's take-in made, by the build's content (issue 1954), so a holder composing again and again
|
|
// takes in only a build it has not seen.
|
|
func StatesReader(s States, fetch Fetch) Reader {
|
|
return statesReader{s: s, fetch: fetch, kept: &keptTakeIns{}}
|
|
}
|
|
|
|
type statesReader struct {
|
|
s States
|
|
fetch Fetch
|
|
kept *keptTakeIns
|
|
}
|
|
|
|
func (r statesReader) takenIn(key string) ([]byte, bool) { return r.kept.takenIn(key) }
|
|
|
|
func (r statesReader) keepTakenIn(key string, raw []byte) { r.kept.keepTakenIn(key, raw) }
|
|
|
|
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) {
|
|
keys := delivery.NewPlacementKeys()
|
|
err := r.each(PlacementRead, func(k string, v []byte) {
|
|
_, _ = keys.Put(k, v, 0, false) // a key that does not read places nothing, as before
|
|
})
|
|
return keys.All(), 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
|
|
}
|