jochen
b6f0bc309b
One module answers 'did my change go out' for a commit and orders a cross-repository change: one compiled state table, its state on the bus, every transition said, noted on the commit and shown on the pull request. The forge's holder gains the note, view and status tools it asks with, and says closed pull requests and a merge's head and statuses.
254 lines
10 KiB
Go
254 lines
10 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"
|
|
"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) }
|
|
|
|
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, 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 + "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: 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 }},
|
|
}
|
|
}
|