jochen
6ae112d299
A group whose order rules disagreed was refused as "its order contradicts itself" with only the member ids: nobody could see which rules clashed or what to write to resolve it. The controller now resolves rules by precedence (hq ADR 0249) and answers, per pair, the rules it won over, and per contradiction no precedence resolves, both rules and how to declare the order. mesh-delivery keeps both, lists them in groups and the plan note, and refuses a group by those words; a cycle through more members names the pairs in it. An older controller's answer reads as before.
300 lines
12 KiB
Go
300 lines
12 KiB
Go
// mesh-delivery: the holder of the mesh-delivery seat (novox/hq ADR 0239, to-be 47). A Go bundle the node's
|
|
// runtime launches. It owns the delivery — one commit in one repository, from its pull request's head to
|
|
// every machine — and the delivery group above it: their states in one compiled table, kept in its own state
|
|
// on the bus, every transition said as an event, noted on the commit and shown on the pull request through
|
|
// the forge's holder. It never sends to a machine: it asks the controller, through its seat's verbs. stdout
|
|
// is the MCP channel; what this module says, it says on stderr.
|
|
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
|
|
|
|
// listening is whether the events this owner follows reach it, in words.
|
|
type listening struct {
|
|
mu sync.Mutex
|
|
now string
|
|
}
|
|
|
|
func (l *listening) set(s string) { l.mu.Lock(); l.now = s; l.mu.Unlock() }
|
|
func (l *listening) get() string { l.mu.Lock(); defer l.mu.Unlock(); return l.now }
|
|
|
|
// The events this owner follows: the forge's pull requests, and the controller's verdicts and walks.
|
|
var followed = []string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed",
|
|
ControllerSeat + ".checked", ControllerSeat + ".plan-moved"}
|
|
|
|
func main() {
|
|
h := &Holder{
|
|
Store: kvStore{},
|
|
Controller: seatController{ask: stdio.Ask},
|
|
Forge: toolForge{ask: stdio.Ask},
|
|
Emit: func(event string, body any) error { return stdio.Emit(event, body) },
|
|
Logf: logf,
|
|
}
|
|
h.init()
|
|
l := &listening{now: "not yet: starting"}
|
|
go run(h, l)
|
|
if err := stdio.Serve("", tools(h, l)); err != nil {
|
|
logf("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// run reads back the state, then the controller's walks, then takes the events, and keeps time — each
|
|
// retried, each failure said, never given up on quietly.
|
|
func run(h *Holder, l *listening) {
|
|
time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := h.Load()
|
|
if err == nil {
|
|
break
|
|
}
|
|
logf("[mesh-delivery] cannot read back its deliveries yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
if err := h.Reconcile(); err != nil {
|
|
logf("[mesh-delivery] the controller's walks cannot be read yet (%v); asked again each half minute", err)
|
|
}
|
|
go func() {
|
|
tick := time.NewTicker(10 * time.Second)
|
|
defer tick.Stop()
|
|
last := time.Now()
|
|
for now := range tick.C {
|
|
if now.Sub(last) >= 30*time.Second {
|
|
last = now
|
|
if err := h.Reconcile(); err != nil {
|
|
logf("[mesh-delivery] the controller's walks cannot be read: %v", err)
|
|
}
|
|
}
|
|
h.Tick()
|
|
h.Flush()
|
|
}
|
|
}()
|
|
for _, pattern := range followed {
|
|
pattern := pattern
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := stdio.Subscribe(pattern, func(e stdio.Envelope) error { return handle(h, e) })
|
|
if err == nil {
|
|
break
|
|
}
|
|
l.set("not yet: " + err.Error())
|
|
logf("[mesh-delivery] not hearing %s yet (%v); asking again in %s", pattern, err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
l.set("listening")
|
|
logf("[mesh-delivery] listening: %s", strings.Join(followed, ", "))
|
|
}
|
|
|
|
// handle takes one event. An event that cannot be read is said and acknowledged: read again it would fail
|
|
// again, and the controller's walks are read back by comparison anyway.
|
|
func handle(h *Holder, e stdio.Envelope) error {
|
|
switch e.Key {
|
|
case "gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed":
|
|
var p PullEvent
|
|
if err := json.Unmarshal(e.Body, &p); err != nil {
|
|
h.Refuse(fmt.Errorf("%s is not readable: %v", e.Key, err))
|
|
return nil
|
|
}
|
|
switch e.Key {
|
|
case "gitea.pull.updated":
|
|
h.PullUpdated(p)
|
|
case "gitea.pull.merged":
|
|
h.PullMerged(p)
|
|
default:
|
|
h.PullClosed(p)
|
|
}
|
|
case ControllerSeat + ".checked":
|
|
var c CheckedEvent
|
|
if err := json.Unmarshal(e.Body, &c); err != nil {
|
|
h.Refuse(fmt.Errorf("a verdict is not readable: %v", err))
|
|
return nil
|
|
}
|
|
h.Checked(c)
|
|
case ControllerSeat + ".plan-moved":
|
|
var w Walk
|
|
if err := json.Unmarshal(e.Body, &w); err != nil {
|
|
h.Refuse(fmt.Errorf("a walk is not readable: %v", err))
|
|
return nil
|
|
}
|
|
h.WalkMoved(w)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func str(description string) map[string]any {
|
|
return map[string]any{"type": "string", "description": description}
|
|
}
|
|
|
|
func strArg(a map[string]any, k string) string { s, _ := a[k].(string); return strings.TrimSpace(s) }
|
|
|
|
// intArg is a whole number given as a number or as text; absent is zero.
|
|
func intArg(a map[string]any, k string) (int, error) {
|
|
switch v := a[k].(type) {
|
|
case nil:
|
|
return 0, nil
|
|
case float64:
|
|
if v == float64(int(v)) {
|
|
return int(v), nil
|
|
}
|
|
case int:
|
|
return v, nil
|
|
case string:
|
|
v = strings.TrimPrefix(strings.TrimSpace(v), "#")
|
|
if v == "" {
|
|
return 0, nil
|
|
}
|
|
if n, err := strconv.Atoi(v); err == nil {
|
|
return n, nil
|
|
}
|
|
}
|
|
return 0, fmt.Errorf("%q is not a whole number: %v", k, a[k])
|
|
}
|
|
|
|
func boolArg(a map[string]any, k string) bool {
|
|
switch v := a[k].(type) {
|
|
case bool:
|
|
return v
|
|
case string:
|
|
return v == "true"
|
|
}
|
|
return false
|
|
}
|
|
|
|
// aPerson is who an act through the seat is by: the seat does not name its caller.
|
|
const aPerson = "a person, through the mesh-delivery seat"
|
|
|
|
func need(a map[string]any, keys ...string) error {
|
|
for _, k := range keys {
|
|
if strArg(a, k) == "" {
|
|
return fmt.Errorf("needs %q", k)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func tools(h *Holder, l *listening) []stdio.Tool {
|
|
seat := "mesh-delivery."
|
|
return []stdio.Tool{
|
|
{Name: seat + "deliveries",
|
|
Description: "Every delivery not final, and those that ended in the last day, one line each: its id " +
|
|
"(owner/repository@commit), its state, its group, what it waits for and since when. Narrowed by state, " +
|
|
"repository or group.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"state": str("one state"),
|
|
"repository": str("owner/repository"), "group": str("a group's id (its branch name)"),
|
|
"all": map[string]any{"type": "string", "enum": []string{"true", "false"}}}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
return h.Deliveries(strArg(a, "state"), strArg(a, "repository"), strArg(a, "group"), boolArg(a, "all")), nil
|
|
}},
|
|
{Name: seat + "show",
|
|
Description: "One delivery or group whole: its delivery plan, every transition with when and why, the " +
|
|
"machine steps of its walk, its group and its order.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("a delivery's id, or a group's")},
|
|
"required": []string{"id"}},
|
|
Run: func(a map[string]any) (any, error) { return h.Show(strArg(a, "id")) }},
|
|
{Name: seat + "groups",
|
|
Description: "Every delivery group: its members in order, why each pair is ordered and the rules it won " +
|
|
"over, each contradiction no precedence resolves with how to declare the order, its composed check " +
|
|
"and its state, derived from its members.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{
|
|
"all": map[string]any{"type": "string", "enum": []string{"true", "false"}}}},
|
|
Run: func(a map[string]any) (any, error) { return h.Groups(boolArg(a, "all")), nil }},
|
|
{Name: seat + "what-if",
|
|
Description: "The delivery plan a change would have, asked of the controller's planner and kept nowhere.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"repository": str("owner/repository"),
|
|
"paths": str("the files it changes, comma-separated"), "base": str("the branch it merges into (default main)")},
|
|
"required": []string{"repository", "paths"}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if err := need(a, "repository", "paths"); err != nil {
|
|
return nil, err
|
|
}
|
|
return h.WhatIf(strArg(a, "repository"), strArg(a, "base"), strings.Split(strArg(a, "paths"), ","))
|
|
}},
|
|
{Name: seat + "checks",
|
|
Description: "What the mesh's checks said of a pull request's head or of one commit: each of the commit's " +
|
|
"statuses (mesh/merge-gate, mesh/repo-check, mesh/delivery, …) with its state, description and when it " +
|
|
"was set; the merge check's full verdict as the controller said it — each layer's summary, the machine " +
|
|
"that ran it, when, its build id and its report; and whether the branch's protection would let it " +
|
|
"merge, every required status being success.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"repository": str("owner/repository"),
|
|
"number": str("a pull request's number: its head is read"),
|
|
"commit": str("a commit's sha, or the start of one, instead of a pull request")},
|
|
"required": []string{"repository"}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if err := need(a, "repository"); err != nil {
|
|
return nil, err
|
|
}
|
|
number, err := intArg(a, "number")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return h.Checks(strArg(a, "repository"), number, strArg(a, "commit"))
|
|
}},
|
|
{Name: seat + "table",
|
|
Description: "The state table every delivery runs by: each transition with its guard, each state's bound " +
|
|
"and what healer H2 may do once it has passed; and the machine steps' table.",
|
|
Run: func(map[string]any) (any, error) { return TableText(), nil }},
|
|
{Name: seat + "stalled",
|
|
Description: "Every delivery held past its state's bound, with the transition the table lets healer H2 take.",
|
|
Run: func(map[string]any) (any, error) { return h.Stalled(), nil }},
|
|
{Name: seat + "recheck",
|
|
Description: "Check a rejected or ready delivery again: its merge check statuses put back to pending on the forge, " +
|
|
"so nothing merges on the old verdict; proposed again, and its check asked. With why.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
|
|
"why": str("why, kept with the transition")}, "required": []string{"id", "why"}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if err := need(a, "id", "why"); err != nil {
|
|
return nil, err
|
|
}
|
|
return h.Recheck(strArg(a, "id"), strArg(a, "why"), aPerson)
|
|
}},
|
|
{Name: seat + "release",
|
|
Description: "A person's word that a held delivery goes on: it starts delivering. With why.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
|
|
"why": str("why, kept with the transition")}, "required": []string{"id", "why"}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if err := need(a, "id", "why"); err != nil {
|
|
return nil, err
|
|
}
|
|
return h.Release(strArg(a, "id"), strArg(a, "why"), aPerson)
|
|
}},
|
|
{Name: seat + "stop",
|
|
Description: "Stop a delivery that is not final, with why: its walk is ended through the controller, and " +
|
|
"in a delivering group every member after it is stopped too, naming it.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
|
|
"why": str("why, kept with the transition")}, "required": []string{"id", "why"}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if err := need(a, "id", "why"); err != nil {
|
|
return nil, err
|
|
}
|
|
return h.Stop(strArg(a, "id"), strArg(a, "why"), aPerson)
|
|
}},
|
|
{Name: seat + "close",
|
|
Description: "Healer H2's verb: for a delivery held past its bound, read its walk again and take the " +
|
|
"transition the table names for that state. Refused for any other.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
|
|
"why": str("the condition it answers")}, "required": []string{"id", "why"}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if err := need(a, "id", "why"); err != nil {
|
|
return nil, err
|
|
}
|
|
return h.Close(strArg(a, "id"), strArg(a, "why"))
|
|
}},
|
|
{Name: "delivery_status",
|
|
Description: "Whether this owner works: how many deliveries in each state, the groups, what is owed to " +
|
|
"the forge and the bus and not yet done, walks with no delivery yet, whether its events arrive, when " +
|
|
"the controller's walks were last read or why they could not be, and the last refusals.",
|
|
Run: func(map[string]any) (any, error) { return h.Status(l.get()), nil }},
|
|
}
|
|
}
|