jochen
404efb764b
mesh/merge-gate fail: builds mesh-delivery → novox; no bus step; a manifest the change touches fails the module check: mesh-delivery claims mesh-delivery a…
mesh/repo-check fail: its merge-check.sh failed: files asking for a setting without saying whether it is trusted: 25
mesh/delivery superseded: a newer head of the same pull request
The operator's budget (a leaf module running everywhere within five minutes of its merge, a core module within ten) cannot be held to without measuring it. Each delivery now keeps its delivery time, class and phases from the walk the controller says; 'times' answers the week's median and worst per class, the deliveries over budget and their longest phase, and one delivery's breakdown; 'deliveries' is headed by the same line. A phase the walk could not measure stays unknown, never a zero. Measurement only.
342 lines
14 KiB
Go
342 lines
14 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, looked := time.Now(), time.Time{}
|
|
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)
|
|
}
|
|
}
|
|
// Every open pull request, from the forge itself, every five minutes (novox/hq issue 347).
|
|
if now.Sub(looked) >= 5*time.Minute {
|
|
looked = now
|
|
h.SeeOpenPulls()
|
|
}
|
|
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) {
|
|
// Headed by the week's delivery times (novox/hq ADR 0282 decision 6).
|
|
return map[string]any{"times": h.TimesLine(),
|
|
"deliveries": h.Deliveries(strArg(a, "state"), strArg(a, "repository"), strArg(a, "group"), boolArg(a, "all"))}, nil
|
|
}},
|
|
{Name: seat + "times",
|
|
Description: "Delivery time (novox/hq ADR 0282): from a merge to every machine of its walk running the build. " +
|
|
"Each class's budget (a leaf module five minutes, a core module ten), the last days' median and worst, " +
|
|
"how many within and over, the newest delivery of each class, every delivery over its budget with its " +
|
|
"longest phase, and the delivered ones whose time is not known and why. Given a delivery's id, its time " +
|
|
"phase by phase: window, queued, word, build, send-first, judgement, send-rest, between-tiers, apply — " +
|
|
"each measured, unknown (never a zero) or none.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{
|
|
"days": str("how many days back (default 7)"), "id": str("one delivery's id: its phases")}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
days, err := intArg(a, "days")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return h.Times(days, strArg(a, "id"))
|
|
}},
|
|
{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: seat + "retire-history",
|
|
Description: "Retire held deliveries that are history, on a person's word, with why: each made from the " +
|
|
"forge's word for a merge on the trunk heard 15 minutes or more after it happened — replayed when this " +
|
|
"owner first held the seat — with no walk of it open. Never released, delivered or put back: retired, " +
|
|
"final. One by id, refused when the rule does not take it; or every held one (of one repository when " +
|
|
"named), answering how many it retired and naming those it left held. dry \"true\" answers what it " +
|
|
"would do and changes nothing.",
|
|
Input: map[string]any{"type": "object", "properties": map[string]any{
|
|
"why": str("why, kept with every transition"), "id": str("one delivery's id; every held one when empty"),
|
|
"repository": str("owner/repository: only its held deliveries"),
|
|
"dry": map[string]any{"type": "string", "enum": []string{"true", "false"}}},
|
|
"required": []string{"why"}},
|
|
Run: func(a map[string]any) (any, error) {
|
|
if err := need(a, "why"); err != nil {
|
|
return nil, err
|
|
}
|
|
return h.RetireHistory(strArg(a, "id"), strArg(a, "repository"), strArg(a, "why"), aPerson,
|
|
boolArg(a, "dry"))
|
|
}},
|
|
{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 }},
|
|
}
|
|
}
|