mesh-catalog/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/lostcheck.go

package main

import (
	"fmt"
	"strings"
	"time"
)

// A check the build queue lost (novox/hq issue 1274).
//
// A proposed delivery waits for a check it asked of the controller. When the build queue loses that ask — taken by a
// build agent that was replaced, or dropped with the queue's own state — nothing answers and nothing asks again: the
// delivery stays proposed until a new head is pushed, and a recheck from proposed was refused. The sweep now reads the
// controller's `queue` and asks again a check that no entry holds (neither waiting nor in flight; a dead one is not
// held) for longer than lostCheckBound. A recheck of a proposed delivery is taken on the same fact. Both are the
// table's EvLost, said in the delivery's transitions; neither asks while the queue holds an ask of the head, so no
// ask is made twice. A queue that cannot be read asks nothing: a guess would risk a duplicate.

// lostReadEvery is how often the queue is read for one proposed delivery's ask: the read is one bus call, and a check
// the queue holds is looked at again no sooner.
const lostReadEvery = 2 * time.Minute

// lostCheckBound is how long a proposed delivery's asked check may go with no entry in the build queue before it is
// asked again: the repository's last measured check time (ADR 0505 §4, lastCheckTime), at least UncheckedAfter (ten
// minutes, the bound a check never asked is said at). Under the lock.
func (h *Holder) lostCheckBound(repository string) time.Duration {
	return max(UncheckedAfter, h.lastCheckTime(repository))
}

// awaitedSince is when a proposed delivery's check was asked, when it waits for one this owner asked: no verdict yet,
// or a verdict out of date with its check against the trunk's head asked. False for a delivery waiting for none.
func awaitedSince(d *Delivery) (time.Time, bool) {
	if d.State != Proposed || d.MergedAs != "" || d.Number == 0 {
		return time.Time{}, false
	}
	switch {
	case d.Check == nil && askedCheck(d):
		return d.CheckAsk.At, true
	case d.Check != nil && d.outOfDate() != "" && d.AskedAgain != nil:
		return *d.AskedAgain, true
	}
	return time.Time{}, false
}

// heldBy is the queue's entry that holds a check of the delivery's head: its own ask by id, or any check of its
// repository at its head — a group's composed check included, which a second ask would only repeat.
func heldBy(d *Delivery, asks []QueuedAsk) *QueuedAsk {
	for i, a := range asks {
		if !a.Holds() {
			continue
		}
		if d.CheckAsk != nil && d.CheckAsk.ID != "" && a.ID == d.CheckAsk.ID {
			return &asks[i]
		}
		if a.Kind == "check" && a.Commit != "" && sameCommit(a.Commit, d.Commit) &&
			strings.Contains(strings.ToLower(a.Repository), strings.ToLower(d.Repository)) {
			return &asks[i]
		}
	}
	return nil
}

func (a QueuedAsk) words() string {
	// The controller's `queue` says the state "in flight", its table's word.
	state := strings.ReplaceAll(a.State, "-", " ")
	if state == "in flight" && a.Machine != "" {
		return a.ID + " is in flight on " + a.Machine
	}
	return a.ID + " is " + state + " in the build queue"
}

// lostChecks are the asks again of proposed deliveries whose asked check the build queue holds no entry of, past the
// bound, at each look. Under the lock; the queue is read, and the asks made, outside it.
func (h *Holder) lostChecks(now time.Time) func() {
	var ids []string
	for id, d := range h.deliveries {
		since, ok := awaitedSince(d)
		if !ok {
			delete(h.lostRead, id)
			continue
		}
		due := since.Add(h.lostCheckBound(d.Repository))
		if now.Before(due) {
			h.lookAt(due, "a proposed delivery's asked check, read in the build queue once its bound passed")
			continue
		}
		if read, was := h.lostRead[id]; was && now.Sub(read) < lostReadEvery {
			h.lookAt(read.Add(lostReadEvery), "a proposed delivery's asked check, looked for in the build queue again")
			continue
		}
		h.lostRead[id] = now
		h.lookAt(now.Add(lostReadEvery), "a proposed delivery's asked check, looked for in the build queue again")
		ids = append(ids, id)
	}
	if len(ids) == 0 {
		return nil
	}
	return func() {
		asks, err := h.Controller.Queue()
		if err != nil {
			h.Logf("[mesh-delivery] the build queue cannot be read, so no check is asked again for a lost ask: %v", err)
			return
		}
		for _, id := range ids {
			h.mu.Lock()
			d := h.deliveries[id]
			since, ok := time.Time{}, false
			if d != nil {
				since, ok = awaitedSince(d)
			}
			var held *QueuedAsk
			if ok {
				held = heldBy(d, asks)
			}
			var why string
			if ok && held == nil {
				what := "its check against the trunk's head"
				if d.Check == nil {
					what = "its check " + d.CheckAsk.ID
				}
				why = fmt.Sprintf("%s, asked %s ago, is held by no entry of the build queue (neither waiting nor in flight): "+
					"lost, asked again", what, h.Now().Sub(since).Round(time.Second))
			}
			h.mu.Unlock()
			if why == "" {
				continue
			}
			if _, err := h.askLost(id, byOwner, why, asks); err != nil {
				h.Logf("[mesh-delivery] %s: its lost check could not be asked again: %v", id, err)
				h.again("a proposed delivery's lost check that could not be asked again")
			}
		}
	}
}

