374 lines
14 KiB
Go
374 lines
14 KiB
Go
// mesh-delivery-gate: 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 asks the Approve of a merge and merges once; it asks the controller nothing and derives nothing
|
|
// after the merge (ADR 0539 §17). 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"
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
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. The forge's status events say that a check answered (events.go).
|
|
var followed = []string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed",
|
|
"gitea.forge.status", "gitea.forge.pull-request.edited", "gitea.forge.pull-request.closed",
|
|
forgePattern, ConfigChangedEvent}
|
|
|
|
// forgePattern is the forge bridge's messages on FORGE (novox/hq ADR 0437 §8): the gate follows a merge and the refusal of one.
|
|
const forgePattern = "mesh-delivery-forge-bridge.forge.**"
|
|
|
|
func main() {
|
|
settingsFile := os.Getenv("MESH_DELIVERY_GATE_SETTINGS")
|
|
var saidMu sync.Mutex
|
|
lastSaid := ""
|
|
settings := func() Settings {
|
|
s, err := readSettings(settingsFile)
|
|
saidMu.Lock()
|
|
defer saidMu.Unlock()
|
|
if err != nil && err.Error() != lastSaid {
|
|
logf("[mesh-delivery-gate] settings cannot be read, the defaults hold: %v", err)
|
|
}
|
|
lastSaid = ""
|
|
if err != nil {
|
|
lastSaid = err.Error()
|
|
}
|
|
return s
|
|
}
|
|
h := &Holder{
|
|
Settings: settings,
|
|
Store: kvStore{},
|
|
Forge: toolForge{ask: stdio.Ask},
|
|
Emit: func(event string, body any) error { return stdio.Emit(event, body) },
|
|
Logf: logf,
|
|
Operator: kvOperator{},
|
|
Merges: kvMerges{},
|
|
Source: kvSource{},
|
|
Owed: kvOwed{},
|
|
Pulls: kvPulls{},
|
|
}
|
|
h.init()
|
|
l := &listening{now: "not yet: starting"}
|
|
go run(h, l)
|
|
if err := announceKept(stdio.Keeps); err != nil {
|
|
logf("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
if err := stdio.Serve("", tools(h, l)); err != nil {
|
|
logf("%v", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// run takes the events and the operator's answers. The state is read back at the start (start), and what the gate may have
|
|
// missed meanwhile is read once then (readOnStart): the forge's open pull requests, the operator's asks — and it acts on
|
|
// events from then on (novox/hq ADR 0406 §1, issue 975): no tick, no forge poll, no reconcile loop. Each step is 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
|
|
loadKept(h, l)
|
|
go h.Work()
|
|
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)
|
|
}
|
|
}
|
|
// The teller's answers: a change of its bucket asks for a look, which reads the answers to the asks pending.
|
|
go func() {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := stdio.State(sdkdelivery.TellerModule+".answers").Watch("", func(stdio.StateChange) error { h.AnswersChanged(); return nil })
|
|
if err == nil {
|
|
return
|
|
}
|
|
logf("[mesh-delivery] not hearing the teller's answers yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
}()
|
|
// The checker's verdicts (issue 1302, ADR 0447 §2): the Approve on mesh/check is asked only on its verdict at exactly the
|
|
// heads an ask covers, and a group's verdict is that one; each change asks for a look.
|
|
go func() {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := stdio.State(checkerVerdicts).Watch("", func(c stdio.StateChange) error {
|
|
if c.Op == "delete" {
|
|
h.CheckerVerdictChanged(c.Key, nil)
|
|
} else {
|
|
h.CheckerVerdictChanged(c.Key, c.Value)
|
|
}
|
|
return nil
|
|
})
|
|
if err == nil {
|
|
return
|
|
}
|
|
logf("[mesh-delivery] not hearing the checker's verdicts yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
}()
|
|
l.set("listening")
|
|
logf("[mesh-delivery] listening: %s", strings.Join(followed, ", "))
|
|
readOnStart(h)
|
|
}
|
|
|
|
// loadKept reads the kept deliveries and groups back, retrying until it can: no event is taken before.
|
|
func loadKept(h *Holder, l *listening) {
|
|
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
|
err := h.Load()
|
|
if err == nil {
|
|
break
|
|
}
|
|
l.set("not yet: " + err.Error())
|
|
logf("[mesh-delivery] the kept state cannot be read yet (%v); asking again in %s", err, wait)
|
|
time.Sleep(wait)
|
|
}
|
|
}
|
|
|
|
// readOnStart reads once what happened while the gate was away: the forge's open pull requests (heads nobody checks) are
|
|
// read by the first look, and the asks the router closed meanwhile are read here.
|
|
func readOnStart(h *Holder) {
|
|
h.wantForge()
|
|
h.hearAsks()
|
|
h.SourceCatchUp() // a merge missed or marked unwritten while the gate was away reaches source.main, and so the checker (issue 1726)
|
|
}
|
|
|
|
// handle takes one event, then asks for a look at every step the event makes possible (Work, React). An event that cannot be read is
|
|
// said and acknowledged: read again it would fail again.
|
|
func handle(h *Holder, e stdio.Envelope) error {
|
|
taken := take(h, e)
|
|
h.Poke()
|
|
return taken
|
|
}
|
|
|
|
func take(h *Holder, e stdio.Envelope) error {
|
|
if strings.HasPrefix(e.Key, "mesh-delivery-forge-bridge.forge.") {
|
|
h.Forged(e.Key, e.Body)
|
|
return nil
|
|
}
|
|
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":
|
|
if h.PullUpdated(p) {
|
|
h.wantForge() // a head not seen before: the forge shows whether anyone checks it
|
|
}
|
|
h.KeepPull(p)
|
|
case "gitea.pull.merged":
|
|
h.PullMerged(p)
|
|
default:
|
|
h.PullClosed(p)
|
|
}
|
|
case "gitea.forge.pull-request.edited":
|
|
// The description edited after a head was announced (issue 1079): the review record in it is read again.
|
|
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
|
|
}
|
|
h.PullEdited(p)
|
|
case "gitea.forge.pull-request.closed":
|
|
// The forge's own close event (issue 1107): the poll's gitea.pull.closed skips a repository whose update time did not
|
|
// move, and a close does not move it. A merged pull request is the webhook's merged event, never this one.
|
|
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
|
|
}
|
|
if !p.Merged {
|
|
h.PullClosed(p)
|
|
}
|
|
case "gitea.forge.status":
|
|
var s ForgeStatusEvent
|
|
if err := json.Unmarshal(e.Body, &s); err != nil {
|
|
h.Refuse(fmt.Errorf("a commit status is not readable: %v", err))
|
|
return nil
|
|
}
|
|
h.CommitStatus(s)
|
|
h.KeepCheckVerdict(s)
|
|
case ConfigChangedEvent:
|
|
var c ConfigChange
|
|
if err := json.Unmarshal(e.Body, &c); err != nil {
|
|
h.Refuse(fmt.Errorf("a config change is not readable: %v", err))
|
|
return nil
|
|
}
|
|
h.ConfigChanged(c)
|
|
}
|
|
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 := Seat + "."
|
|
list := []stdio.Tool{
|
|
{Name: seat + "deliveries",
|
|
Description: "Every delivery not final, those that ended in the last day, and any stop `stalled` still lists, 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 map[string]any{
|
|
"deliveries": h.Deliveries(strArg(a, "state"), strArg(a, "repository"), strArg(a, "group"), boolArg(a, "all"))}, nil
|
|
}},
|
|
{Name: seat + "show",
|
|
Description: "One delivery or group whole: every transition with when and why, and its group. " +
|
|
"What the merge moved is source.main; whether it reached the machines is the view's.",
|
|
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, its composed check and its state, derived from its members. " +
|
|
"A group has no order: it is merged in one pass, and moves source.main once (novox/hq ADR 0418, ADR 0539 clause 17); what " +
|
|
"reached the machines is the delivery view's. On mesh/check its check is the checker's one verdict at every member's head " +
|
|
"(ADR 0447); composed_gate says its members' gate is the controller's composed check's (ADR 0403), for a base still on it.",
|
|
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 + "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/check, mesh/delivery, …) with its state, description and when it " +
|
|
"was set; the merge check's verdict as the checker's mesh/check line says it, and when it was heard; 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; ",
|
|
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. " +
|
|
"Also a merge refused or not made past its bound, a group half merged, and a head with no merge check.",
|
|
Run: func(a map[string]any) (any, error) { return h.Stalled(), nil }},
|
|
{Name: seat + "recheck",
|
|
Description: "Refused for every delivery: the checker checks a head again on every move of the head, its trunk or its " +
|
|
"group, and nothing is asked of anyone here (novox/hq ADR 0447, issue 1318). Kept so a caller is told where to read.",
|
|
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 + "close",
|
|
Description: "Healer H2's verb: for a proposed delivery past its bound, 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, how many owed effects were tried since it started (done, failed " +
|
|
"and tried again, given up) and when the last was, whether its events arrive, and the last refusals.",
|
|
Run: func(map[string]any) (any, error) { return h.Status(l.get()), nil }},
|
|
}
|
|
// A verb that changes a delivery asks for a look at what it makes possible (react.go).
|
|
for i := range list {
|
|
run := list[i].Run
|
|
list[i].Run = func(a map[string]any) (any, error) {
|
|
out, err := run(a)
|
|
h.Poke()
|
|
return out, err
|
|
}
|
|
}
|
|
return list
|
|
}
|