issue 975: no tick and no reconcile loop; a look after each event, one timer for the steps that wait on the clock
This commit is contained in:
@@ -471,7 +471,7 @@ func (h *Holder) lapsedBehind(w Walk, d *Delivery) string {
|
||||
}
|
||||
|
||||
// lapsedSaid is why a walk whose word stops that lapsed give still waits: the members of their groups not yet
|
||||
// delivered before them, what the last tick held it for, or the controller's refusal of the word. Under the lock.
|
||||
// delivered before them, what the last look held it for, or the controller's refusal of the word. Under the lock.
|
||||
func (h *Holder) lapsedSaid(w Walk) string {
|
||||
var parts []string
|
||||
_, behind := h.lapsedOf(w)
|
||||
@@ -592,6 +592,7 @@ func (h *Holder) words() []func() {
|
||||
}
|
||||
h.deliverErr[walk] = err.Error() // said by `stalled` past the grace (review of novox/hq issue 753)
|
||||
h.Logf("[mesh-delivery] the walk %s could not be let go yet: %v", walk, err)
|
||||
h.again("a walk's word the controller refused or did not answer")
|
||||
return
|
||||
}
|
||||
delete(h.deliverErr, walk)
|
||||
@@ -639,7 +640,7 @@ func (h *Holder) endedAsNothingToMove(w Walk) []string {
|
||||
|
||||
// WalksWaiting is `stalled` asked with `walks` (novox/hq issue 529): for each walk waiting for this owner's word,
|
||||
// what holds it, in the shape of a stalled line naming the walk — each held delivery it carries (State held, its
|
||||
// reason and the verb that releases it), else why its word was not given at the last tick (State published), else
|
||||
// reason and the verb that releases it), else why its word was not given at the last look (State published), else
|
||||
// that no live delivery holds it (State unheld). A walk whose word is being given is not listed. The controller's
|
||||
// `plans` says these instead of pointing at a hand start; the self-check never raises a condition for them.
|
||||
func (h *Holder) WalksWaiting() []StalledLine {
|
||||
|
||||
@@ -22,7 +22,7 @@ import (
|
||||
// this point; the Approve covers every member at those heads.
|
||||
// - A push after the ask cancels it (the mark is cleared with it): the new head needs a new mark and a new Approve.
|
||||
// - A Decline merges nothing, is said once on the pull request, and clears the mark.
|
||||
// - An ask that ended without a choice (expired, cancelled, refused by the router) is asked again at the next tick.
|
||||
// - 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.
|
||||
@@ -262,7 +262,7 @@ func (h *Holder) headsHold(d *Delivery, c *ReadyRun) bool {
|
||||
}
|
||||
|
||||
// endStaleAsks ends every ask whose heads are no longer its group's: a push to any member, marked or not, ends the
|
||||
// approval (ADR 0398 §3). Under the lock; the cancels run with the tick's asks.
|
||||
// approval (ADR 0398 §3). Under the lock; the cancels run with the look's asks.
|
||||
func (h *Holder) endStaleAsks() {
|
||||
for _, d := range h.deliveries {
|
||||
r := d.Ready
|
||||
@@ -326,7 +326,8 @@ func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) {
|
||||
h.keep(m)
|
||||
}
|
||||
}
|
||||
h.Logf("[mesh-delivery] the operator could not be asked about %s (%v); asked again at the next tick", d.ID, err)
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -412,7 +413,7 @@ func (h *Holder) decided(id, state string, w *asks.Warrant) []func() {
|
||||
d.Ready.AskID, d.Ready.Ask, d.Ready.Heads = "", nil, nil
|
||||
h.keep(d)
|
||||
}
|
||||
h.Logf("[mesh-delivery] the operator's ask %s ended without a choice (%s); asked again at the next tick", id, why)
|
||||
h.Logf("[mesh-delivery] the operator's ask %s ended without a choice (%s); asked again at once", id, why)
|
||||
}
|
||||
if w == nil || w.Outcome != asks.OutcomeChosen {
|
||||
why := "the router says " + state
|
||||
|
||||
@@ -356,6 +356,38 @@ func (h *Holder) Flush() {
|
||||
retryAfter(j.e.Tries+1))
|
||||
}
|
||||
}
|
||||
h.owedAgain()
|
||||
}
|
||||
|
||||
// owedAgain says when what is still owed is tried again: an effect the forge or the bus could not take, at the time
|
||||
// retryAfter gave it, and a delivery or group not yet kept, after retryFirst. Flush is the bounded retry of both.
|
||||
func (h *Holder) owedAgain() {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for id := range h.dirty {
|
||||
if h.deliveries[id] != nil {
|
||||
h.lookAt(h.Now().Add(retryFirst), "a delivery not yet kept, kept again before it is said")
|
||||
}
|
||||
}
|
||||
for id := range h.dirtyGroup {
|
||||
if h.groups[id] != nil {
|
||||
h.lookAt(h.Now().Add(retryFirst), "a group not yet kept, kept again before it is said")
|
||||
}
|
||||
}
|
||||
for _, d := range h.deliveries {
|
||||
for _, e := range d.Owed {
|
||||
if e.Next != nil {
|
||||
h.lookAt(*e.Next, "an owed effect the forge or the bus could not take")
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, g := range h.groups {
|
||||
for _, e := range g.Owed {
|
||||
if e.Next != nil {
|
||||
h.lookAt(*e.Next, "an owed effect the forge or the bus could not take")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// An owed effect that fails is tried again later each time: retryFirst after the first failure, doubling up
|
||||
|
||||
@@ -91,17 +91,6 @@ func (h *Holder) PullCommented(e ForgePullEvent) {
|
||||
}
|
||||
}
|
||||
|
||||
// React takes every step the facts allow, after an event that can change what they are: the forge's open pull requests
|
||||
// read when a ready mark may have been made, then the asks of a look (look). Never concurrent with itself.
|
||||
func (h *Holder) React() {
|
||||
h.reactMu.Lock()
|
||||
defer h.reactMu.Unlock()
|
||||
if h.takeForgeWanted() {
|
||||
h.SeeOpenPulls()
|
||||
}
|
||||
h.look()
|
||||
}
|
||||
|
||||
// AskDecided takes the router's `decided` event for an ask this owner made: its warrant is applied to the deliveries
|
||||
// the ask covers, and what is said on their pull requests is said.
|
||||
func (h *Holder) AskDecided(id, state string, w *asks.Warrant) {
|
||||
|
||||
@@ -26,6 +26,7 @@ func hear(t *testing.T, w *world, key string, body any) {
|
||||
if err := handle(w.h, stdio.Envelope{Key: key, Body: raw}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w.h.React() // the look the event pokes, taken by the holder's worker
|
||||
}
|
||||
|
||||
func labeled(number int, label string) ForgePullEvent {
|
||||
@@ -146,6 +147,7 @@ func TestAPassingVerdictAsksTheMergeAtOnceEvenJustAfterARefusal(t *testing.T) {
|
||||
if err := handle(w.h, stdio.Envelope{Key: ControllerSeat + ".checked", Body: raw}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w.h.React()
|
||||
if len(w.forge.merges) != 2 || w.delivery(id).Ready.Merged == nil {
|
||||
t.Fatalf("the passing verdict did not ask the merge at once: %v %+v", w.forge.merges, w.delivery(id).Ready)
|
||||
}
|
||||
|
||||
@@ -106,6 +106,8 @@ type fakeController struct {
|
||||
moves map[string][]string
|
||||
plans int
|
||||
down bool
|
||||
// walkReads counts the reads of the controller's walks (Walks).
|
||||
walkReads int
|
||||
// refuseDeliver refuses every word, as a controller that will not take it does (review of novox/hq issue 753).
|
||||
refuseDeliver bool
|
||||
// onCheck runs as a check is asked, before it is answered: what the forge held at that moment.
|
||||
@@ -347,6 +349,7 @@ func (c *fakeController) cut(id string) Walk {
|
||||
func (c *fakeController) Walks(walk string) (bool, []Walk, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.walkReads++
|
||||
if c.down {
|
||||
return false, nil, errors.New("down")
|
||||
}
|
||||
@@ -604,11 +607,25 @@ func (w *world) later(d time.Duration) {
|
||||
w.mu.Unlock()
|
||||
}
|
||||
|
||||
// settleAll ticks and flushes until nothing changes, a few times.
|
||||
// settleAll lets the router's answers reach the holder as its `decided` events, and takes a look, a few times.
|
||||
func (w *world) settleAll() {
|
||||
for range 4 {
|
||||
w.h.Tick()
|
||||
w.h.Flush()
|
||||
w.decide()
|
||||
w.h.React()
|
||||
}
|
||||
}
|
||||
|
||||
// decide hands the holder the router's `decided` event for every ask of its that has closed.
|
||||
func (w *world) decide() {
|
||||
w.h.mu.Lock()
|
||||
ids := w.h.pendingAsks()
|
||||
w.h.mu.Unlock()
|
||||
for _, id := range ids {
|
||||
state, warrant, _ := w.op.Where(id)
|
||||
if state == "" || state == "open" || state == "held" {
|
||||
continue
|
||||
}
|
||||
w.h.AskDecided(id, state, warrant)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -29,7 +29,7 @@ type Holder struct {
|
||||
Logf func(format string, a ...any)
|
||||
// Operator is the operator channel, asked for the Approve of a merge into main (approval.go).
|
||||
Operator Operator
|
||||
// cancels are asks to take back, gathered under the lock and run with the other asks of the tick.
|
||||
// cancels are asks to take back, gathered under the lock and run with the other asks of the look.
|
||||
cancels []func()
|
||||
|
||||
mu sync.Mutex
|
||||
@@ -55,11 +55,11 @@ type Holder struct {
|
||||
// unchecked are the open pull requests' heads seen without a merge check (unchecked.go).
|
||||
unchecked map[string]unchecked
|
||||
// walks are the walks' and batches' last records (answers.go); orphans the waiting walks no live delivery
|
||||
// holds, by when that was first seen; turning the walks whose turn came this tick, for some deliveries.
|
||||
// holds, by when that was first seen; turning the walks whose turn came this look, for some deliveries.
|
||||
walks map[string]seenWalk
|
||||
orphans map[string]time.Time
|
||||
turning map[string]*turnWant
|
||||
// waitsWhy is, by walk, why the last tick did not give a waiting walk its word: a delivery it carries not at
|
||||
// waitsWhy is, by walk, why the last look did not give a waiting walk its word: a delivery it carries not at
|
||||
// its turn, or a stopped change it carries (novox/hq issue 529). Said by WalksWaiting.
|
||||
waitsWhy map[string]string
|
||||
// asking are the walks whose word is being asked right now, outside the lock: not asked twice meanwhile.
|
||||
@@ -87,7 +87,13 @@ type Holder struct {
|
||||
// forgeWanted is whether the forge's open pull requests are to be read at the next React: a ready mark may have been
|
||||
// made (events.go). reactMu keeps one React at a time.
|
||||
forgeWanted bool
|
||||
// rereadWalks is whether the controller's walks are to be read once more: a merge passed the bound a walk opens
|
||||
// within, and the walks were last read before it (unwalkedPast).
|
||||
rereadWalks bool
|
||||
reactMu sync.Mutex
|
||||
// poke asks for a look (react.go); wanted is when the next is wanted, and its timer.
|
||||
poke chan struct{}
|
||||
wanted lookWanted
|
||||
}
|
||||
|
||||
// owedTries is how many owed effects were tried since this owner started, and when the last was: three numbers
|
||||
@@ -129,6 +135,7 @@ func (h *Holder) init() {
|
||||
if h.Emit == nil {
|
||||
h.Emit = func(string, any) error { return nil }
|
||||
}
|
||||
h.poke = make(chan struct{}, 1)
|
||||
h.deliveries = map[string]*Delivery{}
|
||||
h.groups = map[string]*Group{}
|
||||
h.dirty = map[string]bool{}
|
||||
@@ -251,6 +258,7 @@ func (h *Holder) keep(d *Delivery) {
|
||||
h.dirty[d.ID] = true
|
||||
delete(h.written, d.ID)
|
||||
h.Logf("[mesh-delivery] %s could not be kept yet (%v); kept again before it is said", d.ID, err)
|
||||
h.again("a delivery that could not be kept")
|
||||
return
|
||||
}
|
||||
delete(h.dirty, d.ID)
|
||||
@@ -266,6 +274,7 @@ func (h *Holder) keepGroup(g *Group) {
|
||||
if err := h.Store.PutGroup(g); err != nil {
|
||||
h.dirtyGroup[g.ID] = true
|
||||
h.Logf("[mesh-delivery] group %s could not be kept yet (%v)", g.ID, err)
|
||||
h.again("a group that could not be kept")
|
||||
return
|
||||
}
|
||||
delete(h.dirtyGroup, g.ID)
|
||||
@@ -1005,9 +1014,11 @@ const carriedBy = "carried by "
|
||||
// which says, for the log, why it is asked: what the walk is to the delivery that asks.
|
||||
func (h *Holder) askWalk(id, which string, now time.Time) func() {
|
||||
if asked, was := h.walkAsked[id]; was && now.Sub(asked) < time.Hour {
|
||||
h.lookAt(asked.Add(time.Hour), "a walk's record asked of the controller by its id")
|
||||
return nil
|
||||
}
|
||||
h.walkAsked[id] = now
|
||||
h.lookAt(now.Add(time.Hour), "a walk's record asked of the controller by its id")
|
||||
return func() {
|
||||
_, walks, err := h.Controller.Walks(id)
|
||||
h.mu.Lock()
|
||||
@@ -1068,10 +1079,7 @@ func (h *Holder) Reconcile() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// --- the tick: groups, turns, adoption --------------------------------------------------------------------
|
||||
|
||||
// Tick is React, for the tick that still runs beside the events.
|
||||
func (h *Holder) Tick() { h.React() }
|
||||
// --- the look: groups, turns, adoption --------------------------------------------------------------------
|
||||
|
||||
// look takes every step the facts allow that needs another's word: a group's order and composed check, a
|
||||
// delivery's turn, a walk with no delivery made one. The asking happens outside the lock.
|
||||
@@ -1117,17 +1125,20 @@ func (h *Holder) plan() []func() {
|
||||
if !u.walk.Open() {
|
||||
if now.Sub(u.first) > 10*time.Minute {
|
||||
delete(h.unmatched, id)
|
||||
} else {
|
||||
h.lookAt(u.first.Add(10*time.Minute), "a walk no merge was heard for, ended: forgotten after ten minutes")
|
||||
}
|
||||
continue
|
||||
}
|
||||
if u.walk.Created.After(h.started) && now.Sub(u.first) < matchGrace {
|
||||
h.lookAt(u.first.Add(matchGrace), "a walk no merge was heard for: made a delivery after the grace")
|
||||
continue
|
||||
}
|
||||
h.adopt(u.walk)
|
||||
delete(h.unmatched, id)
|
||||
}
|
||||
|
||||
// Plans not known for a delivery that has its diffset: asked once a tick.
|
||||
// Plans not known for a delivery that has its diffset: asked at each look.
|
||||
for _, d := range h.deliveries {
|
||||
if d.Plan == nil && !d.State.Final() && len(d.Paths) > 0 && (d.State == Ready || d.State == Held || d.MergedAs != "") {
|
||||
d := d
|
||||
@@ -1138,6 +1149,7 @@ func (h *Holder) plan() []func() {
|
||||
defer h.mu.Unlock()
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] %s's plan cannot be asked yet: %v", id, err)
|
||||
h.again("a delivery's plan that could not be asked")
|
||||
return
|
||||
}
|
||||
if d := h.deliveries[id]; d != nil && d.Plan == nil {
|
||||
@@ -1191,14 +1203,14 @@ func (h *Holder) plan() []func() {
|
||||
|
||||
// Held, or merged with no walk opened for it: ended by what is already true of it, never asked of a person
|
||||
// (novox/hq issue 319). A plan that moves nothing is delivered; one whose every module a later merge into the
|
||||
// same trunk delivered is covered. Read again each tick, so a delivery read back after a restart, or held
|
||||
// same trunk delivered is covered. Read again at each look, so a delivery read back after a restart, or held
|
||||
// before this was known, ends by itself.
|
||||
// A published delivery whose walk ended done before its word is read the same way (novox/hq issue 474): what
|
||||
// that walk did not build, a later delivered merge may have; and when the cover test says none did, past the
|
||||
// grace, it is stopped saying so — never left waiting for a start that cannot come.
|
||||
// A delivery that failed or was stopped is read for a later finished walk of its repository that delivered its
|
||||
// change after all: a walk that failed in an outage and was resumed, or the walk of a later merge that contains it
|
||||
// (novox/hq issue 747). Read each tick, from what the walks and the kept deliveries say, so one stuck before this
|
||||
// (novox/hq issue 747). Read at each look, from what the walks and the kept deliveries say, so one stuck before this
|
||||
// was built clears by itself after a restart.
|
||||
h.coverAsked = map[string]bool{}
|
||||
if h.coverAskedSince == nil {
|
||||
@@ -1219,6 +1231,9 @@ func (h *Holder) plan() []func() {
|
||||
}
|
||||
// Held back only within the bound: a forge that keeps erroring never holds a walk for good.
|
||||
h.coverAsked[d.ID] = now.Sub(since) < coverAskBound
|
||||
if h.coverAsked[d.ID] {
|
||||
h.lookAt(since.Add(coverAskBound), "a walk held back while the forge is asked what a later walk delivered")
|
||||
}
|
||||
}
|
||||
if by != "" {
|
||||
d.DeliveredBy = by
|
||||
@@ -1363,6 +1378,7 @@ func (h *Holder) strand(d *Delivery, pending []string, now time.Time) func() {
|
||||
ended = d.Since
|
||||
}
|
||||
if now.Sub(ended) < strandedAfter {
|
||||
h.lookAt(ended.Add(strandedAfter), "a merge its walk did not build: stopped after the grace for a later merge to cover it")
|
||||
return nil
|
||||
}
|
||||
note := ""
|
||||
@@ -1526,8 +1542,20 @@ func (h *Holder) unwalkedPast(d *Delivery, now time.Time) bool {
|
||||
if _, batched := h.batchOf(d); batched {
|
||||
return false // its batch is assembling or queued: a walk is not owed yet (ADR 0276)
|
||||
}
|
||||
return d.State == Ready && d.MergedAs != "" && d.Walk == nil && !d.MovesNothing() && d.HeldWhy == "" &&
|
||||
now.Sub(d.MergedAt) > unwalked && h.reconciled.After(d.MergedAt.Add(unwalked))
|
||||
if d.State != Ready || d.MergedAs == "" || d.Walk != nil || d.MovesNothing() || d.HeldWhy != "" {
|
||||
return false
|
||||
}
|
||||
bound := d.MergedAt.Add(unwalked)
|
||||
if now.Sub(d.MergedAt) <= unwalked {
|
||||
h.lookAt(bound, "a merge no walk opened for within ten minutes")
|
||||
return false
|
||||
}
|
||||
if !h.reconciled.After(bound) {
|
||||
// The walks were not read since the bound: read once now (React), the only read besides the one at the start.
|
||||
h.rereadWalks = true
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// coverOf says which later merges cover a merged delivery, or nothing, and which later merge commits the forge
|
||||
@@ -1633,9 +1661,11 @@ func carries(c, d *Delivery, module string) bool {
|
||||
func (h *Holder) askContains(d *Delivery, in string, now time.Time) func() {
|
||||
key := d.ID + " in " + in
|
||||
if tried, was := h.askTried[key]; was && now.Sub(tried) < time.Minute {
|
||||
h.lookAt(tried.Add(time.Minute), "the forge asked whether a later merge contains this one")
|
||||
return nil
|
||||
}
|
||||
h.askTried[key] = now
|
||||
h.lookAt(now.Add(time.Minute), "the forge asked whether a later merge contains this one")
|
||||
id, owner, repo, merged := d.ID, d.Owner(), d.Repo(), d.MergedAs
|
||||
return func() {
|
||||
yes, err := h.Forge.Contains(owner, repo, merged, in)
|
||||
@@ -1668,14 +1698,19 @@ func heldForGroup(d *Delivery, g *Group) bool {
|
||||
// no check this owner asked, and proposed past the grace the controller's own check of its announcement is
|
||||
// given. Nil otherwise — once asked, a late verdict is the bound's and H2's to speak for.
|
||||
func (h *Holder) askCheck(d *Delivery, now time.Time) func() {
|
||||
if d.State != Proposed || d.Check != nil || d.CheckAsk != nil || d.MergedAs != "" || d.Number == 0 ||
|
||||
now.Sub(d.Since) < checkGrace {
|
||||
if d.State != Proposed || d.Check != nil || d.CheckAsk != nil || d.MergedAs != "" || d.Number == 0 {
|
||||
return nil
|
||||
}
|
||||
if now.Sub(d.Since) < checkGrace {
|
||||
h.lookAt(d.Since.Add(checkGrace), "a proposed delivery's check, asked once the grace has passed with no verdict")
|
||||
return nil
|
||||
}
|
||||
if tried, was := h.askTried[d.ID]; was && now.Sub(tried) < time.Minute {
|
||||
h.lookAt(tried.Add(time.Minute), "a proposed delivery's check that could not be asked")
|
||||
return nil
|
||||
}
|
||||
h.askTried[d.ID] = now
|
||||
h.lookAt(now.Add(time.Minute), "a proposed delivery's check that could not be asked")
|
||||
id, m := d.ID, MemberOf(d)
|
||||
why := fmt.Sprintf("proposed for %s with no verdict and no check asked: its announcement was heard from the "+
|
||||
"bus's history, or the controller missed it", now.Sub(d.Since).Round(time.Minute))
|
||||
@@ -1809,6 +1844,7 @@ func (h *Holder) planGroup(g *Group, members []*Delivery, now time.Time) []func(
|
||||
}
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] group %s's order cannot be asked yet: %v", id, err)
|
||||
h.again("a group's order that could not be asked")
|
||||
return
|
||||
}
|
||||
g.Order, g.Pairs, g.Cycle, g.Contradictions, g.OrderOf = o.Order, o.Pairs, o.Cycle, o.Contradictions, commits
|
||||
@@ -1843,6 +1879,7 @@ func (h *Holder) planGroup(g *Group, members []*Delivery, now time.Time) []func(
|
||||
}
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] group %s's composed check cannot be asked yet: %v", id, err)
|
||||
h.again("a group's composed check that could not be asked")
|
||||
return
|
||||
}
|
||||
g.Check = &GroupCheck{Asked: asked, At: h.Now(), Commits: commits}
|
||||
@@ -1940,6 +1977,7 @@ func (h *Holder) groupTurns(g *Group, members []*Delivery) []func() {
|
||||
if !w.until.IsZero() {
|
||||
u := w.until
|
||||
until = &u
|
||||
h.lookAt(u, "a group member's turn that waits for a later merge to carry it, until the wait's bound")
|
||||
}
|
||||
if d.TurnAwaits != w.said || (d.TurnAwaitsUntil == nil) != (until == nil) ||
|
||||
until != nil && !d.TurnAwaitsUntil.Equal(*until) {
|
||||
@@ -2445,6 +2483,7 @@ func (h *Holder) askedToLeave(walk string) {
|
||||
h.leaveAsked = map[string]leaveAsk{}
|
||||
}
|
||||
h.leaveAsked[walk] = leaveAsk{at: h.Now(), revision: h.walks[walk].walk.Revision}
|
||||
h.lookAt(h.Now().Add(5*time.Minute), "a walk asked to leave a stopped merge out, asked again if it went nowhere")
|
||||
}
|
||||
|
||||
// leaveStopped asks, for every open walk or batch that carries a stopped delivery's merge, that the merge be left
|
||||
@@ -2492,6 +2531,7 @@ func (h *Holder) leaveStopped() []func() {
|
||||
// Once per record of it: the record that answers an ask is newer. Asked again after five minutes, should
|
||||
// the ask have gone nowhere.
|
||||
if a, asked := h.leaveAsked[id]; asked && seen.walk.Revision <= a.revision && now.Sub(a.at) < 5*time.Minute {
|
||||
h.lookAt(a.at.Add(5*time.Minute), "a walk asked to leave a stopped merge out, asked again if it went nowhere")
|
||||
continue
|
||||
}
|
||||
walk := id
|
||||
|
||||
@@ -283,7 +283,7 @@ func TestTheSwitchAdoptsRunningWalksAndHoldsUnknownWaitingOnes(t *testing.T) {
|
||||
if err := w.h.Reconcile(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w.h.Tick()
|
||||
w.h.React()
|
||||
if s := w.state(IDOf("novox/mesh-catalog", "cccccccccccc3333")); s != Delivering {
|
||||
t.Fatalf("a running walk at the switch was adopted as %s", s)
|
||||
}
|
||||
@@ -291,7 +291,7 @@ func TestTheSwitchAdoptsRunningWalksAndHoldsUnknownWaitingOnes(t *testing.T) {
|
||||
t.Fatal("a new walk was made a delivery before its merge had time to be heard")
|
||||
}
|
||||
w.later(3 * time.Minute)
|
||||
w.h.Tick()
|
||||
w.h.React()
|
||||
held := IDOf("novox/app", "dddddddddddd4444")
|
||||
if s := w.state(held); s != Held {
|
||||
t.Fatalf("a waiting walk with no pull request is %s", s)
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asker"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
|
||||
@@ -37,26 +38,29 @@ var followed = []string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.c
|
||||
ControllerSeat + ".checked", ControllerSeat + ".plan-moved"}
|
||||
|
||||
func main() {
|
||||
operator := &asker.Client{Module: Module, State: stdio.State("asked")}
|
||||
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,
|
||||
Operator: &asker.Client{Module: Module, State: stdio.State("asked")},
|
||||
Operator: operator,
|
||||
}
|
||||
h.init()
|
||||
l := &listening{now: "not yet: starting"}
|
||||
go run(h, l)
|
||||
go run(h, l, operator)
|
||||
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) {
|
||||
// run reads back the state, then takes the events, then reads once what it may have missed while it was away — the
|
||||
// controller's walks, the forge's open pull requests, the operator's asks — and 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, operator *asker.Client) {
|
||||
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()
|
||||
@@ -66,30 +70,7 @@ func run(h *Holder, l *listening) {
|
||||
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 minute (novox/hq issues 347, 722): what says a
|
||||
// head nobody checks, and the ready marks a merge is asked on.
|
||||
if now.Sub(looked) >= time.Minute {
|
||||
looked = now
|
||||
h.SeeOpenPulls()
|
||||
}
|
||||
h.Tick()
|
||||
h.Flush()
|
||||
}
|
||||
}()
|
||||
go h.Work()
|
||||
for _, pattern := range followed {
|
||||
pattern := pattern
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
@@ -102,15 +83,50 @@ func run(h *Holder, l *listening) {
|
||||
time.Sleep(wait)
|
||||
}
|
||||
}
|
||||
// The operator's answers: the router's `decided` event, in place of reading every pending ask each tick.
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
err := operator.Listen(
|
||||
func(_ asks.Option, w asks.Warrant) error { h.AskDecided(w.Ask, "answered", &w); h.Poke(); return nil },
|
||||
func(handled asker.Handled, w asks.Warrant, err error) {
|
||||
if err != nil {
|
||||
logf("[mesh-delivery] the answer to the ask %s: %s: %v", w.Ask, handled, err)
|
||||
}
|
||||
if handled == asker.Refused || handled == asker.Recorded {
|
||||
h.AskDecided(w.Ask, string(handled), &w)
|
||||
h.Poke()
|
||||
}
|
||||
})
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
l.set("not yet: " + err.Error())
|
||||
logf("[mesh-delivery] not hearing the operator's answers yet (%v); asking again in %s", err, wait)
|
||||
time.Sleep(wait)
|
||||
}
|
||||
l.set("listening")
|
||||
logf("[mesh-delivery] listening: %s", strings.Join(followed, ", "))
|
||||
// Read once, now that events arrive: what happened while this owner was away. The forge's open pull requests (ready
|
||||
// marks, heads nobody checks) are read by the first look; the asks the router closed meanwhile are read here.
|
||||
h.wantForge()
|
||||
h.hearAsks()
|
||||
h.Poke()
|
||||
// retries: the controller's walks, until they have been read once; plan-moved events carry every move after
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
err := h.Reconcile()
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
logf("[mesh-delivery] the controller's walks cannot be read yet (%v); asking again in %s", err, wait)
|
||||
time.Sleep(wait)
|
||||
}
|
||||
h.Poke()
|
||||
}
|
||||
|
||||
// handle takes one event, then takes every step the event makes possible (React). An event that cannot be read is
|
||||
// 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, and the controller's walks are read back by comparison anyway.
|
||||
func handle(h *Holder, e stdio.Envelope) error {
|
||||
taken := take(h, e)
|
||||
h.React()
|
||||
h.Poke()
|
||||
return taken
|
||||
}
|
||||
|
||||
@@ -215,7 +231,7 @@ func need(a map[string]any, keys ...string) error {
|
||||
|
||||
func tools(h *Holder, l *listening) []stdio.Tool {
|
||||
seat := "mesh-delivery."
|
||||
return []stdio.Tool{
|
||||
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, " +
|
||||
@@ -391,4 +407,14 @@ func tools(h *Holder, l *listening) []stdio.Tool {
|
||||
return h.MarkReady(strArg(a, "id"), strArg(a, "reviewed-by"), strArg(a, "why"))
|
||||
}},
|
||||
}
|
||||
// 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
|
||||
}
|
||||
|
||||
@@ -101,7 +101,7 @@ func TestClearMarkWithAnEmptyHeadSaysWhatWasWrongOnceItStayedEmpty(t *testing.T)
|
||||
if len(w.forge.comments) != 0 || len(w.forge.unlabels) != 0 {
|
||||
t.Fatalf("an empty-head mark seen once was cleared: %v", w.forge.comments)
|
||||
}
|
||||
w.later(readyTryEvery + time.Second)
|
||||
w.later(markGrace + time.Second)
|
||||
w.h.SeeOpenPulls()
|
||||
w.settleAll()
|
||||
if len(w.forge.comments) != 1 || !strings.Contains(w.forge.comments[0], "no \"ready at\" comment by a listed account") ||
|
||||
@@ -117,12 +117,12 @@ func TestAMarkReadyJustMadeIsNotClearedByAReadThatPredatesItsComment(t *testing.
|
||||
if _, err := w.h.MarkReady(id, "a@b", "ok"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
w.later(readyTryEvery + time.Second) // the label alone was read long enough ago to count as stale...
|
||||
w.later(markGrace + time.Second) // the label alone was read long enough ago to count as stale...
|
||||
w.forge.open = []OpenPull{markedPull("novox/app", 7, fullHead, "")}
|
||||
w.h.SeeOpenPulls()
|
||||
w.h.mu.Lock()
|
||||
w.h.deliveries[id].Ready.Seen = w.now // ...but the mark was made just now
|
||||
w.h.marks["novox/app#7"].firstEmpty = w.now.Add(-2 * readyTryEvery)
|
||||
w.h.marks["novox/app#7"].firstEmpty = w.now.Add(-2 * markGrace)
|
||||
w.h.mu.Unlock()
|
||||
before := len(w.forge.comments)
|
||||
w.settleAll()
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// How this owner keeps time (novox/hq ADR 0406 §1, issue 975). It has no tick, no forge poll and no reconcile loop:
|
||||
// a look (React) is taken after each event that can change what a step needs, and a step that waits on the clock — a
|
||||
// grace, a bound, a retry after a failure — says when it must be looked at again (lookAt). The one timer below is
|
||||
// armed for the earliest of those and fires one more look; it is never a tick, since a look that finds nothing
|
||||
// waiting arms nothing.
|
||||
|
||||
// askAgainAfter is how long after a failed ask of the controller, the forge or the operator it is asked again: the
|
||||
// failure is the reason, and it is said in the log when it happens.
|
||||
const askAgainAfter = 30 * time.Second
|
||||
|
||||
// lookWanted is the earliest time a look must be taken again, and the bounded step that wants it.
|
||||
type lookWanted struct {
|
||||
mu sync.Mutex
|
||||
at time.Time
|
||||
step string
|
||||
working bool
|
||||
timer *time.Timer
|
||||
}
|
||||
|
||||
// lookAt says a bounded step, waiting on the clock, must be looked at again at a time. Safe to call with or without
|
||||
// the holder's lock.
|
||||
func (h *Holder) lookAt(at time.Time, step string) {
|
||||
h.wanted.mu.Lock()
|
||||
defer h.wanted.mu.Unlock()
|
||||
if h.wanted.at.IsZero() || at.Before(h.wanted.at) {
|
||||
h.wanted.at, h.wanted.step = at, step
|
||||
}
|
||||
}
|
||||
|
||||
// again says an ask that failed is asked again after askAgainAfter: a bounded step retried by the look's timer.
|
||||
func (h *Holder) again(step string) { h.lookAt(h.Now().Add(askAgainAfter), step) }
|
||||
|
||||
// Poke asks for a look; any number of pokes before it starts are one look.
|
||||
func (h *Holder) Poke() {
|
||||
select {
|
||||
case h.poke <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// Work takes a look each time it is poked, for ever: the one goroutine that acts on events.
|
||||
func (h *Holder) Work() {
|
||||
h.wanted.mu.Lock()
|
||||
h.wanted.working = true
|
||||
h.wanted.mu.Unlock()
|
||||
for range h.poke {
|
||||
h.React()
|
||||
}
|
||||
}
|
||||
|
||||
// React takes every step the facts allow, after an event that can change what they are: the forge's open pull requests
|
||||
// read when a ready mark may have been made, the steps that need another's word (look), what is owed (Flush), and then
|
||||
// the time the next look is wanted at, if any step waits on the clock. Never concurrent with itself.
|
||||
func (h *Holder) React() {
|
||||
h.reactMu.Lock()
|
||||
defer h.reactMu.Unlock()
|
||||
h.wanted.mu.Lock()
|
||||
h.wanted.at, h.wanted.step = time.Time{}, ""
|
||||
h.wanted.mu.Unlock()
|
||||
if h.takeForgeWanted() {
|
||||
if err := h.SeeOpenPulls(); err != nil {
|
||||
h.wantForge()
|
||||
h.again("the forge's open pull requests, read when a ready mark or a check may have changed")
|
||||
}
|
||||
}
|
||||
h.look()
|
||||
if h.takeReread() {
|
||||
// The one read of the controller's walks besides the one at the start: a merge no walk opened for, at its bound.
|
||||
if err := h.Reconcile(); err != nil {
|
||||
h.Logf("[mesh-delivery] the controller's walks cannot be read: %v", err)
|
||||
h.mu.Lock()
|
||||
h.rereadWalks = true
|
||||
h.mu.Unlock()
|
||||
h.again("the controller's walks, read once at the bound of a merge no walk opened for")
|
||||
} else {
|
||||
h.look()
|
||||
}
|
||||
}
|
||||
h.Flush()
|
||||
h.arm()
|
||||
}
|
||||
|
||||
// arm sets the timer for the earliest time a step wants a look at, or stops it when none does.
|
||||
func (h *Holder) arm() {
|
||||
h.wanted.mu.Lock()
|
||||
defer h.wanted.mu.Unlock()
|
||||
if !h.wanted.working {
|
||||
return
|
||||
}
|
||||
if h.wanted.timer != nil {
|
||||
h.wanted.timer.Stop()
|
||||
h.wanted.timer = nil
|
||||
}
|
||||
if h.wanted.at.IsZero() {
|
||||
return
|
||||
}
|
||||
wait := max(h.wanted.at.Sub(h.Now()), 100*time.Millisecond)
|
||||
// retries: the time-bound step the last look left waiting (a grace, a bound or an ask that failed), at its time
|
||||
h.wanted.timer = time.AfterFunc(wait, h.Poke)
|
||||
}
|
||||
|
||||
// takeReread answers whether the controller's walks are to be read once more, and takes the want.
|
||||
func (h *Holder) takeReread() bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
reread := h.rereadWalks
|
||||
h.rereadWalks = false
|
||||
return reread
|
||||
}
|
||||
@@ -43,9 +43,14 @@ type ReadyMark struct {
|
||||
// ReadyLabel is the label that marks a pull request ready.
|
||||
const ReadyLabel = "mesh/ready"
|
||||
|
||||
// readyTryEvery is how often a marked head's merge is asked while the merge tool refuses it: a check it asked takes
|
||||
// minutes, and a refusal costs the forge a few reads.
|
||||
const readyTryEvery = time.Minute
|
||||
// mergeRetryAfter is how long after the merge tool refused a marked head's merge it is asked again, when nothing
|
||||
// happened meanwhile that decides whether it may merge (a verdict, an Approve, a group's turn are events, and ask it at
|
||||
// once): a check it asked takes minutes, and a refusal costs the forge a few reads. A retry of a bounded step (askMerge).
|
||||
const mergeRetryAfter = time.Minute
|
||||
|
||||
// markGrace is how long a mark not complete (the label, no "ready at" comment yet), and one this owner has just
|
||||
// cleared, are left before they are judged again: the comment follows the label, and a clearing takes a moment.
|
||||
const markGrace = time.Minute
|
||||
|
||||
// readyRefusedFor is how long the merge tool may refuse a marked head before it is said as stalled: longer than a
|
||||
// check takes (the controller's checkTimeout is 45 minutes), so a check on its way is never said.
|
||||
@@ -114,10 +119,10 @@ func (r *ReadyRun) mergedBy() string {
|
||||
// marked is a pull request read with a ready mark.
|
||||
type marked struct {
|
||||
pull OpenPull
|
||||
// cleared is when the mark was cleared, the label taken off: not cleared again within readyTryEvery.
|
||||
// cleared is when the mark was cleared, the label taken off: not cleared again within markGrace.
|
||||
cleared time.Time
|
||||
// firstEmpty is when a mark with no head (a label no "ready at" comment follows yet) was first read so: it is a
|
||||
// mark not yet complete, and is cleared only once it has stayed so for readyTryEvery (novox/hq issue 873).
|
||||
// mark not yet complete, and is cleared only once it has stayed so for markGrace (novox/hq issue 873).
|
||||
firstEmpty time.Time
|
||||
}
|
||||
|
||||
@@ -168,14 +173,17 @@ func (h *Holder) readyAsks(now time.Time) []func() {
|
||||
p, mark := m.pull, *m.pull.Ready
|
||||
repository := strings.ToLower(p.Owner + "/" + p.Repo)
|
||||
if !sameFull(mark.Head, p.HeadSHA) {
|
||||
if mark.Head == "" && now.Sub(m.firstEmpty) < readyTryEvery {
|
||||
if mark.Head == "" && now.Sub(m.firstEmpty) < markGrace {
|
||||
h.lookAt(m.firstEmpty.Add(markGrace), "a ready label whose comment naming the head has not followed yet")
|
||||
continue // the label first, its comment not yet: a mark not complete, not an old one
|
||||
}
|
||||
if d := h.deliveries[IDOf(repository, p.HeadSHA)]; d != nil && d.Ready != nil &&
|
||||
sameFull(d.Ready.Head, p.HeadSHA) && now.Sub(d.Ready.Seen) < readyTryEvery {
|
||||
sameFull(d.Ready.Head, p.HeadSHA) && now.Sub(d.Ready.Seen) < markGrace {
|
||||
h.lookAt(d.Ready.Seen.Add(markGrace), "a mark made through mark-ready, its comment possibly not yet read")
|
||||
continue // marked through mark-ready a moment ago: this read may predate its comment
|
||||
}
|
||||
if !m.cleared.IsZero() && now.Sub(m.cleared) < readyTryEvery {
|
||||
if !m.cleared.IsZero() && now.Sub(m.cleared) < markGrace {
|
||||
h.lookAt(m.cleared.Add(markGrace), "a stale ready mark whose clearing may not have taken")
|
||||
continue
|
||||
}
|
||||
m.cleared = now
|
||||
@@ -201,10 +209,11 @@ func (h *Holder) readyAsks(now time.Time) []func() {
|
||||
}
|
||||
if r.Declined != nil {
|
||||
// The operator declined: nothing is merged, and the mark is taken off (said once when it was declined).
|
||||
if m.cleared.IsZero() || now.Sub(m.cleared) >= readyTryEvery {
|
||||
if m.cleared.IsZero() || now.Sub(m.cleared) >= markGrace {
|
||||
m.cleared = now
|
||||
asks = append(asks, h.unmark(k, p, mark))
|
||||
}
|
||||
h.lookAt(m.cleared.Add(markGrace), "a declined ready mark whose clearing may not have taken")
|
||||
continue
|
||||
}
|
||||
if d.State == Rejected {
|
||||
@@ -257,7 +266,11 @@ func (h *Holder) readyAsks(now time.Time) []func() {
|
||||
if approvalAsk != nil {
|
||||
asks = append(asks, approvalAsk)
|
||||
}
|
||||
if waits != "" || r.merging || (!r.LastTry.IsZero() && now.Sub(r.LastTry) < readyTryEvery) {
|
||||
if waits != "" || r.merging {
|
||||
continue
|
||||
}
|
||||
if !r.LastTry.IsZero() && now.Sub(r.LastTry) < mergeRetryAfter {
|
||||
h.lookAt(r.LastTry.Add(mergeRetryAfter), "a merge the forge refused")
|
||||
continue
|
||||
}
|
||||
r.Tries++
|
||||
@@ -349,8 +362,9 @@ func (h *Holder) askMerge(id string, p OpenPull, head string) func() {
|
||||
}
|
||||
if answer != r.Answer {
|
||||
h.Logf("[mesh-delivery] %s is marked ready, and the merge tool did not merge it (asked again in %s): %s", id,
|
||||
readyTryEvery, answer)
|
||||
mergeRetryAfter, answer)
|
||||
}
|
||||
h.lookAt(now.Add(mergeRetryAfter), "a merge the forge refused")
|
||||
r.Answer = answer
|
||||
h.keep(d)
|
||||
}
|
||||
@@ -366,6 +380,7 @@ func (h *Holder) clearMark(k string, p OpenPull, mark ReadyMark) func() {
|
||||
if mark.Head == "" {
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] %s's ready label has no \"ready at <head>\" comment of a listed account after it, and could not be taken off yet: %v", k, err)
|
||||
h.again("a ready label that could not be taken off")
|
||||
return
|
||||
}
|
||||
h.Logf("[mesh-delivery] %s's ready label taken off: no \"ready at <head>\" comment of a listed account was written after it; mark-ready writes both in order", k)
|
||||
@@ -373,6 +388,7 @@ func (h *Holder) clearMark(k string, p OpenPull, mark ReadyMark) func() {
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] %s's ready mark at %s names an older head than %s, and could not be cleared yet: %v",
|
||||
k, shortOf(mark.Head, 12), shortOf(p.HeadSHA, 12), err)
|
||||
h.again("a stale ready mark that could not be cleared")
|
||||
return
|
||||
}
|
||||
h.Logf("[mesh-delivery] %s's ready mark at %s cleared: its head is now %s", k, shortOf(mark.Head, 12),
|
||||
@@ -389,6 +405,7 @@ func (h *Holder) unmark(k string, p OpenPull, mark ReadyMark) func() {
|
||||
return func() {
|
||||
if err := h.Forge.Unlabel(p.Owner, p.Repo, p.Number, mark.LabelID); err != nil {
|
||||
h.Logf("[mesh-delivery] %s's ready mark could not be taken off after the operator's Decline yet: %v", k, err)
|
||||
h.again("a declined ready mark that could not be taken off")
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
@@ -442,7 +459,7 @@ func (h *Holder) readyLines(now time.Time) []StalledLine {
|
||||
case r.RefusedSince != nil && now.Sub(*r.RefusedSince) >= readyRefusedFor:
|
||||
out = append(out, StalledLine{ID: d.ID, Number: d.Number, State: d.State,
|
||||
For: now.Sub(*r.RefusedSince).Round(time.Second).String(), Bound: readyRefusedFor.String(),
|
||||
H2: "none: the merge is asked again each minute — the operator's",
|
||||
H2: "none: the merge is asked again after each verdict, Approve or turn, and each minute — the operator's",
|
||||
Says: fmt.Sprintf("%s#%d is marked ready at %s, and the merge tool has refused it since %s: %s", d.Repository,
|
||||
d.Number, shortOf(r.Head, 12), r.RefusedSince.UTC().Format(time.RFC3339), r.Answer)})
|
||||
}
|
||||
|
||||
@@ -232,7 +232,7 @@ var Table = []Row{
|
||||
case d.Check != nil:
|
||||
return errors.New("its verdict is known")
|
||||
case d.CheckAsk == nil:
|
||||
return errors.New("its check was never asked: the tick asks it, not H2")
|
||||
return errors.New("its check was never asked: a look asks it, not H2")
|
||||
case d.CheckAsk.Reasked > 0:
|
||||
return errors.New("its check was re-asked once already: the operator's")
|
||||
}
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// **A step that waits on the clock says when it is looked at again** (novox/hq ADR 0406 §1, issue 975): there is no
|
||||
// tick to find it, so each grace, bound and retry is registered (lookAt) when a look meets it, and the one timer fires
|
||||
// a look for the earliest. A look that meets nothing waiting registers nothing.
|
||||
|
||||
func (w *world) wantedAt() (time.Time, string) {
|
||||
w.h.wanted.mu.Lock()
|
||||
defer w.h.wanted.mu.Unlock()
|
||||
return w.h.wanted.at, w.h.wanted.step
|
||||
}
|
||||
|
||||
func TestALookThatMeetsNothingWaitingOnTheClockWantsNoOtherLook(t *testing.T) {
|
||||
w := newWorld(t)
|
||||
w.h.PullUpdated(pr("novox/app", 7, readyHead, "feat/x", "src/a.go"))
|
||||
w.h.Checked(judgedAt("novox/app", 7, readyHead, "pass", "t1t1t1t1t1t1"))
|
||||
w.h.React()
|
||||
if at, step := w.wantedAt(); !at.IsZero() {
|
||||
t.Fatalf("a look with nothing waiting on the clock wants another at %s for %q", at, step)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAProposedDeliveryWithNoVerdictHasItsCheckAskedWhenItsGraceEndsWithNoTick(t *testing.T) {
|
||||
w := newWorld(t)
|
||||
w.ctl.down = false
|
||||
w.h.PullUpdated(pr("novox/app", 7, readyHead, "feat/x", "src/a.go"))
|
||||
id := IDOf("novox/app", readyHead)
|
||||
w.h.React()
|
||||
since := w.delivery(id).Since
|
||||
if at, step := w.wantedAt(); !at.Equal(since.Add(checkGrace)) || !strings.Contains(step, "check") {
|
||||
t.Fatalf("the look wanted at %s for %q; wanted when the grace ends, %s", at, step, since.Add(checkGrace))
|
||||
}
|
||||
if w.delivery(id).CheckAsk != nil {
|
||||
t.Fatal("the check was asked before the grace ended")
|
||||
}
|
||||
w.later(checkGrace + time.Second)
|
||||
w.h.React()
|
||||
if w.delivery(id).CheckAsk == nil {
|
||||
t.Fatal("the look the grace's end wants did not ask the check")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMergeTheForgeRefusedIsAskedAgainAfterAMinuteByTheTimerAlone(t *testing.T) {
|
||||
w := newWorld(t)
|
||||
w.h.PullUpdated(pr("novox/app", 7, readyHead, "feat/x", "src/a.go"))
|
||||
w.h.Checked(judgedAt("novox/app", 7, readyHead, "pass", "t1t1t1t1t1t1"))
|
||||
w.forge.refuse = errors.New("the forge refused gitea_merge_pull_request: not mergeable yet")
|
||||
w.forge.open = []OpenPull{markedPull("novox/app", 7, readyHead, readyHead)}
|
||||
w.h.wantForge()
|
||||
w.settleAll()
|
||||
if len(w.forge.merges) != 1 {
|
||||
t.Fatalf("merges %v", w.forge.merges)
|
||||
}
|
||||
at, step := w.wantedAt()
|
||||
if want := w.now.Add(mergeRetryAfter); !at.Equal(want) || !strings.Contains(step, "merge") {
|
||||
t.Fatalf("the look wanted at %s for %q; wanted %s for the refused merge", at, step, want)
|
||||
}
|
||||
w.forge.refuse = nil
|
||||
w.later(mergeRetryAfter + time.Second)
|
||||
w.settleAll()
|
||||
if len(w.forge.merges) != 2 || w.delivery(IDOf("novox/app", readyHead)).Ready.Merged == nil {
|
||||
t.Fatalf("the refused merge was not asked again at the timer: %v", w.forge.merges)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMergeNoWalkOpenedForReadsTheControllersWalksOnceAtItsBound(t *testing.T) {
|
||||
w := newWorldOver(t, newMemStore(), newFakeController(), newFakeForge(), atCC)
|
||||
cc := pr(cat, 101, headCC, "feat/claude-code", "modules/claude-code/module.json")
|
||||
w.h.PullUpdated(cc)
|
||||
w.h.Checked(verdictMoving(cat, 101, headCC, "claude-code"))
|
||||
w.h.PullMerged(mergedAt(cc, mergeCC, atCC))
|
||||
w.h.WalkMoved(w.ctl.put(catWalk("plan-5", "eeeeeeeeeeee5555", atCC.Add(time.Minute), "dunst")))
|
||||
w.h.React()
|
||||
if at, step := w.wantedAt(); at.IsZero() || at.After(atCC.Add(unwalked)) {
|
||||
t.Fatalf("the look wanted at %s for %q; wanted by the bound, %s", at, step, atCC.Add(unwalked))
|
||||
}
|
||||
reads := w.ctl.walkReads
|
||||
w.later(unwalked + time.Second)
|
||||
w.settleAll()
|
||||
w.settleAll()
|
||||
id := IDOf(cat, headCC)
|
||||
if d := w.delivery(id); d.State != Held || !strings.Contains(d.HeldWhy, "plan-5") {
|
||||
t.Fatalf("it is %s: %q", d.State, d.HeldWhy)
|
||||
}
|
||||
if got := w.ctl.walkReads - reads; got != 1 {
|
||||
t.Fatalf("the controller's walks were read %d times at the bound; wanted once", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnEffectTheForgeCouldNotTakeIsTriedAgainAtItsTimeByTheTimerAlone(t *testing.T) {
|
||||
w := newWorld(t)
|
||||
w.forge.down = true
|
||||
w.h.PullUpdated(pr("novox/app", 7, readyHead, "feat/x", "src/a.go"))
|
||||
w.h.React()
|
||||
at, step := w.wantedAt()
|
||||
if at.IsZero() || !strings.Contains(step, "owed") || at.After(w.now.Add(retryFirst)) {
|
||||
t.Fatalf("the look wanted at %s for %q; wanted within %s for the owed effect", at, step, retryFirst)
|
||||
}
|
||||
w.forge.down = false
|
||||
w.later(retryFirst + time.Second)
|
||||
w.h.React()
|
||||
if len(w.forge.notes) == 0 {
|
||||
t.Fatal("the owed effect was not tried again at its time")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnAskTheControllerCouldNotAnswerIsAskedAgainAfterAWhileByTheTimerAlone(t *testing.T) {
|
||||
w := newWorld(t)
|
||||
w.h.PullUpdated(pr("novox/app", 7, readyHead, "feat/x", "src/a.go"))
|
||||
w.h.Checked(judgedAt("novox/app", 7, readyHead, "pass", "t1t1t1t1t1t1"))
|
||||
w.settleAll()
|
||||
w.ctl.down = true
|
||||
w.h.mu.Lock()
|
||||
d := w.h.deliveries[IDOf("novox/app", readyHead)]
|
||||
d.Plan = nil
|
||||
w.h.mu.Unlock()
|
||||
w.h.React()
|
||||
at, step := w.wantedAt()
|
||||
if want := w.now.Add(askAgainAfter); !at.Equal(want) || !strings.Contains(step, "plan") {
|
||||
t.Fatalf("the look wanted at %s for %q; wanted %s for the plan that could not be asked", at, step, want)
|
||||
}
|
||||
}
|
||||
@@ -167,11 +167,11 @@ func askedAgain(was, now map[string]string) bool {
|
||||
|
||||
// SeeOpenPulls asks the forge for every open pull request and keeps the heads with a required merge check not
|
||||
// answered. A forge that cannot be asked changes nothing: what was seen stands, and nothing is guessed.
|
||||
func (h *Holder) SeeOpenPulls() {
|
||||
func (h *Holder) SeeOpenPulls() error {
|
||||
open, err := h.Forge.OpenPulls()
|
||||
if err != nil {
|
||||
h.Logf("[mesh-delivery] the forge's open pull requests cannot be read: %v", err)
|
||||
return
|
||||
return err
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
@@ -205,6 +205,7 @@ func (h *Holder) SeeOpenPulls() {
|
||||
}
|
||||
h.unchecked = seen
|
||||
h.seeMarks(open)
|
||||
return nil
|
||||
}
|
||||
|
||||
// started is whether the head's check started, rather than was never asked: it has any merge check, or a delivery
|
||||
|
||||
@@ -385,7 +385,7 @@ func (h *Holder) MarkReady(id, reviewedBy, why string) (string, error) {
|
||||
h.keep(d)
|
||||
waits := h.readyWaits(d)
|
||||
if waits == "" {
|
||||
return fmt.Sprintf("%s is marked ready at %s; its merge is asked once the next tick reads the mark", id, shortOf(head, 12)), nil
|
||||
return fmt.Sprintf("%s is marked ready at %s; its merge is asked once the forge's label and comment events reach this owner", id, shortOf(head, 12)), nil
|
||||
}
|
||||
return fmt.Sprintf("%s is marked ready at %s; its merge waits: %s", id, shortOf(head, 12), waits), nil
|
||||
}
|
||||
@@ -450,8 +450,9 @@ func (h *Holder) Release(id, why, by string) (string, error) {
|
||||
"delivery was stopped, and is asked to leave it out first", id, by, walk, strings.Join(stopped, ", "))
|
||||
}
|
||||
if err := h.Controller.Deliver(walk, "released by "+by+": "+why); err != nil {
|
||||
return "", fmt.Errorf("%s is delivering on %s's word, and its walk %s could not be let go yet (%v): the next "+
|
||||
"tick asks again", id, by, walk, err)
|
||||
h.again("a walk's word the controller refused or did not answer")
|
||||
return "", fmt.Errorf("%s is delivering on %s's word, and its walk %s could not be let go yet (%v): it is "+
|
||||
"asked again in %s", id, by, walk, err, askAgainAfter)
|
||||
}
|
||||
h.mu.Lock()
|
||||
if d := h.deliveries[id]; d != nil {
|
||||
@@ -568,7 +569,7 @@ func (h *Holder) StopWay(id, why, by, way string) (string, error) {
|
||||
a()
|
||||
}
|
||||
if batch != "" {
|
||||
// Asked now, not at the next tick; and the batch's walk is never let go while it carries the merge.
|
||||
// Asked now, not at the next look; and the batch's walk is never let go while it carries the merge.
|
||||
did = fmt.Sprintf("; its merge waits in the batch %s, which was asked now to leave it out and is never let go "+
|
||||
"while it carries it: %s", batch, h.stopLeaving(batch, why, by, []string{merge}))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user