// recheckProposed is `recheck` of a proposed delivery (issue 1274): taken, with why, when no check of its head is
// waiting or in flight in the build queue, and refused naming the one that is. A queue that cannot be read refuses it.
func (h *Holder) recheckProposed(id, why, by string) (string, error) {
	if why == "" {
		return "", fmt.Errorf("a recheck says why")
	}
	asks, err := h.Controller.Queue()
	if err != nil {
		return "", fmt.Errorf("%s is not rechecked: the build queue cannot be read (%v), so whether its check is still held is not known", id, err)
	}
	h.mu.Lock()
	d := h.deliveries[id]
	var held *QueuedAsk
	if d != nil {
		held = heldBy(d, asks)
	}
	h.mu.Unlock()
	if held != nil {
		return "", fmt.Errorf("%s is not rechecked: its check is still held, %s", id, held.words())
	}
	return h.askLost(id, by, why, asks)
}

// askLost asks a proposed delivery's check again, the table's EvLost: tried first on a copy, so an ask the controller
// does not take spends nothing. Its group's composed check goes out of date, and is asked once the member answers.
func (h *Holder) askLost(id, by, why string, asks []QueuedAsk) (string, error) {
	h.mu.Lock()
	d := h.deliveries[id]
	if d == nil {
		h.mu.Unlock()
		return "", fmt.Errorf("no delivery %q", id)
	}
	if held := heldBy(d, asks); held != nil {
		h.mu.Unlock()
		return "", fmt.Errorf("%s is not asked again: its check is still held, %s", id, held.words())
	}
	probe := *d
	probe.Transitions = nil
	if _, err := Apply(&probe, EvLost, true, Facts{Now: h.Now(), Why: why, By: by}); err != nil {
		h.mu.Unlock()
		return "", err
	}
	m := MemberOf(d)
	if d.Check != nil && d.outOfDate() != "" && d.TrunkNow != "" {
		m.BaseSHA = d.TrunkNow
	}
	h.mu.Unlock()
	asked, err := h.Controller.Check("", []Member{m})
	if err != nil {
		return "", fmt.Errorf("%s's check could not be asked again yet: %w", id, err)
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	d = h.deliveries[id]
	if d == nil || d.State != Proposed {
		return "", fmt.Errorf("%s moved while its check was asked again", id)
	}
	now := h.Now()
	t, err := Apply(d, EvLost, true, Facts{Now: now, Why: why + " — asked as " + asked, By: by})
	if err != nil {
		return "", err
	}
	d.CheckAsk = &CheckAsk{ID: asked, At: now, By: by, Why: why}
	if d.Check != nil {
		d.AskedAgain, d.FreshWanted = &now, false
	}
	delete(h.askTried, id)
	if g := h.groups[d.Group]; g != nil && g.Check != nil {
		g.Check.OutOfDate = fmt.Sprintf("composed check out of date: %s's check was lost in the build queue and asked again", d.ID)
		g.Check.OutOfDateAt = now
		g.Updated = now
		h.keepGroup(g)
	}
	h.moved(d, []Transition{t})
	return fmt.Sprintf("%s: its check was held by no entry of the build queue and is asked again as %s (%s)", id, asked, why), nil
}