|
|
|
@@ -85,6 +85,10 @@ type Checker struct {
|
|
|
|
|
Announce Announcer
|
|
|
|
|
Now func() time.Time
|
|
|
|
|
Logf func(format string, args ...any)
|
|
|
|
|
// Settle is how long a group's heads must stay the same after the candidate last changed them before its check starts
|
|
|
|
|
// (issue 1890): the members of a group are announced one after another, and a check of each partial set is thrown
|
|
|
|
|
// away at the next. None when 0; DefaultSettle in the module.
|
|
|
|
|
Settle time.Duration
|
|
|
|
|
|
|
|
|
|
once sync.Once
|
|
|
|
|
wg sync.WaitGroup
|
|
|
|
@@ -96,15 +100,77 @@ type Checker struct {
|
|
|
|
|
epoch uint64 // changes of the nodes seen since this process started
|
|
|
|
|
nodesFail map[string]uint64 // by verdict key: the epoch a failure of the nodes was written at
|
|
|
|
|
held map[string]string // by verdict key: the turn that holds its check off, as last said
|
|
|
|
|
live map[string]liveCheck // by verdict key: the check running in this process, ended when its heads go
|
|
|
|
|
settling map[string]*window // by candidates key: the window its heads settle in
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// liveCheck is a check running in this process: the candidates key of its group and the end of its run.
|
|
|
|
|
type liveCheck struct {
|
|
|
|
|
group string
|
|
|
|
|
end context.CancelCauseFunc
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// window is a group's settle window: aborted with errUnsettled when its heads change again or the group goes, and with
|
|
|
|
|
// errSettled to end it at once.
|
|
|
|
|
type window struct{ abort context.CancelCauseFunc }
|
|
|
|
|
|
|
|
|
|
var (
|
|
|
|
|
errUnsettled = errors.New("the group's heads changed again")
|
|
|
|
|
errSettled = errors.New("the group's heads settled")
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
// settleNow ends the settle window of every group at once, as if Settle had passed.
|
|
|
|
|
func (ch *Checker) settleNow() {
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
defer ch.mu.Unlock()
|
|
|
|
|
for _, w := range ch.settling {
|
|
|
|
|
w.abort(errSettled)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// DefaultSettle is the module's Settle (issue 1890). On 2026-10-26 the three members of a group were announced within 2.8
|
|
|
|
|
// seconds of each other (20:31:52.3, :53.7, :55.1 UTC), a session opening or pushing them one API call after another;
|
|
|
|
|
// fifteen seconds holds five times that, and costs a check of 10 to 18 minutes under 2.5 % of its time. A member that comes
|
|
|
|
|
// later still is no waste beyond its asks already taken: the check of the heads it replaced is ended at once.
|
|
|
|
|
const DefaultSettle = 15 * time.Second
|
|
|
|
|
|
|
|
|
|
// EndedWords open the words of a verdict the checker ended without judging it, its heads no candidate's any more (issue
|
|
|
|
|
// 1890): never a failure of the change, and judged whole if a candidate holds those heads again.
|
|
|
|
|
const EndedWords = "ended: "
|
|
|
|
|
|
|
|
|
|
// IsEnded says whether a verdict was ended without being judged (EndedWords).
|
|
|
|
|
func IsEnded(v delivery.Verdict) bool {
|
|
|
|
|
return v.State == delivery.VerdictFail && v.Failed.Machine == "checker" && strings.HasPrefix(v.Failed.Words, EndedWords)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// superseded is the cause a running check is ended with: the heads that replaced its own, or none when its group went.
|
|
|
|
|
type superseded struct{ by string }
|
|
|
|
|
|
|
|
|
|
func (s superseded) Error() string {
|
|
|
|
|
if s.by == "" {
|
|
|
|
|
return EndedWords + "no candidate holds these heads any more: the group was closed or merged"
|
|
|
|
|
}
|
|
|
|
|
return EndedWords + "superseded by " + s.by
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (ch *Checker) init() {
|
|
|
|
|
ch.once.Do(func() {
|
|
|
|
|
ch.known, ch.inflight, ch.again, ch.tried = map[string]delivery.Candidate{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
|
|
|
|
|
ch.nodesFail, ch.held = map[string]uint64{}, map[string]string{}
|
|
|
|
|
ch.live, ch.settling = map[string]liveCheck{}, map[string]*window{}
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// endOthers ends every check of the group running in this process whose verdict key is not keep: no candidate holds its
|
|
|
|
|
// heads any more (issue 1890). Called with ch.mu held.
|
|
|
|
|
func (ch *Checker) endOthers(group, keep, by string) {
|
|
|
|
|
for k, l := range ch.live {
|
|
|
|
|
if l.group == group && k != keep {
|
|
|
|
|
l.end(superseded{by: by})
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (ch *Checker) logf(format string, args ...any) {
|
|
|
|
|
if ch.Logf != nil {
|
|
|
|
|
ch.Logf(format, args...)
|
|
|
|
@@ -137,17 +203,62 @@ func (ch *Checker) Candidate(ctx context.Context, key string, value []byte) {
|
|
|
|
|
ch.logf("candidates.%s holds what is not a candidate, left: %v", key, err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
vkey, _ := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
prev, had := ch.known[key]
|
|
|
|
|
ch.known[key] = c
|
|
|
|
|
// A check of heads this candidate no longer holds is ended at once (issue 1890): a member's new head, a member joining
|
|
|
|
|
// or leaving.
|
|
|
|
|
ch.endOthers(key, vkey, GateHeads(c))
|
|
|
|
|
moved := !had || prev.HeadsHash != c.HeadsHash
|
|
|
|
|
if ch.Settle > 0 && (moved || ch.settling[key] != nil) {
|
|
|
|
|
// The group's heads changed: its check waits until they have been the same for Settle, so a group whose members
|
|
|
|
|
// are announced one after another is checked once, whole. A later change starts the window again. The wait is a
|
|
|
|
|
// context's bound, no timer (ADR 0437 §8).
|
|
|
|
|
if w := ch.settling[key]; w != nil {
|
|
|
|
|
w.abort(errUnsettled)
|
|
|
|
|
}
|
|
|
|
|
base, abort := context.WithCancelCause(ctx)
|
|
|
|
|
w := &window{abort: abort}
|
|
|
|
|
ch.settling[key] = w
|
|
|
|
|
ch.wg.Add(1)
|
|
|
|
|
go func() {
|
|
|
|
|
defer ch.wg.Done()
|
|
|
|
|
wait, done := context.WithTimeout(base, ch.Settle)
|
|
|
|
|
defer done()
|
|
|
|
|
<-wait.Done()
|
|
|
|
|
settled := errors.Is(wait.Err(), context.DeadlineExceeded) || errors.Is(context.Cause(wait), errSettled)
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
if !settled || ch.settling[key] != w {
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
return // changed again, forgotten, or the checker stops
|
|
|
|
|
}
|
|
|
|
|
delete(ch.settling, key)
|
|
|
|
|
now, ok := ch.known[key]
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
abort(nil)
|
|
|
|
|
if ok {
|
|
|
|
|
ch.consider(ctx, key, now)
|
|
|
|
|
}
|
|
|
|
|
}()
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
ch.consider(ctx, key, c)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Forget takes a candidate out (its key was deleted).
|
|
|
|
|
// Forget takes a candidate out (its key was deleted): a check of its heads still running is ended, as no candidate holds
|
|
|
|
|
// them any more.
|
|
|
|
|
func (ch *Checker) Forget(key string) {
|
|
|
|
|
ch.init()
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
delete(ch.known, key)
|
|
|
|
|
if w := ch.settling[key]; w != nil {
|
|
|
|
|
w.abort(errUnsettled)
|
|
|
|
|
delete(ch.settling, key)
|
|
|
|
|
}
|
|
|
|
|
ch.endOthers(key, "", "")
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
@@ -190,32 +301,49 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
|
|
|
|
|
}
|
|
|
|
|
cs.Queued = ch.now()
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
if ch.settling[key] != nil {
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
return // its heads are settling: the window's end considers it (Settle)
|
|
|
|
|
}
|
|
|
|
|
if ch.inflight[cs.Key] {
|
|
|
|
|
ch.again[cs.Key] = true // looked at once more when the check ends, so a change of the trunk is never lost
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
ch.inflight[cs.Key] = true
|
|
|
|
|
run, end := context.WithCancelCause(ctx)
|
|
|
|
|
ch.live[cs.Key] = liveCheck{group: key, end: end}
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
ch.wg.Add(1)
|
|
|
|
|
go func() {
|
|
|
|
|
defer ch.wg.Done()
|
|
|
|
|
vkey := cs.Key
|
|
|
|
|
defer func() {
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
delete(ch.inflight, vkey)
|
|
|
|
|
delete(ch.again, vkey)
|
|
|
|
|
delete(ch.live, vkey)
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
end(nil)
|
|
|
|
|
}()
|
|
|
|
|
for {
|
|
|
|
|
// Every check is asked at once: the build seats' streams hold the backlog, a new head's check and a check after a
|
|
|
|
|
// failure of the nodes on check, a re-judge after a trunk move on recheck, pulled after them (ADR 0525 §4, §9;
|
|
|
|
|
// ADR 0536 §4). The checker holds no place and no capacity of its own.
|
|
|
|
|
ch.prepare(ctx, cs)
|
|
|
|
|
ch.try(ctx, key, cs)
|
|
|
|
|
ch.try(ctx, run, key, cs)
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
more := ch.again[cs.Key]
|
|
|
|
|
delete(ch.again, cs.Key)
|
|
|
|
|
if !more || ctx.Err() != nil {
|
|
|
|
|
delete(ch.inflight, cs.Key)
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
more := ch.again[vkey]
|
|
|
|
|
delete(ch.again, vkey)
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
if !more || run.Err() != nil {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
ch.mu.Unlock()
|
|
|
|
|
// Looked at again for the same heads only: a candidate that moved to other heads has its own check (issue 1890).
|
|
|
|
|
if next := CaseOf(ch.candidate(key, c), ch.heads()); next != nil {
|
|
|
|
|
if next.Key != vkey {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
cs = next
|
|
|
|
|
}
|
|
|
|
|
cs.Queued = ch.now()
|
|
|
|
@@ -252,6 +380,8 @@ func (ch *Checker) prepare(ctx context.Context, cs *Case) {
|
|
|
|
|
return
|
|
|
|
|
case IsConflictFailure(v):
|
|
|
|
|
return // a head git refused: its next check is whole (ADR 0548 §7)
|
|
|
|
|
case IsEnded(v):
|
|
|
|
|
return // ended unjudged when its heads went (issue 1890): held again, they are checked whole
|
|
|
|
|
}
|
|
|
|
|
cs.Kind = KindTrunk
|
|
|
|
|
if v.State == delivery.VerdictPass && ch.stale(v, cs) {
|
|
|
|
@@ -402,8 +532,8 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
|
|
|
|
|
return &claimed{rev}, nil
|
|
|
|
|
}
|
|
|
|
|
return nil, nil
|
|
|
|
|
case ch.stale(cur, cs):
|
|
|
|
|
retake = true
|
|
|
|
|
case IsEnded(cur), ch.stale(cur, cs):
|
|
|
|
|
retake = true // heads held again after their check was ended unjudged (issue 1890), or a trunk that moved
|
|
|
|
|
case cur.State == delivery.VerdictFail && IsNodesFailure(cur):
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
at, ok := ch.nodesFail[cs.Key]
|
|
|
|
@@ -483,9 +613,13 @@ func IsNodesFailure(v delivery.Verdict) bool {
|
|
|
|
|
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, UnheardWords)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends
|
|
|
|
|
// because ctx did is left running: the checker takes it again when it starts.
|
|
|
|
|
func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
|
|
|
|
|
// try claims the check, judges it within its bound under run and writes the verdict and the owed statuses. A check that ends
|
|
|
|
|
// because ctx did is left running: the checker takes it again when it starts. A check whose run was ended because no
|
|
|
|
|
// candidate holds its heads any more (issue 1890) is ended: its verdict says so, and no status is written.
|
|
|
|
|
func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) {
|
|
|
|
|
if run.Err() != nil {
|
|
|
|
|
return // ended before it was claimed: nothing to write
|
|
|
|
|
}
|
|
|
|
|
ch.mu.Lock()
|
|
|
|
|
first := !ch.tried[cs.Key]
|
|
|
|
|
ch.tried[cs.Key] = true
|
|
|
|
@@ -501,21 +635,26 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
|
|
|
|
|
}
|
|
|
|
|
began := ch.now()
|
|
|
|
|
ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
|
|
|
|
|
out := ch.judge(ctx, cs)
|
|
|
|
|
out := ch.judge(run, cs)
|
|
|
|
|
// An ask the build seat gave up on judged nothing (issue 734): the check is asked again under the same claim, its status
|
|
|
|
|
// pending, and after LostAskTries it fails as not heard, judged again when the nodes change.
|
|
|
|
|
for try := 1; IsLostAsk(out.Verdict) && ctx.Err() == nil; try++ {
|
|
|
|
|
for try := 1; IsLostAsk(out.Verdict) && run.Err() == nil; try++ {
|
|
|
|
|
if try >= LostAskTries {
|
|
|
|
|
out.Verdict.Failed.Words = strings.TrimSpace(out.Verdict.Failed.Words + "\nasked " + fmt.Sprint(LostAskTries) +
|
|
|
|
|
" times and given up each time: " + UnheardWords)
|
|
|
|
|
break
|
|
|
|
|
}
|
|
|
|
|
ch.logf("%s: %s; asked again (%d of %d)", cs.Key, out.Verdict.Failed.Words, try+1, LostAskTries)
|
|
|
|
|
out = ch.judge(ctx, cs)
|
|
|
|
|
out = ch.judge(run, cs)
|
|
|
|
|
}
|
|
|
|
|
if ctx.Err() != nil {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
var why superseded
|
|
|
|
|
if errors.As(context.Cause(run), &why) {
|
|
|
|
|
ch.end(ctx, cs, cl, why)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
v := out.Verdict
|
|
|
|
|
v.Group, v.Heads, v.Trunk, v.Beside = cs.Group, cs.Heads, cs.Trunk, nil
|
|
|
|
|
for r, c := range out.Beside {
|
|
|
|
@@ -569,6 +708,26 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
|
|
|
|
|
ch.logf("%s: %s", cs.Key, v.State)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// end writes the verdict of a check whose heads no candidate holds any more (issue 1890), by compare-and-set on its claim,
|
|
|
|
|
// and says it once in the log with the asks it had put: they stay in the build seats' streams, since an ask once put cannot
|
|
|
|
|
// be taken back (buildrun.Asker has no withdraw), and are no longer waited for. No status is written: the heads that
|
|
|
|
|
// replaced these have their own check, whose pending status stands.
|
|
|
|
|
func (ch *Checker) end(ctx context.Context, cs *Case, cl *claimed, why superseded) {
|
|
|
|
|
v := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail,
|
|
|
|
|
Failed: delivery.FailedFirst{Machine: "checker", Words: why.Error()}}
|
|
|
|
|
data, err := delivery.EncodeVerdict(v)
|
|
|
|
|
if err == nil {
|
|
|
|
|
_, err = ch.Verdicts.Update(ctx, cs.Key, data, cl.rev)
|
|
|
|
|
}
|
|
|
|
|
if err != nil {
|
|
|
|
|
ch.logf("%s: %s, but the verdict could not say so: %v", cs.Key, why.Error(), err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
stage, _ := cs.Progress.Now()
|
|
|
|
|
asks := cs.Asks()
|
|
|
|
|
ch.logf("%s: %s; it was at %q with %d asks put %v, which are no longer waited for", cs.Key, why.Error(), stage, len(asks), asks)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// judge runs the case within the bound. A judge that fails to judge, and a check that ran past its bound, are verdicts of
|
|
|
|
|
// failure naming the machine and the step.
|
|
|
|
|
func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome {
|
|
|
|
@@ -667,7 +826,7 @@ func (ch *Checker) Sweep(ctx context.Context) error {
|
|
|
|
|
continue
|
|
|
|
|
}
|
|
|
|
|
cur.State = delivery.VerdictFail
|
|
|
|
|
cur.Failed = delivery.FailedFirst{Machine: "checker", Words: "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"}
|
|
|
|
|
cur.Failed = delivery.FailedFirst{Machine: "checker", Words: EndedWords + "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"}
|
|
|
|
|
data, err := delivery.EncodeVerdict(cur)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return err
|
|
|
|
|