717 lines
25 KiB
Go
717 lines
25 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"regexp"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// A merge into main waits for the operator's Approve (novox/hq ADR 0398 §3, issue 948, ADR 0421).
|
|
//
|
|
// This owner writes the ask into its bucket asks (novox/hq ADR 0437 §6) as soon as a pull request's checks pass:
|
|
// Approve or Decline, each option binding the act "merge these heads". The teller publishes it to the operator on the
|
|
// router and, on a warrant that holds the ask (its asker, the ask's digest, an option it offered, in time, and the act
|
|
// that option bound), writes the answer into its bucket answers. This owner reads the answer and acts on it only while
|
|
// the heads of the key it was written under are still its group's heads; it asks the router nothing.
|
|
//
|
|
// - Asked once per head. A group is asked once, for all its members' heads together, by the first member that reaches
|
|
// this point; the Approve covers every member at those heads.
|
|
// - A push after the ask ends it, said once on the pull request: the new head is asked on its own.
|
|
// - A Decline merges nothing and is said once on the pull request.
|
|
// - An ask that ended without a choice (expired, cancelled, refused by the router) is asked again when the router says so.
|
|
// - The ask, its state and the warrant's words are kept on the delivery's ReadyRun and shown by `show`.
|
|
//
|
|
// The terminal Approve (ADR 0398 §3) is issue 953; today the proving channel is the operator's phone.
|
|
|
|
// Module is this module's name: the one the bus names its buckets and events under (novox/hq ADR 0437 §3). Seat is
|
|
// the seat it claims, whose name stays `mesh-delivery` (ADR 0437 §11).
|
|
const (
|
|
Module = "mesh-delivery-gate"
|
|
Seat = "mesh-delivery"
|
|
)
|
|
|
|
// The two answers, both at the level approve.
|
|
const (
|
|
OptionApprove = "approve"
|
|
OptionDecline = "decline"
|
|
)
|
|
|
|
// AskLasts is how long the operator has to answer: a day less a margin, under the router's bound for an approval.
|
|
const AskLasts = asks.ApproveLasts - 10*time.Minute
|
|
|
|
// Operator is how this owner asks the operator: through the keys of its bucket asks and the teller's answers. kvOperator in
|
|
// the running module.
|
|
type Operator interface {
|
|
// Ask writes the ask under its key (sdk.MergeAskKey); it writes nothing when that key holds the same ask already.
|
|
Ask(key string, a sdk.Ask) error
|
|
// Cancel takes the ask at a key away, which the teller reads as the ask withdrawn.
|
|
Cancel(key string) error
|
|
// Where is the teller's answer to the ask at a key, or nil when there is none.
|
|
Where(key string) (*sdk.Answer, error)
|
|
}
|
|
|
|
// askGroup is the name a delivery's ask and merge are kept under: its group's, or, for a lone pull request, a group of one
|
|
// under its branch name (novox/hq ADR 0437 §6).
|
|
func askGroup(d *Delivery) string {
|
|
switch {
|
|
case d.Group != "":
|
|
return d.Group
|
|
case d.Branch != "":
|
|
return d.Branch
|
|
}
|
|
return d.ID
|
|
}
|
|
|
|
// headsString is the heads an act names, as an act's value: "<id>=<sha>" for each, sorted, joined by commas.
|
|
func headsString(heads map[string]string) string {
|
|
var out []string
|
|
for id, sha := range heads {
|
|
out = append(out, id+"="+sha)
|
|
}
|
|
sort.Strings(out)
|
|
return strings.Join(out, ",")
|
|
}
|
|
|
|
// mergeAct is what an answer binds: this ask, the answer, and the heads it covers.
|
|
func mergeAct(ask, option string, heads map[string]string) asks.Act {
|
|
return asks.Act{"verb": Seat + ".merge", "ask": ask, "option": option, "heads": headsString(heads)}
|
|
}
|
|
|
|
var unshownPath = regexp.MustCompile(`[/\\@~$]|\d+\.\d+\.\d+|[A-Za-z0-9+/=_]{24,}|(?i)\b[a-z0-9-]+\.[a-z]{2,12}\b`)
|
|
|
|
// phone is a line the operator's phone may be shown: what the router refuses in an ask's words (a path, an address, a
|
|
// host-like name, a long unbroken word) is left out, and the line is cut to n characters.
|
|
func phone(s string, n int) string {
|
|
s = strings.Join(strings.Fields(unshownPath.ReplaceAllString(s, "…")), " ")
|
|
if r := []rune(s); len(r) > n {
|
|
return string(r[:n-1]) + "…"
|
|
}
|
|
return s
|
|
}
|
|
|
|
var issueNamed = regexp.MustCompile(`(?i)\bissues?\s*#?(\d+)`)
|
|
|
|
// IssuesIn reads the issues a pull request's title and body name ("issue 948"), ascending, once each.
|
|
func IssuesIn(text string) []int {
|
|
seen := map[int]bool{}
|
|
var out []int
|
|
for _, m := range issueNamed.FindAllStringSubmatch(text, -1) {
|
|
if n, err := strconv.Atoi(m[1]); err == nil && !seen[n] {
|
|
seen[n] = true
|
|
out = append(out, n)
|
|
}
|
|
}
|
|
sort.Ints(out)
|
|
return out
|
|
}
|
|
|
|
func issuesSaid(ds []*Delivery) string {
|
|
seen := map[int]bool{}
|
|
var out []string
|
|
for _, d := range ds {
|
|
for _, n := range d.Issues {
|
|
if !seen[n] {
|
|
seen[n] = true
|
|
out = append(out, "issue "+strconv.Itoa(n))
|
|
}
|
|
}
|
|
}
|
|
if len(out) == 0 {
|
|
return "none named"
|
|
}
|
|
return strings.Join(out, ", ")
|
|
}
|
|
|
|
func repoName(repository string) string {
|
|
_, name, ok := strings.Cut(repository, "/")
|
|
if !ok {
|
|
return repository
|
|
}
|
|
return name
|
|
}
|
|
|
|
func pullSaid(d *Delivery) string { return fmt.Sprintf("%s#%d", repoName(d.Repository), d.Number) }
|
|
|
|
// composeMergeAsk is the ask for a delivery, or for a group's members together.
|
|
func composeMergeAsk(id string, d *Delivery, group string, members []*Delivery, heads map[string]string, now time.Time) asks.Ask {
|
|
var a asks.Ask
|
|
a.ID, a.Who, a.Expires = id, asks.Operator, now.Add(AskLasts)
|
|
a.OnExpiry = "nothing is merged; the ask is made again"
|
|
a.About = "delivery." + phone(repoName(d.Repository), 40) + "." + strconv.Itoa(d.Number)
|
|
a.About = strings.ReplaceAll(a.About, " ", "-")
|
|
var reviewers, reaches, details []string
|
|
unreviewed := ""
|
|
for _, m := range members {
|
|
words, ok := reviewWords(m)
|
|
if !ok && unreviewed == "" {
|
|
unreviewed = pullSaid(m) + " is " + words
|
|
}
|
|
reviewers = append(reviewers, fmt.Sprintf("%s: %s", pullSaid(m), words))
|
|
plan := "no delivery plan yet"
|
|
if m.Plan != nil && m.Plan.Summary != "" {
|
|
plan = m.Plan.Summary
|
|
}
|
|
reaches = append(reaches, plan)
|
|
details = append(details, fmt.Sprintf("%s %q at %s: issues: %s; review: %s; reaches: %s",
|
|
pullSaid(m), m.Title, shortOf(m.Commit, 12), issuesSaid([]*Delivery{m}), words, plan))
|
|
}
|
|
if group == "" {
|
|
a.Headline = phone("Merge into main: "+pullSaid(d)+" "+d.Title, asks.HeadlineLength)
|
|
summary := fmt.Sprintf("%s %s passed its checks. Approve merges it into main at %s; Decline leaves it unmerged.",
|
|
pullSaid(d), d.Title, shortOf(d.Commit, 12))
|
|
if unreviewed != "" {
|
|
summary = strings.ToUpper(unreviewed[:1]) + unreviewed[1:] + ". " + summary
|
|
}
|
|
a.Explanation = strings.Join([]string{
|
|
phone(summary, asks.SummaryLength),
|
|
phone("Issues: "+issuesSaid(members), asks.PointLength),
|
|
phone("Review: "+reviewers[0], asks.PointLength),
|
|
phone("Reaches: "+reaches[0], asks.PointLength),
|
|
}, "\n")
|
|
} else {
|
|
var names []string
|
|
for _, m := range members {
|
|
names = append(names, pullSaid(m))
|
|
}
|
|
a.Headline = phone("Merge into main: group "+group+" ("+strings.Join(names, ", ")+")", asks.HeadlineLength)
|
|
summary := fmt.Sprintf("The group %s, %d pull requests, passed its checks. Approve merges all of them into main in one pass, at the heads named; Decline leaves them unmerged.",
|
|
group, len(members))
|
|
if unreviewed != "" {
|
|
summary = strings.ToUpper(unreviewed[:1]) + unreviewed[1:] + ". " + summary
|
|
}
|
|
a.Explanation = strings.Join([]string{
|
|
phone(summary, asks.SummaryLength),
|
|
phone("Issues: "+issuesSaid(members), asks.PointLength),
|
|
phone("Review: "+strings.Join(reviewers, "; "), asks.PointLength),
|
|
phone("Reaches: "+strings.Join(reaches, "; "), asks.PointLength),
|
|
}, "\n")
|
|
}
|
|
a.Details = phone(strings.Join(details, "\n"), 1500)
|
|
if len(details) > 0 { // keep the lines apart: phone joined them
|
|
a.Details = strings.Join(func() []string {
|
|
var out []string
|
|
for _, l := range details {
|
|
out = append(out, phone(l, 400))
|
|
}
|
|
return out
|
|
}(), "\n")
|
|
}
|
|
for _, o := range []struct{ id, label, does string }{
|
|
{OptionApprove, "Approve", "the pull request is merged into main at the head named (a group: all its members, in one pass)"},
|
|
{OptionDecline, "Decline", "nothing is merged"},
|
|
} {
|
|
binds, _ := asks.ActDigest(mergeAct(id, o.id, heads))
|
|
a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds})
|
|
}
|
|
return a
|
|
}
|
|
|
|
// askMembers are the deliveries one ask covers: the group's members not yet merged, by id, or the delivery.
|
|
func (h *Holder) askMembers(d *Delivery) []*Delivery {
|
|
g := h.groups[d.Group]
|
|
if d.Group == "" || g == nil {
|
|
return []*Delivery{d}
|
|
}
|
|
var out []*Delivery
|
|
seen := map[string]bool{}
|
|
ids := append([]string(nil), g.Members...)
|
|
sort.Strings(ids)
|
|
for _, id := range ids {
|
|
if o := h.deliveries[id]; o != nil && !seen[id] && o.MergedAs == "" && !o.State.Final() {
|
|
seen[id] = true
|
|
out = append(out, o)
|
|
}
|
|
}
|
|
if len(out) == 0 {
|
|
return []*Delivery{d}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// covering is the ready run holding an ask (pending or approved) that covers d at its head: its own, or a group
|
|
// sibling's, whose ask named d's head too.
|
|
func (h *Holder) covering(d *Delivery) *ReadyRun {
|
|
has := func(o *Delivery) *ReadyRun {
|
|
if o != nil && o.Ready != nil && (o.Ready.AskID != "" || o.Ready.Approved != nil) && o.Ready.Heads[d.ID] == d.Commit {
|
|
return o.Ready
|
|
}
|
|
return nil
|
|
}
|
|
if r := has(d); r != nil {
|
|
return r
|
|
}
|
|
if g := h.groups[d.Group]; d.Group != "" && g != nil {
|
|
for _, id := range g.Members {
|
|
if r := has(h.deliveries[id]); r != nil {
|
|
return r
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func headsEqual(a, b map[string]string) bool {
|
|
if len(a) != len(b) {
|
|
return false
|
|
}
|
|
for k, v := range a {
|
|
if b[k] != v {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// headsHold is whether an ask's heads are still the heads of d's group: every member now at the head asked, and every
|
|
// head asked that is no member now a member already merged.
|
|
func (h *Holder) headsHold(d *Delivery, c *ReadyRun) bool {
|
|
now := map[string]bool{}
|
|
for _, m := range h.askMembers(d) {
|
|
now[m.ID] = true
|
|
if c.Heads[m.ID] != m.Commit {
|
|
return false
|
|
}
|
|
}
|
|
for id := range c.Heads {
|
|
if !now[id] {
|
|
if o := h.deliveries[id]; o == nil || o.MergedAs == "" {
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// endStaleAsks ends every ask whose heads are no longer its group's: a push to any member ends the approval (ADR 0398
|
|
// §3), and says so once on each pull request it covered (ADR 0421 §6). An ask for a head a newer one replaced ends too.
|
|
// A Decline stands, and is not ended. Under the lock; the cancels run with the look's asks.
|
|
func (h *Holder) endStaleAsks() {
|
|
for _, d := range h.deliveries {
|
|
r := d.Ready
|
|
if r == nil || r.Merged != nil || d.MergedAs != "" || r.Declined != nil || (r.AskID == "" && r.Approved == nil) || len(r.Heads) == 0 {
|
|
continue
|
|
}
|
|
switch {
|
|
case d.State.Final():
|
|
h.cancels = append(h.cancels, h.dropAsk(r.AskID, "its head "+shortOf(d.Commit, 12)+" is no longer the pull request's head, or the pull request closed"))
|
|
case !h.headsHold(d, r):
|
|
h.cancels = append(h.cancels, h.dropAsk(r.AskID, "a head it covers moved on: the heads it covers are no longer the group's heads"))
|
|
case r.Approved == nil && r.AskID != "" && !r.AskedAt.IsZero() && !h.Now().Before(r.AskedAt.Add(AskLasts)):
|
|
// The ask lapsed at the router with nobody's answer, and nothing in the gate's inputs says so: it is taken away and
|
|
// written again, which the teller asks afresh (ADR 0441 §1).
|
|
h.cancels = append(h.cancels, h.withdrawAsk(r.AskID, "nobody answered in "+AskLasts.String()))
|
|
}
|
|
}
|
|
}
|
|
|
|
// approval is what a delivery that may merge waits on for the operator: empty once its head is approved, and the ask
|
|
// to make when none is made yet. Under the lock.
|
|
func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) {
|
|
r := d.Ready
|
|
c := h.covering(d)
|
|
if c != nil {
|
|
if c.Approved != nil {
|
|
r.Approved, r.Warrant, r.AskID, r.Ask, r.Heads = c.Approved, c.Warrant, c.AskID, c.Ask, c.Heads
|
|
return "", nil
|
|
}
|
|
if !c.AskedAt.IsZero() {
|
|
h.lookAt(c.AskedAt.Add(AskLasts), "the operator's Approve, unanswered until the ask lapses")
|
|
}
|
|
return "the operator's Approve is asked (" + c.AskID + ")", nil
|
|
}
|
|
if r.Refused != "" {
|
|
return "the operator's answer was refused here: " + r.Refused, nil
|
|
}
|
|
if h.Operator == nil {
|
|
return "the operator cannot be asked from here", nil
|
|
}
|
|
members := h.askMembers(d)
|
|
group := ""
|
|
if len(members) > 1 {
|
|
group = d.Group
|
|
}
|
|
heads := map[string]string{}
|
|
var covered []sdk.Head
|
|
for _, m := range members {
|
|
heads[m.ID] = m.Commit
|
|
covered = append(covered, sdk.Head{Repository: m.Repository, Number: m.Number, Head: m.Commit})
|
|
}
|
|
name := askGroup(d)
|
|
id, err := sdk.MergeAskKey(name, covered)
|
|
if err != nil {
|
|
return "the operator cannot be asked: " + err.Error(), nil
|
|
}
|
|
r.Asks++
|
|
// Each asking at these heads has its own router id (sdk.AskingID): the router keeps an ask by its id and drops a second ask
|
|
// under it as a redelivery, so an ask written again at unchanged heads (a lapse, a status or the review record changing) is
|
|
// the next asking, with a new id (review round 1, G2; issue 1370).
|
|
asking := 1
|
|
for _, m := range members {
|
|
if m.Ready != nil && m.Ready.AskedKey == id {
|
|
asking = max(asking, m.Ready.Asking+1)
|
|
}
|
|
}
|
|
ask := composeMergeAsk(sdk.AskingID(id, asking), d, group, members, heads, now)
|
|
taken := []*Delivery{}
|
|
var reads []memberRead
|
|
for _, m := range members {
|
|
h.runOf(m, now)
|
|
m.Ready.AskID, m.Ready.Ask, m.Ready.Heads, m.Ready.AskedAt, m.Ready.Asks = id, &ask, heads, now, max(m.Ready.Asks, r.Asks)
|
|
m.Ready.AskedKey, m.Ready.Asking = id, asking
|
|
h.keep(m)
|
|
taken = append(taken, m)
|
|
reads = append(reads, readOf(m, m.Commit))
|
|
}
|
|
h.Logf("[mesh-delivery] %s passed its checks: the operator is asked to Approve its merge into main (%s, ask %d)",
|
|
d.ID, id, r.Asks)
|
|
gid := ""
|
|
if group != "" {
|
|
gid = d.Group
|
|
}
|
|
return "the operator's Approve is asked (" + id + ")", func() {
|
|
// A group is not asked while a member's required status is not a success at its head (issue 1275): the group
|
|
// says which, and is asked once every member may merge.
|
|
var holdMember, holdSays string
|
|
var holdOwn bool
|
|
var readErr error
|
|
for _, m := range reads {
|
|
if gid == "" {
|
|
break
|
|
}
|
|
says, ownOnly, err := h.notASuccess(m)
|
|
if err != nil {
|
|
readErr = err
|
|
break
|
|
}
|
|
if says != "" {
|
|
holdMember, holdSays, holdOwn = m.id, says, ownOnly
|
|
break
|
|
}
|
|
}
|
|
undo := func() {
|
|
for _, m := range taken {
|
|
if m.Ready != nil && m.Ready.AskID == id {
|
|
m.Ready.AskID, m.Ready.Ask, m.Ready.Heads = "", nil, nil
|
|
h.keep(m)
|
|
}
|
|
}
|
|
}
|
|
// An earlier asking of this key may still stand in the bucket: not written again, it is taken away.
|
|
stale := func() {
|
|
h.mu.Lock()
|
|
standing := h.standing(id)
|
|
h.mu.Unlock()
|
|
if standing {
|
|
if err := h.Operator.Cancel(id); err != nil {
|
|
h.Logf("[mesh-delivery] the operator's ask %s could not be withdrawn (%v)", id, err)
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
h.markWritten(id, false)
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
if readErr != nil || holdSays != "" {
|
|
stale()
|
|
h.mu.Lock()
|
|
var own func()
|
|
// A member's own repository check pending or missing (a sibling's new head put it out of date, issue 1792) is
|
|
// asked again here, once for the head: nothing else asks it, and the composed check sets no repository check.
|
|
if d := h.deliveries[holdMember]; holdOwn && d != nil && d.Ready != nil {
|
|
own = h.askOwnAfterRefusal(d)
|
|
}
|
|
defer func() {
|
|
h.mu.Unlock()
|
|
if own != nil {
|
|
own()
|
|
}
|
|
}()
|
|
undo()
|
|
if g := h.groups[gid]; g != nil && holdSays != "" {
|
|
if g.Hold == nil || g.Hold.Kind != "unasked" || g.Hold.Says != holdSays || g.Hold.Member != holdMember {
|
|
h.Logf("[mesh-delivery] group %s is not asked for its Approve: %s waits on a required status: %s", gid, holdMember, holdSays)
|
|
g.Hold = &GroupHold{Kind: "unasked", Member: holdMember, Says: holdSays, Since: h.Now()}
|
|
h.keepGroup(g)
|
|
}
|
|
}
|
|
if readErr != nil {
|
|
h.Logf("[mesh-delivery] group %s: the required statuses cannot be read, so the operator is not asked yet: %v", gid, readErr)
|
|
}
|
|
h.again("a group's Approve held for a required status")
|
|
return
|
|
}
|
|
if g := func() *Group { h.mu.Lock(); defer h.mu.Unlock(); return h.groups[gid] }(); g != nil && g.Hold != nil && g.Hold.Kind == "unasked" {
|
|
h.mu.Lock()
|
|
g.Hold = nil
|
|
h.keepGroup(g)
|
|
h.mu.Unlock()
|
|
}
|
|
if err := h.Operator.Ask(id, sdk.Ask{Group: name, Heads: covered, Asking: asking, Ask: ask}); err == nil {
|
|
h.mu.Lock()
|
|
h.markWritten(id, true)
|
|
h.mu.Unlock()
|
|
} else {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
undo()
|
|
h.again("the operator's Approve, which could not be asked")
|
|
h.Logf("[mesh-delivery] the operator could not be asked about %s (%v); asked again in %s", d.ID, err, askAgainAfter)
|
|
}
|
|
}
|
|
}
|
|
|
|
// dropAsk forgets a pending ask on every delivery it covers and takes it back from the router; under the lock. What it
|
|
// says is said once on each pull request it covered (ADR 0421 §6): the Approve is ended, and why.
|
|
func (h *Holder) dropAsk(id, why string) func() {
|
|
var said []func()
|
|
seen := map[string]bool{}
|
|
var covered []*Delivery
|
|
for _, d := range h.deliveries {
|
|
if d.Ready != nil && d.Ready.AskID == id {
|
|
covered = append(covered, d)
|
|
}
|
|
}
|
|
sort.Slice(covered, func(i, j int) bool { return covered[i].ID < covered[j].ID })
|
|
for _, d := range covered {
|
|
d.Ready.AskID, d.Ready.Ask, d.Ready.Heads = "", nil, nil
|
|
if d.Ready.Merged == nil {
|
|
d.Ready.Approved, d.Ready.Warrant = nil, ""
|
|
}
|
|
h.keep(d)
|
|
if k := fmt.Sprintf("%s#%d", d.Repository, d.Number); !seen[k] && d.Number != 0 {
|
|
seen[k] = true
|
|
said = append(said, h.sayOn(d, "mesh-delivery: the operator's Approve of this pull request's merge into main is ended: "+why+
|
|
". Nothing is merged on it. The operator is asked again once the checks pass at the new head."))
|
|
}
|
|
}
|
|
h.Logf("[mesh-delivery] the operator's ask %s is cancelled: %s", id, why)
|
|
return func() {
|
|
for _, s := range said {
|
|
s()
|
|
}
|
|
if h.Operator == nil {
|
|
return
|
|
}
|
|
if err := h.Operator.Cancel(id); err != nil {
|
|
h.Logf("[mesh-delivery] the operator's ask %s could not be cancelled (%v)", id, err)
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
h.markWritten(id, false)
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
// withdrawAsk takes back an ask nobody answered, on every delivery it covers, so the next look asks again with what is
|
|
// known now (issue 1079). Unlike dropAsk it ends no Approve and says nothing on a pull request: there was none. Under the
|
|
// lock; the router's cancel runs with the look's asks.
|
|
func (h *Holder) withdrawAsk(id, why string) func() {
|
|
for _, d := range h.deliveries {
|
|
if d.Ready != nil && d.Ready.AskID == id && d.Ready.Approved == nil {
|
|
d.Ready.AskID, d.Ready.Ask, d.Ready.Heads = "", nil, nil
|
|
h.keep(d)
|
|
}
|
|
}
|
|
// The key is not taken away: the next look writes it again with the next asking, if the group may be asked, and takes it away
|
|
// only if it may not (readyAsks, withdrawnKeys), so the key is never absent between the two.
|
|
h.withdrawn[id] = true
|
|
h.Logf("[mesh-delivery] the operator's ask %s is withdrawn to be asked again: %s", id, why)
|
|
return func() {}
|
|
}
|
|
|
|
// withdrawnKeys are, after a look, the closures that take away the asks withdrawn and not written again by it. Under the lock.
|
|
func (h *Holder) withdrawnKeys() []func() {
|
|
var out []func()
|
|
for key := range h.withdrawn {
|
|
delete(h.withdrawn, key)
|
|
again := false
|
|
for _, d := range h.deliveries {
|
|
again = again || (d.Ready != nil && d.Ready.AskID == key)
|
|
}
|
|
if again || h.Operator == nil {
|
|
continue
|
|
}
|
|
key := key
|
|
out = append(out, func() {
|
|
if err := h.Operator.Cancel(key); err != nil {
|
|
h.Logf("[mesh-delivery] the operator's ask %s could not be withdrawn (%v)", key, err)
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
h.markWritten(key, false)
|
|
h.mu.Unlock()
|
|
})
|
|
}
|
|
return out
|
|
}
|
|
|
|
// pendingAsks are the asks waiting for an answer, by id; under the lock.
|
|
func (h *Holder) pendingAsks() []string {
|
|
seen := map[string]bool{}
|
|
var out []string
|
|
for _, d := range h.deliveries {
|
|
if r := d.Ready; r != nil && r.AskID != "" && r.Approved == nil && r.Declined == nil && !seen[r.AskID] {
|
|
seen[r.AskID] = true
|
|
out = append(out, r.AskID)
|
|
}
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// hearAsks reads the teller's answer to every pending ask and applies each that is there.
|
|
func (h *Holder) hearAsks() {
|
|
if h.Operator == nil {
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
keys := h.pendingAsks()
|
|
h.mu.Unlock()
|
|
for _, key := range keys {
|
|
answer, err := h.Operator.Where(key)
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] the teller's answer to the ask %s cannot be read: %v", key, err)
|
|
continue
|
|
}
|
|
if answer == nil {
|
|
continue
|
|
}
|
|
h.mu.Lock()
|
|
said := h.answered(key, *answer)
|
|
h.mu.Unlock()
|
|
for _, s := range said {
|
|
s()
|
|
}
|
|
}
|
|
}
|
|
|
|
// answered applies the teller's answer to the ask at a key to the deliveries it covers; under the lock. The teller has
|
|
// held the warrant to the ask (novox/hq ADR 0437 §6); this owner acts on the answer only if the key it was written under is
|
|
// still the key of the heads asked: an answer to an ask taken away, or to other heads, merges nothing (ADR 0441 §1). It
|
|
// returns what to say on the pull requests.
|
|
func (h *Holder) answered(key string, ans sdk.Answer) []func() {
|
|
var covered []*Delivery
|
|
for _, d := range h.deliveries {
|
|
if d.Ready != nil && d.Ready.AskID == key {
|
|
covered = append(covered, d)
|
|
}
|
|
}
|
|
if len(covered) == 0 {
|
|
h.Logf("[mesh-delivery] the answer to the ask %s names no ask that is asked now: nothing is merged on it", key)
|
|
return nil
|
|
}
|
|
sort.Slice(covered, func(i, j int) bool { return covered[i].ID < covered[j].ID })
|
|
first, now := covered[0].Ready, h.Now()
|
|
refuse := func(why string) {
|
|
h.Logf("[mesh-delivery] REFUSED the answer to the ask %s: %s", key, why)
|
|
for _, d := range covered {
|
|
d.Ready.AskID, d.Ready.Ask, d.Ready.Heads = "", nil, nil
|
|
d.Ready.Refused = "was answered, and the answer does not hold here: " + clip(oneLine(why), 300)
|
|
h.keep(d)
|
|
}
|
|
}
|
|
var heads []sdk.Head
|
|
for id, sha := range first.Heads {
|
|
m := h.deliveries[id]
|
|
if m == nil {
|
|
refuse("the head " + id + " it covers is no delivery of this owner")
|
|
return nil
|
|
}
|
|
heads = append(heads, sdk.Head{Repository: m.Repository, Number: m.Number, Head: sha})
|
|
}
|
|
if want, err := sdk.MergeAskKey(askGroup(covered[0]), heads); err != nil || want != key {
|
|
refuse(fmt.Sprintf("its key %s is not the key %s of the heads asked", key, want))
|
|
return nil
|
|
}
|
|
// The answer is to the asking asked now: the teller names the router id it was given in the warrant's id (`<ask id>@<time>`),
|
|
// and an answer to an earlier asking is never taken for a later one (issue 1370).
|
|
if now, at, ok := strings.Cut(ans.Warrant, "@"); ok && at != "" && now != sdk.AskingID(key, first.Asking) {
|
|
h.Logf("[mesh-delivery] the answer to the ask %s was given to an earlier asking (%s), not the one asked now (%s): not taken",
|
|
key, now, sdk.AskingID(key, first.Asking))
|
|
return nil
|
|
}
|
|
if ans.Option != OptionApprove && ans.Option != OptionDecline {
|
|
refuse("it chose " + ans.Option + ", which the ask did not offer")
|
|
return nil
|
|
}
|
|
var said []func()
|
|
for _, d := range covered {
|
|
r := d.Ready
|
|
r.Warrant = ans.Words
|
|
r.Repeats = 0 // the operator chose: whatever the tool refused before is asked afresh (issue 1106)
|
|
h.mergeAgainNow(d) // the operator chose: the merge is asked at once
|
|
switch ans.Option {
|
|
case OptionApprove:
|
|
r.Approved = &now
|
|
h.Logf("[mesh-delivery] %s: the operator approved its merge into main (%s)", d.ID, r.Warrant)
|
|
case OptionDecline:
|
|
r.Declined = &now
|
|
h.Logf("[mesh-delivery] %s: the operator declined its merge into main (%s)", d.ID, r.Warrant)
|
|
said = append(said, h.sayOn(d, fmt.Sprintf("mesh-delivery: **not merged.** The merge of this pull request into main at %s was "+
|
|
"declined: %s. A new head asks the operator again.", d.Commit, r.Warrant)))
|
|
}
|
|
h.keep(d)
|
|
}
|
|
return said
|
|
}
|
|
|
|
// kvOperator is the operator over the runtime: the ask is a key of the gate's bucket asks, the answer a key of the teller's
|
|
// bucket answers.
|
|
type kvOperator struct{}
|
|
|
|
func (kvOperator) Ask(key string, a sdk.Ask) error {
|
|
raw, err := sdk.EncodeAsk(a)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if e, err := stdio.State("asks").Get(key); err == nil && e != nil {
|
|
if kept, err := sdk.DecodeAsk(e.Value); err == nil && sdk.SameAsk(kept, a) {
|
|
return nil // the same ask is there: the teller has it
|
|
}
|
|
}
|
|
_, err = stdio.State("asks").Put(key, json.RawMessage(raw))
|
|
return err
|
|
}
|
|
|
|
func (kvOperator) Cancel(key string) error { return stdio.State("asks").Delete(key) }
|
|
|
|
func (kvOperator) Where(key string) (*sdk.Answer, error) {
|
|
e, err := stdio.State(sdk.TellerModule + ".answers").Get(key)
|
|
if err != nil || e == nil {
|
|
return nil, err
|
|
}
|
|
a, err := sdk.DecodeAnswer(e.Value)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the answer %s is not readable: %w", key, err)
|
|
}
|
|
return &a, nil
|
|
}
|
|
|
|
// standing is whether an ask is written under a key in the gate's asks, as this owner last knew. Under the lock.
|
|
func (h *Holder) standing(key string) bool {
|
|
for _, d := range h.deliveries {
|
|
if d.Ready != nil && d.Ready.AskedKey == key && d.Ready.Written {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// markWritten keeps whether an ask stands under a key, on every delivery that asked it. Under the lock.
|
|
func (h *Holder) markWritten(key string, written bool) {
|
|
for _, d := range h.deliveries {
|
|
if d.Ready != nil && d.Ready.AskedKey == key && d.Ready.Written != written {
|
|
d.Ready.Written = written
|
|
h.keep(d)
|
|
}
|
|
}
|
|
}
|