jschoubben
224a7a66a5
A key a pull request adds is declared by no registered manifest, so its value was refused and the group's composed check could not pass. A proposal may now name the pull request: judged against what the controller told of its head, asked of the operator, kept apart on Approve, applied at the merge, dropped at a close.
340 lines
13 KiB
Go
340 lines
13 KiB
Go
// mesh-config: the owner of every config value, holder of the mesh-config seat (novox/hq ADR 0348 §9, design 57
|
|
// K3). A Go bundle served by a runtime of its own, on its own bus account (`runs-as`, ADR 0259 §8), so that its
|
|
// account alone writes its buckets: mesh-config_values (each value set, at the mesh, a node, or a module on a
|
|
// node), mesh-config_resolved (each module's value on each node, with its source), mesh-config_predefined (the
|
|
// published list of predefined keys) and mesh-config_proposals (each value proposed for the operator's Approve,
|
|
// asked under this module's own name, design 57 K4).
|
|
// It resolves from the controller's mirrors of the assignments, the modules' declarations and the list it kept
|
|
// (design 57 F3, K3). stdout is the MCP channel; what this module says, it says on stderr.
|
|
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
)
|
|
|
|
// Seat is the role this module holds, once for the mesh, and Verb its one verb.
|
|
const (
|
|
Seat = "mesh-config"
|
|
Verb = "config"
|
|
)
|
|
|
|
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
|
|
|
|
// runtimeStore is the bus, through the runtime that launched this bundle.
|
|
type runtimeStore struct{}
|
|
|
|
func (runtimeStore) Get(state, key string) (*Entry, error) {
|
|
e, err := stdio.State(state).Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
return &Entry{Key: e.Key, Value: e.Value, Revision: e.Revision}, nil
|
|
}
|
|
|
|
func (runtimeStore) Put(state, key string, value any) error {
|
|
_, err := stdio.State(state).Put(key, value)
|
|
return err
|
|
}
|
|
|
|
func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) }
|
|
|
|
func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
|
|
|
|
func (runtimeStore) Append(log, key string, value any) error {
|
|
_, err := stdio.Log(log).Append(key, value)
|
|
return err
|
|
}
|
|
|
|
func (runtimeStore) Read(log string) ([]LogEntry, error) {
|
|
var out []LogEntry
|
|
for from := uint64(0); ; {
|
|
page, err := stdio.Log(log).Read(from, "", 1000)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, e := range page.Entries {
|
|
out = append(out, LogEntry{Key: e.Key, Value: e.Value, At: e.At})
|
|
}
|
|
if page.Next == 0 {
|
|
return out, nil
|
|
}
|
|
from = page.Next
|
|
}
|
|
}
|
|
|
|
func (runtimeStore) Emit(event string, body any) error { return stdio.Emit(event, body) }
|
|
|
|
// HandAct records a change in the controller's hand-act log (to-be 45 §7) through its seat's verb.
|
|
func (runtimeStore) HandAct(what, why, cause string) error {
|
|
_, err := stdio.Ask("seat:mesh-controller.hand-act", map[string]any{"what": what, "why": why, "cause": cause})
|
|
return err
|
|
}
|
|
|
|
// runtimeAsks is the operator channel through the runtime: asks published under this module's own name, the
|
|
// router's records of them read under that name alone (its `uses` of the seat, ADR 0259 §3), and the controller's
|
|
// hand-act log for a warrant acted on.
|
|
type runtimeAsks struct{}
|
|
|
|
func (runtimeAsks) Publish(subject string, body any, id string) error {
|
|
return stdio.SeatPublish(subject, body, id)
|
|
}
|
|
func (runtimeAsks) Record(id string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", id) }
|
|
func (runtimeAsks) Kinds() ([]stdio.KindHeld, error) { return stdio.SeatKinds() }
|
|
func (runtimeAsks) Warranted(ask string) error {
|
|
_, err := stdio.Ask("seat:mesh-controller.warranted", map[string]any{"asker": Module, "ask": ask})
|
|
return err
|
|
}
|
|
|
|
func main() {
|
|
c := &Config{Store: runtimeStore{}, Logf: logf, Asks: runtimeAsks{}, Forge: toolForge{ask: stdio.Ask}}
|
|
go resolving(c)
|
|
go hearing(c)
|
|
if err := stdio.Serve("", Tools(c)); err != nil {
|
|
logf("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// resolving resolves at start and again whenever a source moves, and every five minutes besides, so a change
|
|
// heard nowhere is still resolved. The runtime answers only once Serve runs, so the watches are asked again
|
|
// until they are taken.
|
|
func resolving(c *Config) {
|
|
moved := make(chan struct{}, 1)
|
|
poke := func(stdio.StateChange) error {
|
|
select {
|
|
case moved <- struct{}{}:
|
|
default:
|
|
}
|
|
return nil
|
|
}
|
|
for _, read := range []string{AssignmentsRead, DeclaredConfigRead, PredefinedRead} {
|
|
go func(read string) {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
time.Sleep(wait)
|
|
if err := stdio.State(read).Watch("", poke); err != nil {
|
|
logf("[mesh-config] %s is not watched yet (%v); asking again in %s", read, err, wait)
|
|
continue
|
|
}
|
|
return
|
|
}
|
|
}(read)
|
|
}
|
|
tick := time.NewTicker(5 * time.Minute)
|
|
defer tick.Stop()
|
|
time.Sleep(time.Second)
|
|
for sweep := true; ; {
|
|
if sweep {
|
|
// A merge or a close of a pull request a value is held for, heard nowhere, is found here (ADR 0502).
|
|
for _, s := range c.SweepPulls() {
|
|
logf("[mesh-config] %s", s)
|
|
}
|
|
}
|
|
sweep = false
|
|
r, err := c.Reconcile()
|
|
switch {
|
|
case err != nil:
|
|
logf("[mesh-config] the values are not all resolved: %v", err)
|
|
case len(r.Moved) > 0:
|
|
logf("[mesh-config] resolved: %d value(s) moved", len(r.Moved))
|
|
}
|
|
if r.Published != "" {
|
|
logf("[mesh-config] published %s", r.Published)
|
|
}
|
|
select {
|
|
case <-moved:
|
|
time.Sleep(500 * time.Millisecond) // a burst of writes is one resolution
|
|
select {
|
|
case <-moved:
|
|
default:
|
|
}
|
|
case <-tick.C:
|
|
sweep = true
|
|
}
|
|
}
|
|
}
|
|
|
|
// hearing takes the router's word on this module's asks, asked again until the runtime takes the subscription, and
|
|
// reads the router's record of every waiting ask once a minute besides, so an answer heard nowhere is still taken.
|
|
func hearing(c *Config) {
|
|
go func() {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
time.Sleep(wait)
|
|
err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
|
|
var w asks.Warrant
|
|
if json.Unmarshal(e.Body, &w) == nil {
|
|
for _, s := range c.HeardEnd(w) {
|
|
logf("[mesh-config] %s", s)
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
if err == nil {
|
|
logf("[mesh-config] hears the operator's answers")
|
|
return
|
|
}
|
|
logf("[mesh-config] does not hear the operator's answers yet (%v); asking again in %s", err, wait)
|
|
}
|
|
}()
|
|
for _, pattern := range []string{"gitea.pull.merged", "gitea.pull.closed", "gitea.forge.pull-request.closed"} {
|
|
go func(pattern string) {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
time.Sleep(wait)
|
|
err := stdio.Subscribe(pattern, func(e stdio.Envelope) error {
|
|
var p PullEvent
|
|
if json.Unmarshal(e.Body, &p) != nil {
|
|
logf("[mesh-config] %s is not readable", pattern)
|
|
return nil
|
|
}
|
|
for _, s := range c.HeardPull(p, pattern == "gitea.pull.merged") {
|
|
logf("[mesh-config] %s", s)
|
|
}
|
|
return nil
|
|
})
|
|
if err == nil {
|
|
return
|
|
}
|
|
logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait)
|
|
}
|
|
}(pattern)
|
|
}
|
|
tick := time.NewTicker(time.Minute)
|
|
defer tick.Stop()
|
|
for range tick.C {
|
|
said, err := c.Decide()
|
|
if err != nil {
|
|
logf("[mesh-config] reading the operator's answers: %v", err)
|
|
}
|
|
for _, s := range said {
|
|
logf("[mesh-config] %s", s)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Tools is the seat's verb as the runtime serves it.
|
|
func Tools(c *Config) []stdio.Tool {
|
|
return []stdio.Tool{{
|
|
Name: Seat + "." + Verb,
|
|
Description: "Set, clear, list or read the history of config values (novox/hq ADR 0348). A value is set at " +
|
|
"one level: mesh config (no module, no node) or node config (a node) for a predefined key, or a " +
|
|
"module's own value on one node (module and node). Each set or clear needs why, is validated against " +
|
|
"its declaration, names each module on each node whose value moves, and is kept in the history with " +
|
|
"who and why. A trusted value is set only at the controller's terminal or on the operator's Approve: " +
|
|
"propose asks the operator, under mesh-config's own name on a channel that proves who answered, and sets " +
|
|
"the value on Approve if what it was shown against still stands; proposals lists them, or one with " +
|
|
"proposal. A proposal naming the open pull request that introduces its key (pull) is judged against what " +
|
|
"that pull request declares and asked of the operator, and on Approve is held until the pull request " +
|
|
"merges, then set; list shows the held values. pull-declares and held are the controller's. list and " +
|
|
"history are narrowed by module, node and key.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{
|
|
"action": map[string]any{"type": "string",
|
|
"enum": []string{"list", "set", "clear", "history", "propose", "proposals", "pull-declares", "held"},
|
|
"description": "list when not given"},
|
|
"key": map[string]any{"type": "string", "description": "the config key"},
|
|
"value": map[string]any{"description": "for set: the value, of the key's declared type"},
|
|
"module": map[string]any{"type": "string", "description": "a module: its own value, with node"},
|
|
"node": map[string]any{"type": "string", "description": "a node"},
|
|
"why": map[string]any{"type": "string", "description": "for set and clear: why, in a sentence"},
|
|
"by": map[string]any{"type": "string",
|
|
"description": "who gave the value at the controller's terminal: said by the controller alone"},
|
|
"limit": map[string]any{"type": "number", "description": "for history: at most this many, newest"},
|
|
"clear": map[string]any{"type": "boolean",
|
|
"description": "for propose: propose clearing the value at its level instead of setting one"},
|
|
"proposal": map[string]any{"type": "string", "description": "for proposals: one proposal's id, read whole"},
|
|
"pull": map[string]any{"type": "string", "description": "owner/repository#number: for propose, the open " +
|
|
"pull request that introduces the key, so the value is judged against what it declares and held until " +
|
|
"it merges (a module's own value on one node); for pull-declares, the pull request declared for"},
|
|
"head": map[string]any{"type": "string", "description": "for pull-declares: the commit the declarations are read at"},
|
|
"declares": map[string]any{"type": "object", "description": "for pull-declares (the controller's): the " +
|
|
"config declarations the head adds or changes against its base, by module and key"},
|
|
"pulls": map[string]any{"type": "array", "description": "for held (the controller's): the pull requests " +
|
|
"asked about, each {pull, head}", "items": map[string]any{"type": "object"}},
|
|
}},
|
|
Run: func(args map[string]any) (any, error) {
|
|
call, err := callOf(args)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
call.Caller = stdio.CallerOf(args)
|
|
return c.Do(call)
|
|
},
|
|
}}
|
|
}
|
|
|
|
// callOf reads a call's arguments strictly: an argument it does not know is refused, not ignored.
|
|
func callOf(args map[string]any) (Call, error) {
|
|
known := map[string]bool{"action": true, "key": true, "value": true, "module": true, "node": true, "why": true,
|
|
"by": true, "limit": true, "clear": true, "proposal": true, "pull": true, "head": true, "declares": true,
|
|
"pulls": true}
|
|
var unknown []string
|
|
for k := range args {
|
|
if !known[k] {
|
|
unknown = append(unknown, k)
|
|
}
|
|
}
|
|
if len(unknown) > 0 {
|
|
return Call{}, fmt.Errorf("config takes no %s; it takes action, key, value, module, node, why, by, limit, "+
|
|
"clear, proposal, pull, head, declares and pulls",
|
|
strings.Join(unknown, ", "))
|
|
}
|
|
str := func(k string) (string, error) {
|
|
v, ok := args[k]
|
|
if !ok || v == nil {
|
|
return "", nil
|
|
}
|
|
s, ok := v.(string)
|
|
if !ok {
|
|
return "", fmt.Errorf("%s is a string", k)
|
|
}
|
|
return strings.TrimSpace(s), nil
|
|
}
|
|
var call Call
|
|
var problems []error
|
|
for k, into := range map[string]*string{"action": &call.Action, "key": &call.Key, "module": &call.Module,
|
|
"node": &call.Node, "why": &call.Why, "by": &call.By, "proposal": &call.Proposal, "pull": &call.Pull,
|
|
"head": &call.Head} {
|
|
s, err := str(k)
|
|
problems = append(problems, err)
|
|
*into = s
|
|
}
|
|
if v, ok := args["limit"]; ok && v != nil {
|
|
n, ok := v.(float64)
|
|
if !ok || n < 0 || n != float64(int(n)) {
|
|
problems = append(problems, errors.New("limit is a whole number"))
|
|
}
|
|
call.Limit = int(n)
|
|
}
|
|
if v, ok := args["clear"]; ok && v != nil {
|
|
b, ok := v.(bool)
|
|
if !ok {
|
|
problems = append(problems, errors.New("clear is true or false"))
|
|
}
|
|
call.Clear = b
|
|
}
|
|
if v, ok := args["declares"]; ok && v != nil {
|
|
m, ok := v.(map[string]any)
|
|
if !ok {
|
|
problems = append(problems, errors.New("declares is an object: a module to its keys' declarations"))
|
|
}
|
|
call.Declares = m
|
|
}
|
|
if v, ok := args["pulls"]; ok && v != nil {
|
|
raw, _ := json.Marshal(v)
|
|
if err := json.Unmarshal(raw, &call.Pulls); err != nil {
|
|
problems = append(problems, errors.New("pulls is a list of {pull, head}"))
|
|
}
|
|
}
|
|
if v, ok := args["value"]; ok {
|
|
// The value as given, a JSON value: a number for a number, never its text; the validation judges it.
|
|
call.Value = v
|
|
}
|
|
return call, errors.Join(problems...)
|
|
}
|