jochen
a73b04843d
mesh/merge-gate pass: builds mesh-issues → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered
mesh/delivery-group group feat/399-fixing-status delivered: every member is delivered
The note's by is who acted as the runtime proves it (ADR 0275 §5); the note's text names both workers. Tests for a resolved issue corrected, link and set keeping the worker current, and a take-over reason with a secret's shape. The ready-to-verify note says an issue resolves from fixing.
461 lines
16 KiB
Go
461 lines
16 KiB
Go
package main
|
|
|
|
// Whether an issue's fix is delivered (novox/hq ADR 0281). The tracker hears mesh-delivery's `transition` event
|
|
// for every pull request an issue's fixed-by names and keeps, per pull request, the state the delivery last said
|
|
// — merged or not, held and why, covered by a later merge, delivered. From that it derives one line for `get` and
|
|
// `list`: a measured fact, never set by a verb. Delivered is not resolved: when every fix of an issue not yet
|
|
// resolved is delivered, the issue gains a note and the event `ready-to-verify`, and an agent proves its
|
|
// acceptance live before resolving it. A fix merged and not delivered past a bound is said on the issue and to
|
|
// the operator's channel.
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// EventReadyToVerify is said once when every fix of an issue not yet resolved is delivered.
|
|
const EventReadyToVerify = "ready-to-verify"
|
|
|
|
// DefaultStuckAfter is how long a fix may be merged and not delivered before it is said stuck: a walk takes
|
|
// minutes, a gate ten at most, and a hold waits for a person; two hours is past every ordinary delivery and
|
|
// still the same working day.
|
|
const DefaultStuckAfter = 2 * time.Hour
|
|
|
|
// Fix is one pull request of an issue's fixed-by, as the delivery last said it.
|
|
type Fix struct {
|
|
PullRequest string `json:"pull-request"`
|
|
State string `json:"state"`
|
|
Why string `json:"why,omitempty"`
|
|
Merged bool `json:"merged,omitempty"`
|
|
MergedAt *time.Time `json:"merged-at,omitempty"`
|
|
Since time.Time `json:"since"`
|
|
// StuckSaid says it was said stuck, on the issue and to the operator's channel.
|
|
StuckSaid bool `json:"stuck-said,omitempty"`
|
|
}
|
|
|
|
// reached says whether a delivery's state says the change reached the machines (mesh-delivery's Reached).
|
|
func reached(state string) bool {
|
|
return state == "delivered" || state == "covered" || state == "retired"
|
|
}
|
|
|
|
// afterMerge says whether a state is one a delivery reaches only once merged.
|
|
func afterMerge(state string) bool {
|
|
switch state {
|
|
case "published", "delivering", "held", "delivered", "failed", "stopped", "covered", "retired":
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func fixOf(is *Issue, pr string) *Fix {
|
|
for i := range is.Fixes {
|
|
if is.Fixes[i].PullRequest == pr {
|
|
return &is.Fixes[i]
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// allDelivered says whether an issue names a pull request and every one is delivered.
|
|
func allDelivered(is *Issue) bool {
|
|
prs := pullRequests(is)
|
|
if len(prs) == 0 {
|
|
return false
|
|
}
|
|
for _, pr := range prs {
|
|
if !contains(is.Delivered, pr) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// DeliveryLine is the derived delivery state of an issue's fix, in words; empty when fixed-by names no pull
|
|
// request.
|
|
func DeliveryLine(is *Issue) string {
|
|
prs := pullRequests(is)
|
|
if len(prs) == 0 {
|
|
return ""
|
|
}
|
|
var held, failed, merged, open, unheard []string
|
|
covered := false
|
|
for _, pr := range prs {
|
|
f := fixOf(is, pr)
|
|
switch {
|
|
case contains(is.Delivered, pr):
|
|
covered = covered || (f != nil && f.State == "covered")
|
|
case f == nil:
|
|
unheard = append(unheard, pr)
|
|
case f.State == "held":
|
|
held = append(held, strings.TrimSpace(pr+": "+f.Why))
|
|
case f.State == "failed" || f.State == "stopped":
|
|
failed = append(failed, strings.TrimSpace(pr+" "+f.State+": "+f.Why))
|
|
case f.Merged:
|
|
merged = append(merged, pr)
|
|
default:
|
|
open = append(open, pr)
|
|
}
|
|
}
|
|
switch {
|
|
case len(failed) > 0:
|
|
return "fix not delivered: " + strings.Join(failed, "; ")
|
|
case len(held) > 0:
|
|
return "held: " + strings.Join(held, "; ")
|
|
case len(merged) > 0:
|
|
return "fix merged, not delivered: " + strings.Join(merged, ", ")
|
|
case len(open) > 0:
|
|
return "fix not merged: " + strings.Join(open, ", ")
|
|
case len(unheard) > 0:
|
|
return "nothing heard from the delivery of " + strings.Join(unheard, ", ")
|
|
case covered:
|
|
return "delivered to every machine it should reach (by a later merge that contains it)"
|
|
}
|
|
return "delivered to every machine it should reach"
|
|
}
|
|
|
|
// recordFix keeps what the delivery said of a pull request on every issue whose fixed-by names it (only, when
|
|
// given, on that one issue), and answers the issues it changed.
|
|
func (t *Tracker) recordFix(pr, state, why string, merged bool, at time.Time, only uint64) ([]uint64, error) {
|
|
all, err := t.allIssues()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var changed []uint64
|
|
for n, is := range all {
|
|
if (only != 0 && n != only) || !contains(pullRequests(is), pr) {
|
|
continue
|
|
}
|
|
kept := cut(why, MostReasonKept)
|
|
if f := fixOf(is, pr); f != nil && (at.Before(f.Since) || (f.State == state && f.Why == kept && f.Merged == (merged || f.Merged))) {
|
|
// Older than what is kept, or nothing new: transitions can arrive out of order.
|
|
continue
|
|
}
|
|
if contains(is.Delivered, pr) && !reached(state) {
|
|
// Delivered stays delivered: a later state that has not reached the machines is a newer delivery of
|
|
// the same pull request at most, and an older one is history.
|
|
continue
|
|
}
|
|
if _, err := t.change(n, func(i *Issue) error {
|
|
f := fixOf(i, pr)
|
|
if f == nil {
|
|
i.Fixes = append(i.Fixes, Fix{PullRequest: pr})
|
|
f = &i.Fixes[len(i.Fixes)-1]
|
|
}
|
|
f.State, f.Why, f.Since = state, kept, at
|
|
if merged && !f.Merged {
|
|
f.Merged = true
|
|
m := at
|
|
f.MergedAt = &m
|
|
}
|
|
if reached(state) && !contains(i.Delivered, pr) {
|
|
i.Delivered = append(i.Delivered, pr)
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
return changed, err
|
|
}
|
|
changed = append(changed, n)
|
|
}
|
|
sort.Slice(changed, func(i, j int) bool { return changed[i] < changed[j] })
|
|
return changed, nil
|
|
}
|
|
|
|
// afterFix is what follows a change of an issue's fixes: ready to verify, once, when every fix is delivered and
|
|
// the issue is not yet resolved; a stuck fix delivered at last cleared on the operator's channel; and what
|
|
// waited on it let go.
|
|
func (t *Tracker) afterFix(n uint64) error {
|
|
is, _, err := read(t.Store, n)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := t.clearStuck(is); err != nil {
|
|
t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel (tried again at the next sweep): %v", n, err)
|
|
}
|
|
if allDelivered(is) {
|
|
if unresolved(is) {
|
|
prs := pullRequests(is)
|
|
sort.Strings(prs)
|
|
if strings.Join(is.VerifyFor, ",") != strings.Join(prs, ",") {
|
|
// The note first, kept once by its words; then the claim; then the event. A failure before the claim
|
|
// is tried again at the next offer, and the note is not written twice.
|
|
prefix := fmt.Sprintf("Ready to verify: every fix of this issue is delivered (%s).", strings.Join(prs, ", "))
|
|
if !noted(is, prefix) {
|
|
if _, err := t.note(GroomBy, n, prefix+" Delivered is not resolved: prove the acceptance live, then "+
|
|
"resolve it, from fixing: move it to fixing with by first if it is not there. (novox/hq ADR 0281, ADR 0283)"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
claimed := false
|
|
if _, err := t.change(n, func(i *Issue) error {
|
|
claimed = false
|
|
if strings.Join(i.VerifyFor, ",") == strings.Join(prs, ",") {
|
|
return nil
|
|
}
|
|
i.VerifyFor, claimed = prs, true
|
|
return nil
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if claimed {
|
|
t.emit(EventReadyToVerify, map[string]any{"number": n, "fixed-by": prs, "at": t.now()})
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return t.letGo(n)
|
|
}
|
|
|
|
// stillStuck says whether an issue has a fix said stuck that is still named, not delivered, and the issue not
|
|
// closed as wontfix.
|
|
func stillStuck(is *Issue) bool {
|
|
if is.Status == StatusWontfix {
|
|
return false
|
|
}
|
|
named := pullRequests(is)
|
|
for _, f := range is.Fixes {
|
|
if f.StuckSaid && contains(named, f.PullRequest) && !contains(is.Delivered, f.PullRequest) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// clearStuck clears the warning said for an issue on the operator's channel once nothing it was said for is
|
|
// stuck — delivered, no longer named, or the issue closed as wontfix; one that fails is tried again at the next
|
|
// sweep.
|
|
func (t *Tracker) clearStuck(is *Issue) error {
|
|
if !is.StuckOpen || stillStuck(is) || t.Notify == nil {
|
|
return nil
|
|
}
|
|
if err := t.Notify(stuckKey(is.Number), "", "", "", "", "the fix of issue "+keyOf(is.Number)+" is delivered or no longer waited for", true); err != nil {
|
|
return err
|
|
}
|
|
_, err := t.change(is.Number, func(i *Issue) error { i.StuckOpen = false; return nil })
|
|
return err
|
|
}
|
|
|
|
func stuckKey(n uint64) string { return "issue-tracker.fix-stuck." + keyOf(n) }
|
|
|
|
func prKey(repository string, number int) string {
|
|
_, repo, ok := strings.Cut(repository, "/")
|
|
if !ok {
|
|
repo = repository
|
|
}
|
|
return fmt.Sprintf("%s#%d", strings.ToLower(repo), number)
|
|
}
|
|
|
|
// OnTransition is the handler of mesh-delivery's `transition` event: every state a pull request's delivery moves
|
|
// to is kept on each issue whose fixed-by names it.
|
|
func (t *Tracker) OnTransition(e stdio.Envelope) error {
|
|
var body struct {
|
|
Repository string `json:"repository"`
|
|
Number int `json:"number"`
|
|
To string `json:"to"`
|
|
Why string `json:"why"`
|
|
MergedAs string `json:"merged_as"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
if err := json.Unmarshal(e.Body, &body); err != nil || body.To == "" || body.Number <= 0 {
|
|
return nil
|
|
}
|
|
at := body.At.UTC()
|
|
if at.IsZero() {
|
|
at = t.now()
|
|
}
|
|
merged := body.MergedAs != "" || afterMerge(body.To)
|
|
changed, err := t.recordFix(prKey(body.Repository, body.Number), body.To, body.Why, merged, at, 0)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, n := range changed {
|
|
if err := t.afterFix(n); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Delivery is a pull request's delivery as mesh-delivery's `checks` answers it; a test stubs the asking.
|
|
type Delivery struct {
|
|
State string
|
|
Why string
|
|
// Since is when the delivery entered its state, as it says; zero when it does not.
|
|
Since time.Time
|
|
}
|
|
|
|
// DeliveryState asks mesh-delivery for one pull request's delivery.
|
|
type DeliveryState func(repository string, number string) (Delivery, error)
|
|
|
|
// busDeliveryState asks mesh-delivery's checks over the bus.
|
|
func busDeliveryState(owner string) DeliveryState {
|
|
return func(repository, number string) (Delivery, error) {
|
|
raw, err := stdio.Ask("seat:mesh-delivery.checks", map[string]any{"repository": owner + "/" + repository, "number": number})
|
|
if err != nil {
|
|
return Delivery{}, err
|
|
}
|
|
var out struct {
|
|
Delivery struct {
|
|
State string `json:"state"`
|
|
Waits string `json:"waits"`
|
|
Since time.Time `json:"since"`
|
|
} `json:"delivery"`
|
|
}
|
|
if err := json.Unmarshal(raw, &out); err != nil {
|
|
return Delivery{}, fmt.Errorf("mesh-delivery's checks answered no delivery: %.200s", raw)
|
|
}
|
|
return Delivery{State: out.Delivery.State, Why: out.Delivery.Waits, Since: out.Delivery.Since.UTC()}, nil
|
|
}
|
|
}
|
|
|
|
// OnMoved is the handler of the tracker's own `moved` event: each pull request of the issue's fix not heard of
|
|
// yet is asked of mesh-delivery, at most a few per handling, for a delivery that moved before the issue named
|
|
// it; then what follows its fixes, and an issue closed lets go what waited on it.
|
|
func (t *Tracker) OnMoved(e stdio.Envelope) error {
|
|
var body struct {
|
|
Number uint64 `json:"number"`
|
|
}
|
|
if err := json.Unmarshal(e.Body, &body); err != nil || body.Number == 0 {
|
|
return nil
|
|
}
|
|
is, _, err := read(t.Store, body.Number)
|
|
if errors.Is(err, ErrNoIssue) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if t.Deliveries != nil {
|
|
// Each ask up to the bus's 30 seconds, under the runtime's two minutes; one that fails, or more left to
|
|
// ask, has the event offered again, so a delivery never asked ends in max-deliveries, not in silence.
|
|
asked := 0
|
|
for _, pr := range pullRequests(is) {
|
|
if fixOf(is, pr) != nil {
|
|
continue
|
|
}
|
|
if asked == MostDeliveryAsks {
|
|
return fmt.Errorf("issue %d names more pull requests than are asked at once; the rest at the next offer", body.Number)
|
|
}
|
|
asked++
|
|
repo, number, _ := strings.Cut(pr, "#")
|
|
d, err := t.Deliveries(repo, number)
|
|
if err != nil {
|
|
return fmt.Errorf("the delivery of %s (issue %d) could not be asked: %v", pr, body.Number, err)
|
|
}
|
|
state := d.State
|
|
if state == "" {
|
|
// The delivery knows nothing of it yet: not kept, so it is asked again at the next move, and its own
|
|
// events say it when they come.
|
|
continue
|
|
}
|
|
// The delivery's own time, so a transition stamped the same moment is not taken as older.
|
|
since := d.Since
|
|
if since.IsZero() {
|
|
since = t.now()
|
|
}
|
|
if _, err := t.recordFix(pr, state, d.Why, afterMerge(state), since, body.Number); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return t.afterFix(body.Number)
|
|
}
|
|
|
|
// Notifier tells the operator's channel something (operator-channel.notify), or clears it.
|
|
type Notifier func(key, severity, summary, headline, explanation, resolved string, clear bool) error
|
|
|
|
// busNotifier asks the operator's channel over the bus.
|
|
func busNotifier() Notifier {
|
|
return func(key, severity, summary, headline, explanation, resolved string, clear bool) error {
|
|
body := map[string]any{"key": key, "subject": "the issue tracker"}
|
|
if clear {
|
|
body["clear"] = true
|
|
body["resolved"] = resolved
|
|
} else {
|
|
body["severity"], body["summary"], body["headline"], body["explanation"] = severity, summary, headline, explanation
|
|
}
|
|
_, err := stdio.Ask("seat:operator-channel.notify", body)
|
|
return err
|
|
}
|
|
}
|
|
|
|
// stuckAfter is the setting's bound, or the default.
|
|
func (t *Tracker) stuckAfter() time.Duration {
|
|
if s, err := t.Settings(); err == nil && strings.TrimSpace(s.StuckAfter) != "" {
|
|
if d, err := time.ParseDuration(strings.TrimSpace(s.StuckAfter)); err == nil && d > 0 {
|
|
return d
|
|
}
|
|
}
|
|
return DefaultStuckAfter
|
|
}
|
|
|
|
// Sweep says every fix merged and not delivered past the bound: a note on its issue and a warning on the
|
|
// operator's channel, once; it answers how many it said. Run every few minutes.
|
|
func (t *Tracker) Sweep() (int, error) {
|
|
all, err := t.allIssues()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
bound, now := t.stuckAfter(), t.now()
|
|
said := 0
|
|
for n, is := range all {
|
|
if is.Status == StatusWontfix {
|
|
if err := t.clearStuck(is); err != nil {
|
|
t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel: %v", n, err)
|
|
}
|
|
continue
|
|
}
|
|
if err := t.clearStuck(is); err != nil {
|
|
t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel: %v", n, err)
|
|
}
|
|
named := pullRequests(is)
|
|
for _, f := range is.Fixes {
|
|
if !contains(named, f.PullRequest) || contains(is.Delivered, f.PullRequest) || !f.Merged || f.MergedAt == nil ||
|
|
reached(f.State) || f.StuckSaid || now.Sub(*f.MergedAt) < bound {
|
|
continue
|
|
}
|
|
state := f.State
|
|
if f.Why != "" {
|
|
state += ": " + f.Why
|
|
}
|
|
prefix := fmt.Sprintf("Stuck: the fix %s", f.PullRequest)
|
|
if !noted(is, prefix) {
|
|
if _, err := t.note(GroomBy, n, fmt.Sprintf("%s was merged %s ago and is not delivered (%s). A merged fix "+
|
|
"that does not reach the machines is said on the operator's channel. (novox/hq ADR 0281)", prefix,
|
|
now.Sub(*f.MergedAt).Round(time.Minute), state)); err != nil {
|
|
return said, err
|
|
}
|
|
}
|
|
if t.Notify != nil {
|
|
if err := t.Notify(stuckKey(n), "warning",
|
|
fmt.Sprintf("issue %d: fix %s merged %s ago, not delivered", n, f.PullRequest, now.Sub(*f.MergedAt).Round(time.Minute)),
|
|
fmt.Sprintf("A fix is merged and not delivered (issue %d)", n),
|
|
fmt.Sprintf("The fix of issue %d was merged %s ago and has not reached the machines: %s. Look at its "+
|
|
"delivery; nothing else is needed from you unless it waits for a person.", n,
|
|
now.Sub(*f.MergedAt).Round(time.Minute), state), "", false); err != nil {
|
|
// Tried again at the next sweep: the note is not written twice.
|
|
t.Logf("[mesh-issues] the stuck fix of %d could not be said on the operator's channel: %v", n, err)
|
|
continue
|
|
}
|
|
}
|
|
pr := f.PullRequest
|
|
if _, err := t.change(n, func(i *Issue) error {
|
|
if g := fixOf(i, pr); g != nil {
|
|
g.StuckSaid = true
|
|
}
|
|
i.StuckOpen = t.Notify != nil
|
|
return nil
|
|
}); err != nil {
|
|
return said, err
|
|
}
|
|
said++
|
|
}
|
|
}
|
|
return said, nil
|
|
}
|