package main import ( "context" "encoding/json" "errors" "fmt" "strings" "sync" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/inventory" ) // **Where the controller reads config values once the move has run** (novox/hq ADR 0348 §9, design 57 K5, issue 664): // mesh-config_resolved, the bucket mesh-config alone writes, keyed `..`, each value with the // source it came from. Read only for a module the switch reaches (inventory.ReadsConfig), so a controller before the // move never asks the bus for config. The exception to ADR 0348 §9 ("the controller holds no config value") that // ADR 0349 clause 17's old way needs, until design 57's R1 hands every module to the provisioner. // resolvedEntry is one key of mesh-config_resolved as the controller reads it: the value and where it came from. type resolvedEntry struct { Node string `json:"node"` Module string `json:"module"` Key string `json:"key"` Value json.RawMessage `json:"value,omitempty"` Source string `json:"source"` } // The sources a resolved value is laid over a module's files from: mesh config, node config and the module's own // value. Its manifest's default is composition's already, and a taken key's default is never laid over its file; // a required value nothing sets (`unset`) is refused by composition by name, as it is today. var laidSources = map[string]bool{"mesh": true, "node": true, "module": true} // configLaid is the values of one module on one node that are laid over its files, from the resolved entries. func configLaid(entries []resolvedEntry, node, module string) (map[string]any, error) { out := map[string]any{} for _, e := range entries { if e.Node != node || e.Module != module || !laidSources[e.Source] { continue } var v any if err := json.Unmarshal(e.Value, &v); err != nil { return nil, fmt.Errorf("mesh-config's value %s of %s on %s cannot be read: %w", e.Key, module, node, err) } out[e.Key] = v } return out, nil } // resolvedRead is what one read of the whole bucket held, and when it was read. type resolvedRead struct { mu sync.Mutex at time.Time entries []resolvedEntry } // resolvedFresh is how long one read of the bucket answers: a plan reads every module of a machine, and one read // serves them all; a value set since is read by the next composition. const resolvedFresh = 5 * time.Second var theResolved resolvedRead // readConfigValues is where an opened inventory reads config values once the switch is on; a test replaces it. var readConfigValues inventory.ConfigValues = busConfigValues // busConfigValues reads one module's config values on one node from mesh-config_resolved (inventory.ConfigValues). func busConfigValues(ctx context.Context, node, module string) (map[string]any, error) { theResolved.mu.Lock() defer theResolved.mu.Unlock() if theResolved.entries == nil || time.Since(theResolved.at) > resolvedFresh { entries, err := readResolved(ctx) if err != nil { return nil, err } theResolved.entries, theResolved.at = entries, time.Now() } return configLaid(theResolved.entries, node, module) } // readResolved is every entry of mesh-config_resolved, read on the bus. func readResolved(ctx context.Context) ([]resolvedEntry, error) { entries := []resolvedEntry{} err := onTheBus(func(conn *nats.Conn) error { js, err := jetstream.New(conn) if err != nil { return err } ctx, cancel := context.WithTimeout(ctx, 20*time.Second) defer cancel() kv, err := js.KeyValue(ctx, broker.ConfigResolvedBucket) if err != nil { return fmt.Errorf("%s cannot be opened: %w", broker.ConfigResolvedBucket, err) } w, err := kv.WatchAll(ctx, jetstream.IgnoreDeletes()) if err != nil { return err } defer func() { _ = w.Stop() }() for { select { case <-ctx.Done(): return fmt.Errorf("%s was not read whole: %w", broker.ConfigResolvedBucket, ctx.Err()) case e, open := <-w.Updates(): if !open { return errors.New(broker.ConfigResolvedBucket + " stopped answering") } if e == nil { return nil } var r resolvedEntry if err := json.Unmarshal(e.Value(), &r); err != nil { return fmt.Errorf("%s holds %q, which is not a resolved value: %w", broker.ConfigResolvedBucket, e.Key(), err) } if r.Node == "" || r.Module == "" || r.Key == "" || !strings.HasPrefix(e.Key(), r.Node+"."+r.Module+".") { return fmt.Errorf("%s holds %q, whose value names another key", broker.ConfigResolvedBucket, e.Key()) } entries = append(entries, r) } } }) return entries, err }